OAuth 1.0（RFC 5849）虽然诞生已久，但在大量遗留系统与桌面/命令行客户端集成场景中仍是不可回避的技术选型。Authlib 在authlib.oauth1核心模块之上，为 Flask 与 Django 提供了开箱即用的 OAuth 1.0 Provider 集成（对应文档为 docs/oauth1/provider/index.rst），一个 Provider 由授权服务器（颁发 token）与资源服务器（保护用户资源）两部分组成。读完本文，你将掌握：如何在 Flask 或 Django 中定义 Client、Temporary Credential、Token 三类模型，如何通过 cache 或数据库两种方式注册非ce 与临时凭证 hooks，如何暴露/initiate、/authorize、/token三个标准端点，以及如何用ResourceProtector保护受控资源并自定义 HMAC-SHA256 等签名方法。

OAuth 1.0 Provider 的总体架构

按 Flask Provider 文档 与 Django Provider 文档 的定义，一个 OAuth 1.0 Provider 包含两个逻辑服务器：

• Authorization Server（授权服务器）：负责颁发 access token。它提供三类端点——发放临时凭证（Temporary Credentials）、资源所有者授权（Authorization）、以及用临时凭证换取正式 token 凭证（Token Credentials）；
• Resources Server（资源服务器）：保护用户资源，只允许持有授权 token 的客户端访问。

OAuth 1.0 的凭证生命周期分为两跳：客户端先从授权服务器拿到临时凭证（request token / request secret），引导资源所有者在授权页面确认后拿到oauth_verifier，最后再用“临时凭证 + verifier”换得token 凭证（access token / token secret）。与 OAuth 2 不同，OAuth 1 的 token 凭证不会过期，因此文档建议将 token 持久化到数据库，而临时凭证、timestamp/nonce 这类短期数据放在 cache 中更合适。

实现一个 Provider 前，先理解两个内置角色：

• Resource Owner（资源所有者）：即使用你服务的用户，可以用用户名/邮箱+密码或其他方式登录；
• Client（客户端）：代替资源所有者发起受保护资源请求的第三方应用，至少包含三要素：client_id、client_secret，以及（若支持 RSA-SHA1 签名方法时）客户端 RSA 公钥。

本地开发必知：AUTHLIB_INSECURE_TRANSPORT

文档中特别强调：如果你在 localhost 上开发，必须设置环境变量，否则 OAuth 1.0 的签名校验与安全传输检查会拦截本地 HTTP 请求：

export AUTHLIB_INSECURE_TRANSPORT=true

其原理在 authlib/common/security.py 的is_secure_transport()中：默认只信任https请求，但一旦检测到环境变量AUTHLIB_INSECURE_TRANSPORT便直接放行；此外http://localhost与回环地址（loopback，含 IPv4-mapped IPv6）也会被识别为安全传输。测试代码 tests/flask/test_oauth1/conftest.py 正是通过 autouse fixture 设置该环境变量来跑通本地测试的。

Flask 实现：从数据模型到三个端点

第一步：定义四类模型

Resource Owner。资源所有者模型必须实现get_user_id()方法：

class User(db.Model): id = db.Column(db.Integer, primary_key=True) def get_user_id(self): return self.id

Client。客户端模型必须补齐authlib.oauth1.ClientMixin的缺失方法。以 Flask-SQLAlchemy 为例（参见 authlib/oauth1/rfc5849/models.py 中ClientMixin的三个抽象方法）：

from authlib.oauth1 import ClientMixin class Client(ClientMixin, db.Model): id = db.Column(db.Integer, primary_key=True) client_id = db.Column(db.String(48), index=True) client_secret = db.Column(db.String(120), nullable=False) default_redirect_uri = db.Column(db.Text, nullable=False, default='') user_id = db.Column( db.Integer, db.ForeignKey('user.id', ondelete='CASCADE') ) user = db.relationship('User') def get_default_redirect_uri(self): return self.default_redirect_uri def get_client_secret(self): return self.client_secret def get_rsa_public_key(self): return None

