【免费下载链接】authlibThe ultimate Python library in building OAuth, OpenID Connect clients and servers. JWS, JWE, JWK, JWA, JWT included.项目地址https://gitcode.com/gh_mirrors/au/authlib点击查看免费下载OAuth 1.0RFC 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 CredentialsResources 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_TRANSPORTtrue其原理在 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_keyTrue) def get_user_id(self): return self.idClient。客户端模型必须补齐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_keyTrue) client_id db.Column(db.String(48), indexTrue) client_secret db.Column(db.String(120), nullableFalse) default_redirect_uri db.Column(db.Text, nullableFalse, default) user_id db.Column( db.Integer, db.ForeignKey(user.id, ondeleteCASCADE) ) 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 Noneget_rsa_public_key()在客户端不使用 RSA-SHA1 签名时返回None即可若支持 RSA-SHA1则返回该客户端的 RSA 公钥。Temporary Credential临时凭证。它用于换取 token 凭证本质是“request token and secret”。因为是临时的文档建议存 cache 而非数据库——一个 cache 实例只需具备三个方法.get(key).set(key, value, expiresNone).delete(key)cache 可以是 memcache、redis 或任何实现上述接口的存储。没有 cache 时也可以落库用 SQLAlchemy 实现TemporaryCredentialMixinfrom authlib.oauth1 import TemporaryCredentialMixin class TemporaryCredential(TemporaryCredentialMixin, db.Model): id db.Column(db.Integer, primary_keyTrue) user_id db.Column( db.Integer, db.ForeignKey(user.id, ondeleteCASCADE) ) user db.relationship(User) client_id db.Column(db.String(48), indexTrue) oauth_token db.Column(db.String(84), uniqueTrue, indexTrue) 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.pyTemporaryCredentialMixin除了继承 token 相关的get_oauth_token/get_oauth_token_secret外还要求实现get_client_id()、get_redirect_uri()返回oauth_callback与check_verifier()。Token Credentialtoken 凭证。它用于访问资源所有者的资源不设过期时间应持久化到数据库而非 cache。实现TokenCredentialMixin缺失方法from authlib.oauth1 import TokenCredentialMixin class TokenCredential(TokenCredentialMixin, db.Model): id db.Column(db.Integer, primary_keyTrue) user_id db.Column( db.Integer, db.ForeignKey(user.id, ondeleteCASCADE) ) user db.relationship(User) client_id db.Column(db.String(48), indexTrue) oauth_token db.Column(db.String(84), uniqueTrue, indexTrue) 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, nameunique_nonce ), ) id db.Column(db.Integer, primary_keyTrue) client_id db.Column(db.String(48), nullableFalse) timestamp db.Column(db.Integer, nullableFalse) nonce db.Column(db.String(48), nullableFalse) 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。第三步定义 AuthorizationServerAuthlib 提供开箱即用的authlib.integrations.flask_oauth1.AuthorizationServer自带请求/响应处理能力对应实现见 authlib/integrations/flask_oauth1/authorization_server.pyfrom authlib.integrations.flask_oauth1 import AuthorizationServer def query_client(client_id): return Client.query.filter_by(client_idclient_id).first() server AuthorizationServer(app, query_clientquery_client)也可以延迟初始化server AuthorizationServer() server.init_app(app, query_clientquery_client)文档强烈建议使用 cache——这样很多 hook 方法都不必自己重写。init_app时还会读取app.config中的签名方法配置见源码第 56-58 行。第四步Token 生成配置AuthorizationServer默认即可工作但有几个可配置项配置项说明OAUTH1_TOKEN_GENERATOR一个模块路径字符串用于导入生成oauth_token的函数OAUTH1_TOKEN_SECRET_GENERATOR模块路径字符串用于导入生成oauth_token_secret的函数OAUTH1_TOKEN_LENGTH未配置OAUTH1_TOKEN_GENERATOR时随机函数生成的oauth_token长度默认 42OAUTH1_TOKEN_SECRET_LENGTH随机函数生成的oauth_token_secret长度默认 48在源码 create_token_generator 中可以看到若配置为字符串则通过werkzeug.utils.import_string动态导入否则用authlib.common.security.generate_token按默认长度生成。这些配置最终汇总成一个返回{oauth_token: ..., oauth_token_secret: ...}的token_generator函数。你也可以在初始化时直接传入自定义token_generatordef token_generator(): return { oauth_token: random_string(20), oauth_token_secret: random_string(46) } server AuthorizationServer( app, query_clientquery_client, token_generatortoken_generator )第五步注册 Server HooksAuthorizationServer有一组必须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( noncenonce, timestamptimestamp, client_idclient_id, ) if oauth_token: q q.filter_by(oauth_tokenoauth_token) rv q.first() if rv: return True item TimestampNonce( noncenonce, timestamptimestamp, client_idclient_id, oauth_tokenoauth_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_idclient_id, oauth_tokentoken[oauth_token], oauth_token_secrettoken[oauth_token_secret], oauth_callbackredirect_uri, ) db.session.add(item) db.session.commit() return item def get_temporary_credential(oauth_token): return TemporaryCredential.query.filter_by(oauth_tokenoauth_token).first() def delete_temporary_credential(oauth_token): q TemporaryCredential.query.filter_by(oauth_tokenoauth_token) q.delete(synchronize_sessionFalse) 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_credentialhookdef create_token_credential(token, temporary_credential): credential TokenCredential( oauth_tokentoken[oauth_token], oauth_token_secrettoken[oauth_token_secret], client_idtemporary_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, reqreq) except OAuth1Error as error: return render_template(error.html, errorerror) granted request.form.get(granted) if granted: grant_user current_user else: grant_user None try: return server.create_authorization_response(grant_usergrant_user) except OAuth1Error as error: return render_template(error.html, errorerror)这里的check_authorization_request()会校验oauth_token并取出临时凭证grant_userNone时用户拒绝授权服务器按 核心实现 返回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_clientquery_client, query_tokenquery_token, exists_nonceexists_nonce, ) # or initialize it lazily require_oauth ResourceProtector() require_oauth.init_app( app, query_clientquery_client, query_tokenquery_token, exists_nonceexists_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 需要的三个函数query_clientquery_tokenexists_nonceSQLAlchemy 场景下的实现示例def query_client(client_id): # assuming Client is the model return Client.query.filter_by(client_idclient_id).first() def query_token(client_id, oauth_token): return TokenCredential.query.filter_by(client_idclient_id, oauth_tokenoauth_token).first()exists_nonce如果复用授权服务器的 cacheAuthlib 提供了内置工具函数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}为 keycache.has判定是否重复再set(key, 1, timeout86400)记录一次默认有效期一天。用数据库时则与授权服务器端一致def exists_nonce(nonce, timestamp, client_id, oauth_token): q db.session.query(TimestampNonce.nonce).filter_by( noncenonce, timestamptimestamp, client_idclient_id, ) if oauth_token: q q.filter_by(oauth_tokenoauth_token) rv q.first() if rv: return True tn TimestampNonce( noncenonce, timestamptimestamp, client_idclient_id, oauth_tokenoauth_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-SHA256AuthorizationServer与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_METHODSOAUTH1_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.pyfrom your_project.models import Client, Token from authlib.integrations.django_oauth1 import CacheAuthorizationServer authorization_server CacheAuthorizationServer(Client, Token)Django 中的模型定义Resource Owner可直接用 Django 内置的 contrib userfrom django.contrib.auth.models import UserClient。Authlib 没有为 Django 提供 Client 模型需要自己实现ClientMixinfrom 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_deleteCASCADE) client_id models.CharField(max_length48, uniqueTrue, db_indexTrue) client_secret models.CharField(max_length48, blankTrue) default_redirect_uri models.TextField(blankFalse, 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 NoneToken。token 凭证持久化到数据库实现TokenCredentialMixinfrom 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_deleteCASCADE) client_id models.CharField(max_length48, db_indexTrue) oauth_token models.CharField(max_length84, uniqueTrue, db_indexTrue) oauth_token_secret models.CharField(max_length84) def get_oauth_token(self): return self.oauth_token def get_oauth_token_secret(self): return self.oauth_token_secretDjango 中的三个视图端点临时凭证端点POSTfrom 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 端点POSTfrom 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字典读取以下配置配置键说明默认值signature_methods支持的签名方法列表空仅 HMAC-SHA1nonce_expires_innonce 在 cache 中的存活秒数86400temporary_credential_expires_in临时凭证在 cache 中的存活秒数86400temporary_credential_key_prefix临时凭证 cache key 前缀temporary_credential:临时凭证通过cache.set(key_prefix oauth_token, token, timeout...)写入 Django 缓存nonce 校验逻辑在 authlib/integrations/django_oauth1/nonce.pykey 格式为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(usernameuser.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_TRANSPORTtrue并配置OAUTH1_SUPPORTED_SIGNATURE_METHODS [PLAINTEXT, HMAC-SHA1, RSA-SHA1]tests/flask/test_oauth1/test_authorize.py 以use_cacheTrue/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 走 cachetoken 凭证落库AuthorizationServer 的六个 hooks 必须注册cache 场景可用辅助函数一键完成资源服务器需要 query_client / query_token / exists_nonce 三个函数本地开发务必设置AUTHLIB_INSECURE_TRANSPORTtrue。如果你要查阅完整 API 签名可参考 Flask 版 api.rstAuthorizationServer、ResourceProtector、current_credential与 Django 版 api.rstCacheAuthorizationServer、ResourceProtector想了解客户端如何接入可继续阅读 docs/oauth1/client/index.rstOAuth 1.0 协议本身的握手细节与签名规范则记录在 docs/oauth1/specs/rfc5849.rst。需要提醒的是从 v1.0.0 起 Authlib 移除了内置的 SQLAlchemy 集成文档中以versionchanged标注所有 ORM 模型都需按本文方式自行实现 mixin这也让 Provider 的存储层完全掌握在开发者自己手中。赞分享【免费下载链接】authlibThe ultimate Python library in building OAuth, OpenID Connect clients and servers. JWS, JWE, JWK, JWA, JWT included.项目地址https://gitcode.com/gh_mirrors/au/authlib点击查看免费下载相关推荐基于 Authlib 在 Django 中构建 OAuth 1.0 授权服务器与资源服务器基于 Authlib 在 Django 中构建 OAuth 1.0 授权服务器与资源服务器 本指南完整讲解如何使用 Authlib 的 authlib.inte使用 Authlib 在 Flask 中构建 OAuth 1.0 Provider授权服务器与资源服务器完整实战指南使用 Authlib 在 Flask 中构建 OAuth 1.0 Provider授权服务器与资源服务器完整实战指南 Authlib 提供了开箱即用的 FlaAuthlib 实战在 Django 中构建 OAuth 2.0 授权服务器与 OpenID Connect ProviderAuthlib 实战在 Django 中构建 OAuth 2.0 授权服务器与 OpenID Connect Provider 本指南以 Authlib 官方上一篇3大华硕笔记本痛点一个轻量级开源工具全搞定下一篇Area51代码重构估算工具功能点与LOC方法创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考