【免费下载链接】authlib
The ultimate Python library in building OAuth, OpenID Connect clients and servers. JWS, JWE, JWK, JWA, JWT included.
OAuth 1.0(RFC 5849)至今仍是部分老牌服务商(如 Twitter 早期 API)采用的授权协议。Authlib 在 authlib/integrations/requests_client 与 authlib/integrations/httpx_client 两个模块中,分别提供了基于requests和httpx的 OAuth 1.0 客户端实现。本文依据 docs/oauth1/client/http/api.rst 所列的 API 清单,并结合 docs/oauth1/client/http/index.rst 的通用流程指南,系统讲解OAuth1Session、OAuth1Client、AsyncOAuth1Client与OAuth1Auth的完整使用方法:从初始化客户端、获取临时凭证(Request Token)、跳转授权页、换取访问令牌(Access Token)到签名访问受保护资源,并深入源码说明签名算法与签名位置的底层原理。
Authlib 提供的 OAuth 1.0 客户端类一览
按照 api.rst 的 API 参考划分,Authlib 在两类 HTTP 库上提供了对等的实现,二者共享同一套 API 设计:
| HTTP 库 | 会话类 | 认证对象 |
|---|---|---|
| requests | OAuth1Session | OAuth1Auth |
| httpx(同步) | OAuth1Client | OAuth1Auth |
| httpx(异步) | AsyncOAuth1Client | OAuth1Auth |
OAuth1Session是requests-oauthlib的直接替代品,继承自requests.Session,见 oauth1_session.py。AsyncOAuth1Client是异步 OAuth 1.0 客户端,继承自httpx.AsyncClient,见 oauth1_client.py。- 同步的
OAuth1Client则继承自httpx.Client。 - 会话类需要调用的四个核心方法完全一致:
create_authorization_url、fetch_request_token、fetch_access_token、parse_authorization_response。
这些方法的具体实现集中在基类 authlib/oauth1/client.py 中,两个 HTTP 集成只是分别绑定到各自的 Session/Auth 机制上,因此掌握一份流程即可在两个生态中通用。
第一步:初始化 OAuth 1.0 客户端
OAuth 1.0 获取访问令牌总共需要三步:
- 获取一个临时凭证(Temporary Credential,Twitter 中称为 Request Token);
- 访问授权页面,用户完成授权;
- 用临时凭证换取访问令牌(Access Token)。
但在这之前,需要先基于client_id与client_secret初始化客户端:
>>> client_id = 'Your Twitter client key' >>> client_secret = 'Your Twitter client secret' >>> # using requests client >>> from authlib.integrations.requests_client import OAuth1Session >>> client = OAuth1Session(client_id, client_secret) >>> # using httpx client >>> from authlib.integrations.httpx_client import AsyncOAuth1Client >>> client = AsyncOAuth1Client(client_id, client_secret)同步的 httpx 用法与此相同,只需改用OAuth1Client。从源码看,OAuth1Session与OAuth1Client的构造函数签名完全一致(见 oauth1_session.py 与 oauth1_client.py):
OAuth1Session( client_id, client_secret=None, token=None, token_secret=None, redirect_uri=None, rsa_key=None, verifier=None, signature_method=SIGNATURE_HMAC_SHA1, signature_type=SIGNATURE_TYPE_HEADER, force_include_body=False, **kwargs, )各参数含义如下:
client_id/client_secret:应用在服务商处注册获得的消费者密钥(Consumer Key / Consumer Secret),client_id缺失时会直接抛出ValueError('Missing "client_id"');token/token_secret:已持有的 OAuth Token 及其 Secret,用于会话重建或恢复已保存令牌(见下文"会话重建"一节);redirect_uri:授权完成后的回调地址,与注册时不一致时可在客户端动态覆盖;rsa_key:使用 RSA-SHA1 签名时所需的私钥(PEM 格式字符串);verifier:授权回调中拿到的验证码,可在构造时预填;signature_method:签名算法,默认HMAC-SHA1,可选RSA-SHA1、PLAINTEXT;signature_type:OAuth 参数携带位置,默认HEADER,可选QUERY(放入 URL 查询串)或BODY(放入表单体);force_include_body:强制对非application/x-www-form-urlencoded的请求体也参与签名;**kwargs:透传给底层 HTTP 库的参数,如 requests 的proxies、verify、cert(见 utils.py),或 httpx 的timeout、follow_redirects、transport(见 utils.py)。
三个签名方法常量与三个签名类型常量同样可以从两个集成模块中直接导入(见 requests_client/init.py 与 httpx_client/init.py):
from authlib.integrations.requests_client import ( SIGNATURE_HMAC_SHA1, SIGNATURE_RSA_SHA1, SIGNATURE_PLAINTEXT, SIGNATURE_TYPE_HEADER, SIGNATURE_TYPE_QUERY, SIGNATURE_TYPE_BODY, )第二步:获取临时凭证(Request Token)
调用fetch_request_token向服务商的 Request Token 端点发起(已签名的)POST 请求:
>>> request_token_url = 'https://api.twitter.com/oauth/request_token' >>> request_token = client.fetch_request_token(request_token_url) >>> print(request_token) {'oauth_token': 'gA..H', 'oauth_token_secret': 'lp..X', 'oauth_callback_confirmed': 'true'}返回的是一个字典,其中oauth_token与oauth_token_secret是后续流程必需的临时凭证,应当保存下来备用。源码层面(client.py)的_fetch_token会先做一次签名 POST,再根据响应内容解析令牌:状态码 ≥ 400 时抛出OAuthError(错误类型fetch_token_denied),响应体以{开头则按 JSON 解析,否则按application/x-www-form-urlencoded解析;两者都失败时抛出ValueError。这一点在 test_oauth1_session.py 的test_fetch_token_invalid_response中有完整覆盖(分别断言了 400/401/403 与非法响应体的报错行为)。
若希望回调到与注册时不同的地址,可以在取 Request Token 之前临时修改redirect_uri:
>>> client.redirect_uri = 'https://your-domain.org/auth' >>> client.fetch_request_token(request_token_url)redirect_uri是会话类上可读写的属性,最终会传递到底层ClientAuth对象上(见 client.py 的redirect_uriproperty 与 client_auth.py 中oauth_callback参数的生成逻辑)。
第三步:生成并访问授权 URL
拿到 Request Token 后,用create_authorization_url拼接授权页地址:
>>> authenticate_url = 'https://api.twitter.com/oauth/authenticate' >>> client.create_authorization_url(authenticate_url, request_token['oauth_token']) 'https://api.twitter.com/oauth/authenticate?oauth_token=gA..H'因为会话被复用(Request Token 已保存在会话内部),第二个参数request_token可以省略:
>>> client.create_authorization_url(authenticate_url)源码实现(client.py)会将request_token or self.auth.token作为oauth_token参数拼入 URL;如果设置了redirect_uri,还会自动附加oauth_callback参数。之后引导用户访问该授权 URL 并完成授权即可。
第四步:解析回调并获取访问令牌
用户授权后,服务商会带着临时凭证与验证码重定向回注册的回调地址。例如:
https://example.com/twitter?oauth_token=gA..H&oauth_verifier=fcg..1Dq如果前面设置了redirect_uri,回调地址则是:
https://your-domain.org/auth?oauth_token=gA..H&oauth_verifier=fcg..1Dq在同一个会话内,用parse_authorization_response解析回调 URL,再用fetch_access_token换取最终令牌:
>>> resp_url = 'https://example.com/twitter?oauth_token=gA..H&oauth_verifier=fcg..1Dq' >>> client.parse_authorization_response(resp_url) >>> access_token_url = 'https://api.twitter.com/oauth/access_token' >>> token = client.fetch_access_token(access_token_url) >>> print(token) { 'oauth_token': '12345-st..E', 'oauth_token_secret': 'o67..X', 'user_id': '12345', 'screen_name': 'lepture', 'x_auth_expires': '0' } >>> save_access_token(token)parse_authorization_response从回调 URL 的查询串中解出oauth_token、oauth_verifier等参数并写入会话(client.py)。fetch_access_token则会检查 verifier:既可以在调用时显式传入fetch_access_token(url, verifier),也可以依赖之前解析或构造时设置的 verifier;两者都缺失时会抛出OAuthError('missing_verifier', 'Missing "verifier" value')。测试用例 test_oauth1_session.py 与 test_oauth1_client.py 都专门验证了"未传 verifier 必须报错"这一约束。
最终得到的 token 字典需保存(如存入数据库),其中oauth_token与oauth_token_secret是访问受保护资源的凭证。
会话重建:跨请求场景的正确姿势
上面的流程是"单一会话内完成全部步骤"的理想情形,但真实项目中往往不是这样:当用户被重定向到授权端点时,会话已经结束了。服务商把用户送回服务器时,你需要用保存的 Request Token 重新创建一个会话:
>>> # restore your saved request token, which is a dict >>> request_token = restore_request_token() >>> oauth_token = request_token['oauth_token'] >>> oauth_token_secret = request_token['oauth_token_secret'] >>> from authlib.integrations.requests_client import OAuth1Session >>> # if using httpx: from authlib.integrations.httpx_client import AsyncOAuth1Client >>> client = OAuth1Session( ... client_id, client_secret, ... token=oauth_token, ... token_secret=oauth_token_secret) >>> # there is no need for `parse_authorization_response` if you can get `verifier` >>> verifier = request.args.get('verifier') >>> access_token_url = 'https://api.twitter.com/oauth/access_token' >>> token = client.fetch_access_token(access_token_url, verifier)这里的关键点在于:把之前保存的 Request Token 以token/token_secret参数传入构造器,即可在"新会话"中继续 OAuth 1.0 流程的收尾步骤;如果框架层面能直接拿到verifier查询参数,则parse_authorization_response可以省略。OAuth1Session还支持通过token属性以字典形式整体注入或清空令牌(token = None会同时清空 token、token_secret 与 verifier),这在 test_oauth1_session.py 的test_set_token中有完整验证。
签名访问受保护资源
拿到 Access Token 后即可请求受保护资源。若继续复用之前那个会话,直接调用即可:
>>> account_url = 'https://api.twitter.com/1.1/account/verify_credentials.json' >>> resp = client.get(account_url) <Response [200]> >>> resp.json() {...}更贴合真实场景的做法是:像上一节那样用数据库中保存的 Access Token 重建会话,再进行请求:
>>> access_token = restore_access_token_from_database() >>> oauth_token = access_token['oauth_token'] >>> oauth_token_secret = access_token['oauth_token_secret'] >>> # if using httpx: from authlib.integrations.httpx_client import AsyncOAuth1Client >>> client = OAuth1Session( ... client_id, client_secret, ... token=oauth_token, ... token_secret=oauth_token_secret) >>> account_url = 'https://api.twitter.com/1.1/account/verify_credentials.json' >>> resp = client.get(account_url)每个请求发出时,底层OAuth1Auth(requests 版见 oauth1_session.py,httpx 版见 oauth1_client.py)都会生成新的 nonce 与 timestamp,对请求进行签名并注入Authorization头。需要注意:requests 版的rebuild_auth在重定向时会先移除旧的Authorization头再重新签名,因为nonce 按 OAuth 规范不可复用。
异步客户端(AsyncOAuth1Client)的用法与同步完全一致,只是需要在必要的环节加上await(参考 httpx.rst):
# fetching request token request_token = await client.fetch_request_token(request_token_url) # fetching access token access_token = await client.fetch_access_token(access_token_url) # normal requests await client.get(...) await client.post(...) await client.put(...) await client.delete(...)异步版的fetch_access_token在 oauth1_client.py 中被单独重写:它通过await self.post(...)发送签名请求,并在成功后清空self.auth.verifier,防止 verifier 泄漏到后续请求中。
轻量方案:直接使用 OAuth1Auth
如果不需要完整的会话状态管理,也可以只拿OAuth1Auth对象配合原生 HTTP 库使用。先构造一个携带 Access Token 的认证对象:
# if using requests from authlib.integrations.requests_client import OAuth1Auth # if using httpx from authlib.integrations.httpx_client import OAuth1Auth auth = OAuth1Auth( client_id='..', client_secret='..', token='oauth_token value', token_secret='oauth_token_secret value', ... )requests 用法——把auth传给requests.get:
import requests url = 'https://api.twitter.com/1.1/account/verify_credentials.json' resp = requests.get(url, auth=auth)httpx 用法——把auth传给httpx.get(异步接口同理):
import httpx url = 'https://api.twitter.com/1.1/account/verify_credentials.json' resp = await httpx.get(url, auth=auth)sess = OAuth1Session(...)本质上等价于"Session + 内置 OAuth1Auth"的组合:OAuth1Session的auth_class = OAuth1Auth,每次请求都经由auth完成签名。因此当你在一个 Web 框架里只需要给少数请求附加 OAuth 1.0 签名时,OAuth1Auth是侵入性更小的选择。requests 版本更完整的示例还可以参考 requests.rst。
底层原理:签名算法与签名位置
理解上面的所有 API 行为,离不开 OAuth 1.0 签名机制。签名由 authlib/oauth1/rfc5849/client_auth.py 中的ClientAuth.sign()驱动,流程如下:
- 生成
oauth_nonce(随机串)与oauth_timestamp(Unix 时间戳); - 组装基础 OAuth 参数:
oauth_nonce、oauth_timestamp、oauth_version="1.0"、oauth_signature_method、oauth_consumer_key,并按需追加oauth_token、oauth_callback、oauth_verifier(见get_oauth_params); - 对非表单类型且有内容的请求体,按 draft-eaton-oauth-bodyhash 规范额外附加
oauth_body_hash(对 body 做 SHA-1 后 Base64); - 渲染一次请求以计算签名,再把
oauth_signature追加到参数中二次渲染出最终请求; - 按
signature_type决定参数落位:HEADER放入Authorization: OAuth ...头,QUERY拼进 URL,BODY写入application/x-www-form-urlencoded表单体。
签名算法本身实现在 authlib/oauth1/rfc5849/signature.py,支持三种:
- HMAC-SHA1(默认):密钥为
escape(client_secret) + "&" + escape(token_secret),对 RFC 5849 定义的签名基串做 HMAC-SHA1 后 Base64; - RSA-SHA1:使用
rsa_key私钥对签名基串做 RSASSA-PKCS1-v1_5 签名,适合需要更强密钥管理能力的场景; - PLAINTEXT:直接把两个密钥拼接(
client_secret & token_secret)作为签名,必须配合 TLS/SSL 传输层保护,且不使用 timestamp 与 nonce。
签名基串由 HTTP 方法、规范化后的 URI(统一小写、去除默认端口、空路径补/)以及按字节序排序后的参数对拼接而成,具体逻辑见construct_base_string与normalize_parameters。ClientAuth还提供register_signature_method(name, sign)类方法,允许注册自定义签名函数来扩展签名算法(见 client_auth.py)。
三种签名方法与三种签名位置的组合行为,在 test_oauth1_session.py(test_signature_types、test_signature_methods)以及 test_oauth1_client.py、test_async_oauth1_client.py(test_fetch_request_token_via_header/body/query等)中都有逐字验证:测试断言了oauth_consumer_key、oauth_signature出现在 Authorization 头 / 表单体 / URL 查询串中,并用verify_hmac_sha1反向校验签名真实性。
小结
Authlib 将 OAuth 1.0(RFC 5849)客户端抽象为"会话类 + 认证对象"两层:OAuth1Session/OAuth1Client/AsyncOAuth1Client负责三步授权流程与受保护资源访问,OAuth1Auth提供轻量级签名能力;同步与异步 API 完全对齐,切换成本极低。使用时牢记三个要点:Request Token 与 Access Token 都要持久化保存、跨请求场景需要用token/token_secret重建会话、fetch_access_token必须携带 verifier。至于 HMAC-SHA1 / RSA-SHA1 / PLAINTEXT 三种签名算法以及 HEADER / QUERY / BODY 三种签名位置,则按服务商要求与安全级别在初始化时配置即可。
【免费下载链接】authlib
The ultimate Python library in building OAuth, OpenID Connect clients and servers. JWS, JWE, JWK, JWA, JWT included.
相关推荐
JavaScript 中 `if ("0")` 为何为真:深入解析字符串与布尔转换(en.javascript.info 实战任务)
JavaScript 中 if "0" 为何为真:深入解析字符串与布尔转换(en.javascript.info 实战任务) if "0" 的判断结果是什么?这
Elixir 状态管理实战:用 Agent 封装可共享状态,从 KV.Bucket 入门 OTP
Elixir 状态管理实战:用 Agent 封装可共享状态,从 KV.Bucket 入门 OTP 本文是 Elixir 官方 Mix & OTP 指南 的“用
Chamber高级用法:标签管理和版本控制完全解析
Chamber高级用法:标签管理和版本控制完全解析 Chamber作为一款强大的AWS密钥管理CLI工具,提供了丰富的标签管理和版本控制功能,帮助团队高效管理敏
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考