get_rsa_public_key()在客户端不使用 RSA-SHA1 签名时返回None即可；若支持 RSA-SHA1，则返回该客户端的 RSA 公钥。

Temporary Credential（临时凭证）。它用于换取 token 凭证，本质是“request token and secret”。因为是临时的，文档建议存 cache 而非数据库——一个 cache 实例只需具备三个方法：

• .get(key)
• .set(key, value, expires=None)
• .delete(key)

cache 可以是 memcache、redis 或任何实现上述接口的存储。没有 cache 时也可以落库，用 SQLAlchemy 实现TemporaryCredentialMixin：

from authlib.oauth1 import TemporaryCredentialMixin class TemporaryCredential(TemporaryCredentialMixin, db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column( db.Integer, db.ForeignKey('user.id', ondelete='CASCADE') ) user = db.relationship('User') client_id = db.Column(db.String(48), index=True) oauth_token = db.Column(db.String(84), unique=True, index=True) oauth_token_secret = db.Column(db.String(84)) oauth_verifier = db.Column(db.String(84)) oauth_callback = db.Column(db.Text, default='') def get_client_id(self): return self.client_id def get_redirect_uri(self): return self.oauth_callback def check_verifier(self, verifier): return self.oauth_verifier == verifier def get_oauth_token(self): return self.oauth_token def get_oauth_token_secret(self): return self.oauth_token_secret

对照 models.py，TemporaryCredentialMixin除了继承 token 相关的get_oauth_token/get_oauth_token_secret外，还要求实现get_client_id()、get_redirect_uri()（返回oauth_callback）与check_verifier()。

Token Credential（token 凭证）。它用于访问资源所有者的资源，不设过期时间，应持久化到数据库而非 cache。实现TokenCredentialMixin缺失方法：

from authlib.oauth1 import TokenCredentialMixin class TokenCredential(TokenCredentialMixin, db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column( db.Integer, db.ForeignKey('user.id', ondelete='CASCADE') ) user = db.relationship('User') client_id = db.Column(db.String(48), index=True) oauth_token = db.Column(db.String(84), unique=True, index=True) oauth_token_secret = db.Column(db.String(84)) def get_oauth_token(self): return self.oauth_token def get_oauth_token_secret(self): return self.oauth_token_secret

第二步：Timestamp 与 Nonce 校验

oauth_nonce必须在同一 timestamp、同一客户端凭证与 token 组合下唯一。Flask 集成内置了基于 cache 的校验。若使用数据库，可以建立带唯一约束的表：

class TimestampNonce(db.Model): __table_args__ = ( db.UniqueConstraint( 'client_id', 'timestamp', 'nonce', 'oauth_token', name='unique_nonce' ), ) id = db.Column(db.Integer, primary_key=True) client_id = db.Column(db.String(48), nullable=False) timestamp = db.Column(db.Integer, nullable=False) nonce = db.Column(db.String(48), nullable=False) oauth_token = db.Column(db.String(84))

值得说明的是，Authlib 核心层（authlib/oauth1/rfc5849/base_server.py）在调用exists_noncehook 之前，还会先校验oauth_timestamp：它必须是正整数，且默认过期窗口EXPIRY_TIME = 300秒——即time.time() - timestamp > 300的请求会被判定为无效 timestamp。只有使用 PLAINTEXT 签名方法时才允许省略 timestamp/nonce。

第三步：定义 AuthorizationServer

Authlib 提供开箱即用的authlib.integrations.flask_oauth1.AuthorizationServer，自带请求/响应处理能力（对应实现见 authlib/integrations/flask_oauth1/authorization_server.py）：

from authlib.integrations.flask_oauth1 import AuthorizationServer def query_client(client_id): return Client.query.filter_by(client_id=client_id).first() server = AuthorizationServer(app, query_client=query_client)

也可以延迟初始化：

server = AuthorizationServer() server.init_app(app, query_client=query_client)

文档强烈建议使用 cache——这样很多 hook 方法都不必自己重写。init_app时还会读取app.config中的签名方法配置（见源码第 56-58 行）。

第四步：Token 生成配置

AuthorizationServer默认即可工作，但有几个可配置项：

在源码 create_token_generator 中可以看到：若配置为字符串，则通过werkzeug.utils.import_string动态导入；否则用authlib.common.security.generate_token按默认长度生成。这些配置最终汇总成一个返回{'oauth_token': ..., 'oauth_token_secret': ...}的token_generator函数。

你也可以在初始化时直接传入自定义token_generator：

def token_generator(): return { 'oauth_token': random_string(20), 'oauth_token_secret': random_string(46) } server = AuthorizationServer( app, query_client=query_client, token_generator=token_generator )

第五步：注册 Server Hooks

AuthorizationServer有一组必须register_hook的钩子（内部_hooks字典定义于源码第 36-43 行，共 6 个）：exists_nonce、create_temporary_credential、get_temporary_credential、delete_temporary_credential、create_authorization_verifier、create_token_credential。如果注册了未定义名字的 hook，会抛出ValueError。

有 cache 时的快捷注册。使用 authlib/integrations/flask_oauth1/cache.py 提供的辅助函数即可：

from authlib.integrations.flask_oauth1.cache import ( register_nonce_hooks, register_temporary_credential_hooks ) register_nonce_hooks(server, cache) register_temporary_credential_hooks(server, cache)

没有 cache 时的数据库实现。你必须逐个注册上述 hooks：

# check if nonce exists def exists_nonce(nonce, timestamp, client_id, oauth_token): q = TimestampNonce.query.filter_by( nonce=nonce, timestamp=timestamp, client_id=client_id, ) if oauth_token: q = q.filter_by(oauth_token=oauth_token) rv = q.first() if rv: return True item = TimestampNonce( nonce=nonce, timestamp=timestamp, client_id=client_id, oauth_token=oauth_token, ) db.session.add(item) db.session.commit() return False server.register_hook('exists_nonce', exists_nonce) # hooks for temporary credential def create_temporary_credential(token, client_id, redirect_uri): item = TemporaryCredential( client_id=client_id, oauth_token=token['oauth_token'], oauth_token_secret=token['oauth_token_secret'], oauth_callback=redirect_uri, ) db.session.add(item) db.session.commit() return item def get_temporary_credential(oauth_token): return TemporaryCredential.query.filter_by(oauth_token=oauth_token).first() def delete_temporary_credential(oauth_token): q = TemporaryCredential.query.filter_by(oauth_token=oauth_token) q.delete(synchronize_session=False) db.session.commit() def create_authorization_verifier(credential, grant_user, verifier): credential.user_id = grant_user.id # assuming your end user model has `.id` credential.oauth_verifier = verifier db.session.add(credential) db.session.commit() return credential server.register_hook('create_temporary_credential', create_temporary_credential) server.register_hook('get_temporary_credential', get_temporary_credential) server.register_hook('delete_temporary_credential', delete_temporary_credential) server.register_hook('create_authorization_verifier', create_authorization_verifier)

无论临时凭证用 cache 还是数据库，都必须注册create_token_credentialhook：

def create_token_credential(token, temporary_credential): credential = TokenCredential( oauth_token=token['oauth_token'], oauth_token_secret=token['oauth_token_secret'], client_id=temporary_credential.get_client_id() ) credential.user_id = temporary_credential.user_id db.session.add(credential) db.session.commit() return credential server.register_hook('create_token_credential', create_token_credential)

注意 hooks 的函数签名与核心层回调严格对应：exists_nonce(nonce, timestamp, client_id, oauth_token)对应 authorization_server.py 中把 request 拆解后的调用；create_authorization_verifier(credential, grant_user, verifier)中的verifier由框架用generate_token(36)生成（源码第 129-136 行）。

第六步：实现三个端点

临时凭证端点，供客户端以 POST 方式获取临时凭证：

@app.route('/initiate', methods=['POST']) def initiate_temporary_credential(): return server.create_temporary_credentials_response()

资源所有者授权端点。OAuth 1 客户端会把用户重定向到该授权页，让资源所有者批准或拒绝请求：

@app.route('/authorize', methods=['GET', 'POST']) def authorize(): # make sure that user is logged in for yourself if request.method == 'GET': try: req = server.check_authorization_request() return render_template('authorize.html', req=req) except OAuth1Error as error: return render_template('error.html', error=error) granted = request.form.get('granted') if granted: grant_user = current_user else: grant_user = None try: return server.create_authorization_response(grant_user=grant_user) except OAuth1Error as error: return render_template('error.html', error=error)

这里的check_authorization_request()会校验oauth_token并取出临时凭证；grant_user=None时用户拒绝授权，服务器按 核心实现 返回AccessDeniedError并重定向；批准时生成 verifier 并带着oauth_token、oauth_verifier两个参数 302 回oauth_callback。

最终 token 端点。客户端用临时凭证 + 用户授权的oauth_verifier换取 token 凭证：

@app.route('/token', methods=['POST']) def issue_token(): return server.create_token_response()

从核心层的 create_token_response 可以看到完整流程：校验oauth_consumer_key、oauth_token、oauth_verifier（含token.check_verifier(verifier)），校验 timestamp/nonce 与签名，创建 token 凭证后立即删除临时凭证，并以application/x-www-form-urlencoded格式返回oauth_token与oauth_token_secret。任何校验失败都会先删除临时凭证再返回错误，防止凭证被重放。

Flask 资源服务器：用 ResourceProtector 保护接口

资源服务器可以与授权服务器是不同进程。保护用户资源的典型写法：

from flask import jsonify from authlib.integrations.flask_oauth1 import ResourceProtector, current_credential # we will define ``query_client``, ``query_token``, and ``exists_nonce`` later. require_oauth = ResourceProtector( app, query_client=query_client, query_token=query_token, exists_nonce=exists_nonce, ) # or initialize it lazily require_oauth = ResourceProtector() require_oauth.init_app( app, query_client=query_client, query_token=query_token, exists_nonce=exists_nonce, ) @app.route('/user') @require_oauth def user_profile(): user = current_credential.user return jsonify(user)

current_credential是对前面定义的 Token 模型的代理（实现见 authlib/integrations/flask_oauth1/resource_protector.py，内部通过 Flaskg存储解析出的凭证）。因为 Token 模型上定义了user关系，所以可以用current_credential.user访问用户。

初始化 ResourceProtector 需要的三个函数

1. query_client
2. query_token
3. exists_nonce

SQLAlchemy 场景下的实现示例：

def query_client(client_id): # assuming ``Client`` is the model return Client.query.filter_by(client_id=client_id).first() def query_token(client_id, oauth_token): return TokenCredential.query.filter_by(client_id=client_id, oauth_token=oauth_token).first()

exists_nonce如果复用授权服务器的 cache，Authlib 提供了内置工具函数：

from authlib.integrations.flask_oauth1 import create_exists_nonce_func exists_nonce = create_exists_nonce_func(cache)

该函数（cache.py）以nonce:{nonce}-{timestamp}-{client_id}（有 token 时再拼接-{token}）为 key，cache.has判定是否重复，再set(key, 1, timeout=86400)记录一次，默认有效期一天。用数据库时则与授权服务器端一致：

def exists_nonce(nonce, timestamp, client_id, oauth_token): q = db.session.query(TimestampNonce.nonce).filter_by( nonce=nonce, timestamp=timestamp, client_id=client_id, ) if oauth_token: q = q.filter_by(oauth_token=oauth_token) rv = q.first() if rv: return True tn = TimestampNonce( nonce=nonce, timestamp=timestamp, client_id=client_id, oauth_token=oauth_token, ) db.session.add(tn) db.session.commit() return False

在 MethodView 与 Flask-Restful 中使用

require_oauth还可以直接放进flask.views.MethodView与flask_restful.Resource的装饰器列表：

from flask.views import MethodView class UserAPI(MethodView): decorators = [require_oauth] from flask_restful import Resource class UserAPI(Resource): method_decorators = [require_oauth]

自定义签名方法：扩展 HMAC-SHA256

AuthorizationServer与ResourceProtector默认只支持HMAC-SHA1。框架内置三种签名方法，可用配置启用：

OAUTH1_SUPPORTED_SIGNATURE_METHODS = ['HMAC-SHA1', 'PLAINTEXT', 'RSA-SHA1']

BaseServer.SIGNATURE_METHODS注册表与SUPPORTED_SIGNATURE_METHODS白名单定义在 authlib/oauth1/rfc5849/base_server.py，签名算法实现在 authlib/oauth1/rfc5849/signature.py。以扩展一个HMAC-SHA256方法为例：

import hmac from authlib.common.encoding import to_bytes from authlib.oauth1.rfc5849 import signature def verify_hmac_sha256(request): text = signature.generate_signature_base_string(request) key = escape(request.client_secret or '') key += '&' key += escape(request.token_secret or '') sig = hmac.new(to_bytes(key), to_bytes(text), hashlib.sha256) return binascii.b2a_base64(sig.digest())[:-1] AuthorizationServer.register_signature_method( 'HMAC-SHA256', verify_hmac_sha256 ) ResourceProtector.register_signature_method( 'HMAC-SHA256', verify_hmac_sha256 )

然后把它加入SUPPORTED_SIGNATURE_METHODS：

OAUTH1_SUPPORTED_SIGNATURE_METHODS = ['HMAC-SHA256']

此时服务器将只支持 HMAC-SHA256；想支持更多方法就把它们都加进列表。从 base_server.py 可以看到，register_signature_method是类方法，verify 函数接收OAuth1Request并返回布尔值，校验时先看方法是否在白名单（否则抛UnsupportedSignatureMethodError），再查SIGNATURE_METHODS注册表执行验证。验证链路本身包含generate_signature_base_string（按 RFC 5849 Section 3.4.1 拼接方法、规范化 URI 与排序后的参数）以及hmac.compare_digest常量时间比较（signature.py）。

Django 实现：CacheAuthorizationServer 与 ResourceProtector

初始化授权服务器

Django 集成的最大特点是重度使用 cache：临时凭证、timestamp 与 nonce 全部内置，无需自己处理。创建授权服务器只需要Client和Token两个模型（见 authlib/integrations/django_oauth1/authorization_server.py）：

from your_project.models import Client, Token from authlib.integrations.django_oauth1 import CacheAuthorizationServer authorization_server = CacheAuthorizationServer(Client, Token)

Django 中的模型定义

Resource Owner可直接用 Django 内置的 contrib user：

from django.contrib.auth.models import User

Client。Authlib 没有为 Django 提供 Client 模型，需要自己实现ClientMixin：

from django.db import models from django.contrib.auth.models import User from authlib.oauth1 import ClientMixin class Client(models.Model, ClientMixin): user = models.ForeignKey(User, on_delete=CASCADE) client_id = models.CharField(max_length=48, unique=True, db_index=True) client_secret = models.CharField(max_length=48, blank=True) default_redirect_uri = models.TextField(blank=False, default='') def get_default_redirect_uri(self): return self.default_redirect_uri def get_client_secret(self): return self.client_secret def get_rsa_public_key(self): return None

Token。token 凭证持久化到数据库，实现TokenCredentialMixin：

from django.db import models from django.contrib.auth.models import User from authlib.oauth1 import TokenCredentialMixin class Token(models.Model, TokenCredentialMixin): user = models.ForeignKey(User, on_delete=CASCADE) client_id = models.CharField(max_length=48, db_index=True) oauth_token = models.CharField(max_length=84, unique=True, db_index=True) oauth_token_secret = models.CharField(max_length=84) def get_oauth_token(self): return self.oauth_token def get_oauth_token_secret(self): return self.oauth_token_secret

Django 中的三个视图端点

临时凭证端点（POST）：

from django.views.decorators.http import require_http_methods @require_http_methods(["POST"]) def initiate_temporary_credential(request): return server.create_temporary_credential_response(request)

资源所有者授权端点：

from django.shortcuts import render def authorize(request): # make sure that user is logged in for yourself if request.method == 'GET': try: req = server.check_authorization_request(request) context = {'req': req} return render(request, 'authorize.html', context) except OAuth1Error as error: context = {'error': error} return render(request, 'error.html', context) granted = request.POST.get('granted') if granted: grant_user = request.user else: grant_user = None try: return server.create_authorization_response(request, grant_user) except OAuth1Error as error: context = {'error': error} return render(request, 'error.html', context)

token 端点（POST）：

from django.views.decorators.http import require_http_methods @require_http_methods(["POST"]) def issue_token(request): return server.create_token_response(request)

最后把这三个视图注册到 URL patterns 即可。

Django 的 AUTHLIB_OAUTH1_PROVIDER 配置

从 authorization_server.py 与 resource_protector.py 的源码可以看到，Django 集成从settings.AUTHLIB_OAUTH1_PROVIDER字典读取以下配置：

临时凭证通过cache.set(key_prefix + oauth_token, token, timeout=...)写入 Django 缓存，nonce 校验逻辑在 authlib/integrations/django_oauth1/nonce.py，key 格式为nonce:{nonce}-{timestamp}-{client_id}（可再拼 token），与 Flask 版本保持一致。

Django 资源服务器

from django.http import JsonResponse from authlib.integrations.django_oauth1 import ResourceProtector require_oauth = ResourceProtector(Client, TokenCredential) @require_oauth def user_api(request): user = request.oauth1_credential.user return JsonResponse(dict(username=user.username))

require_oauth装饰器会把解析出的凭证以oauth1_credential属性附加到request上（源码见 resource_protector.py），该对象就是 Token 模型实例，因此可以直接访问request.oauth1_credential.user。校验失败时返回带Cache-Control: no-store、Pragma: no-cache的 JSON 错误响应。

测试佐证：看看框架如何被验证

仓库的测试是理解整套流程的最佳范例：

• tests/flask/test_oauth1/conftest.py 通过 autouse fixture 设置AUTHLIB_INSECURE_TRANSPORT=true，并配置OAUTH1_SUPPORTED_SIGNATURE_METHODS = ['PLAINTEXT', 'HMAC-SHA1', 'RSA-SHA1']；
• tests/flask/test_oauth1/test_authorize.py 以use_cache=True/False参数化跑通“缺失oauth_token返回missing_required_parameter”“不存在的 token 返回invalid_token”等授权校验用例；
• 同目录下的test_temporary_credentials.py、test_token_credentials.py、test_resource_protector.py则分别覆盖临时凭证、token 凭证签发与资源保护的完整链路，可作为你实现自定义 Provider 时的行为参照。

小结与延伸阅读

本文从模型、hook、端点、签名方法到资源保护，完整走通了 Authlib 在 Flask 与 Django 上的 OAuth 1.0 Provider 落地路径。核心要点可归纳为：临时凭证与 nonce 走 cache，token 凭证落库；AuthorizationServer 的六个 hooks 必须注册（cache 场景可用辅助函数一键完成）；资源服务器需要 query_client / query_token / exists_nonce 三个函数；本地开发务必设置AUTHLIB_INSECURE_TRANSPORT=true。

如果你要查阅完整 API 签名，可参考 Flask 版 api.rst（AuthorizationServer、ResourceProtector、current_credential）与 Django 版 api.rst（CacheAuthorizationServer、ResourceProtector）；想了解客户端如何接入，可继续阅读 docs/oauth1/client/index.rst；OAuth 1.0 协议本身的握手细节与签名规范则记录在 docs/oauth1/specs/rfc5849.rst。需要提醒的是，从 v1.0.0 起 Authlib 移除了内置的 SQLAlchemy 集成（文档中以versionchanged标注），所有 ORM 模型都需按本文方式自行实现 mixin，这也让 Provider 的存储层完全掌握在开发者自己手中。