1. 项目背景业务场景某安全团队的渗透测试发现一个严重漏洞部分微服务在调用第三方 API 时即使 HTTP 状态码是 401认证失败代码仍然继续处理响应体——因为开发者用resp.status_code 200做判断而 requests 在收到 401 后不会自动抛异常。这个漏洞的修复需要理解 requests 的认证流程auth 是如何在 Session.send() 中被应用的为什么raise_for_status()不会自动被调用另一个场景更隐蔽他们在实现一个 OAuth2 客户端遇到了 Token 刷新时多线程并发刷新的问题。通过阅读auth.py中HTTPDigestAuth的 401 处理流程他们理解了 auth 重试机制的单次性——requests 只重试一次不会无限循环。痛点问题一AuthBase 协议的本质是 PreparedRequest 的修改器。auth.py的核心抽象极简——一个__call__(self, r: PreparedRequest) - PreparedRequest方法。但很少有开发者理解auth 在什么时机被调用如何与 Session 的其他设置交互如果 auth 返回了一个全新的 PreparedRequest 而不是修改原对象会发生什么问题二HTTPDigestAuth 的完整实现是 auth.py 中最复杂的代码。它涉及 MD5 哈希计算、nonce 管理、qopquality of protection协商、401 响应解析——约 150 行代码。理解这段代码不仅帮你理解 Digest Auth更是学习如何在 AuthBase 中实现有状态的认证协议的最佳案例。问题三认证与重定向的交互。当发生 3xx 重定向时auth 信息是否会被带过去默认行为是对跨域重定向requests 会剥离 Authorization 头防止凭据泄露到不信任的域。这个行为在rebuild_auth()函数中实现。auth.py 核心类关系图AuthBase (抽象基类) ├── __call__(self, r: PreparedRequest) - PreparedRequest │ ├── HTTPBasicAuth │ └── __call__: r.headers[Authorization] Basic b64(user:pass) │ ├── HTTPDigestAuth │ ├── __call__: 处理 401 响应, 计算 Digest response │ ├── build_digest_header(): 构造 WWW-Authenticate 响应 │ ├── handle_401(): 解析 nonce/qop/algorithm/opaque │ └── HA1/HA2/response 的 MD5 计算链 │ └── 自定义 AuthBase (用户实现) └── __call__: 任意逻辑修改 PreparedRequest 工具函数: extract_cookies_to_jar(): 从响应中提取 Cookie 到 CookieJar _basic_auth_str(): 构造 Basic Auth 的 Base64 字符串2. 项目设计小胖对着日志输出迷惑“大师我用 Session 设置了authHTTPBasicAuth(user,pass)然后请求一个不需要认证的接口——为什么日志里同一个 URL 出现了两次请求一次 401一次 200”大师微笑着打开 auth.py 源码“这就是 requests 的 auth 重试机制。Session.send()在发送请求时不会主动带上 Authorization 头避免凭据泄露给不信任的服务器。它先发一个不带凭据的请求——如果收到 401并且响应头中有WWW-Authenticate就再用 auth 处理器重试一次。所以你看到的两次请求是正常的——第一次试探第二次认证。”小白推了推眼镜“那为什么HTTPDigestAuth比HTTPBasicAuth复杂那么多Basic Auth 就一行 Base64Digest Auth 好像有上百行代码。”大师“因为 Digest Auth 需要解析服务器返回的 401 响应——提取 nonce、realm、qop、algorithm、opaque 等参数用这些参数加上用户名密码计算出 response 值。而且它需要在第一次请求失败后收到 401解析 WWW-Authenticate 头重新计算再发第二次请求。整个状态机比 Basic Auth 复杂得多。”# auth.py 中 HTTPDigestAuth.handle_401() 的核心逻辑 (简化)defhandle_401(self,r,**kwargs):# 1. 解析 WWW-Authenticate: Digest realm..., nonce...www_authr.headers.get(www-authenticate,)# 2. 提取参数: realm, nonce, qop, algorithm, opaque# 3. 计算 HA1 MD5(user:realm:pass)# 4. 计算 HA2 MD5(method:uri)# 5. 计算 response MD5(HA1:nonce:nc:cnonce:qop:HA2)# 6. 构建 Authorization: Digest ...头# 7. 重新发送带认证头的请求小胖“那rebuild_auth()是干什么的我发现在重定向时 Authorization 头会被丢弃”大师“对——这是安全设计。rebuild_auth()在重定向时判断如果新 URL 的域名和原 URL 不同跨域重定向就剥离 Authorization 头防止凭据泄露到不信任的域。如果同域就保留。”# Session.send() 中的 redirect 循环 (简化)whileTrue:respadapter.send(request)ifresp.status_codein(301,302,303,307,308):requestrebuild_auth(request,resp)# 剥离或保留 authcontinuebreak生活比喻技术映射先敲门再报身份首次无凭据请求 - 401 - 带凭据请求银行柜台的挑战-应答HTTPDigestAuth 的 nonce 机制跨省办事需要重新出示身份证跨域重定向剥离 Authorization自助制卡机插卡-认证-出卡AuthBase.call修改 PreparedRequest万能钥匙一把钥匙开多道门Session.auth 全局认证3. 项目实战环境准备python-cimport requests.auth; print(requests.auth.__file__)分步实现步骤一HTTPBasicAuth 源码走读目标从源码层面理解 Basic Auth 的实现。importrequestsfromrequests.authimportHTTPBasicAuth,_basic_auth_strimportbase64defdemo_basic_auth_internals():演示 HTTPBasicAuth 的源码实现# HTTPBasicAuth.__call__ 的核心逻辑 (源码简化版):# def __call__(self, r):# r.headers[Authorization] _basic_auth_str(self.username, self.password)# return r# _basic_auth_str 的实现:# def _basic_auth_str(username, password):# if isinstance(username, str):# username username.encode(latin1)# if isinstance(password, str):# password password.encode(latin1)# auth base64.b64encode(b:.join((username, password))).strip()# return Basic auth.decode(ascii)# 关键细节: 用户名和密码用 latin1 编码非 UTF-8!# 这意味着中文用户名/密码可能编码失败authHTTPBasicAuth(admin,secret123)# 构建 PreparedRequestsessionrequests.Session()reqrequests.Request(GET,https://httpbin.org/basic-auth/admin/secret123)preparedsession.prepare_request(req)print(发送前 (无auth):)print(f Authorization:{prepared.headers.get(Authorization,(无))})# 应用 authpreparedauth(prepared)print(f\n应用 auth 后:)print(f Authorization:{prepared.headers[Authorization]})# 解码验证encodedprepared.headers[Authorization].split( ,1)[1]decodedbase64.b64decode(encoded).decode(latin1)print(f 解码后:{decoded})# 源码中的 latin1 编码限制:# username.encode(latin1) — 非 latin1 字符会报错try:HTTPBasicAuth(中文用户,pass)print(\n 注意: 用户名含中文时 latin1 编码可能失败)exceptUnicodeEncodeError:print(\n ⚠ 中文用户名会导致 latin1 编码错误!)session.close()demo_basic_auth_internals()步骤二HTTPDigestAuth 的 401 处理流程目标理解 Digest Auth 的 challenge-response 机制。importrequestsfromrequests.authimportHTTPDigestAuthimporthashlibdefdemo_digest_auth_flow():演示 Digest Auth 的 challenge-response 流程# 简化版计算演示真实的 HTTPDigestAuth 更复杂usernameuserpasswordpasswdrealmmekennethreitz.com# httpbin 的 realmmethodGETuri/digest-auth/auth/user/passwdnonceexample-nonce# 服务器返回的随机数# 1. HA1 MD5(username:realm:password)ha1hashlib.md5(f{username}:{realm}:{password}.encode()).hexdigest()print(fHA1 MD5({username}:{realm}:{password}))print(f {ha1})# 2. HA2 MD5(method:uri)ha2hashlib.md5(f{method}:{uri}.encode()).hexdigest()print(f\nHA2 MD5({method}:{uri}))print(f {ha2})# 3. response MD5(HA1:nonce:HA2)responsehashlib.md5(f{ha1}:{nonce}:{ha2}.encode()).hexdigest()print(f\nresponse MD5({ha1[:8]}...:{nonce}:{ha2[:8]}...))print(f {response})# 4. 构造 Authorization 头auth_header(fDigest username{username}, frealm{realm}, fnonce{nonce}, furi{uri}, fresponse{response})print(f\nAuthorization:{auth_header[:80]}...)# 实际使用print(f\n 真实 Digest Auth 请求 )try:resprequests.get(https://httpbin.org/digest-auth/auth/user/passwd,authHTTPDigestAuth(user,passwd),)print(f状态码:{resp.status_code})print(f认证:{resp.json().get(authenticated,N/A)})exceptExceptionase:print(f注意:{e})demo_digest_auth_flow()步骤三从源码角度实现一个自定义 Auth——自动 Token 刷新目标基于对 auth 源码的理解实现一个生产级的 OAuth2 AuthBase。importrequestsfromrequests.authimportAuthBaseimporttimeimportthreadingclassAutoRefreshOAuth2Auth(AuthBase):生产级 OAuth2 自动刷新认证处理器 设计思路 (源自对 auth.py 源码的理解): - AuthBase.__call__ 在 Session.send() 中被调用 - 在每次请求前检查 Token 是否过期 - 过期时刷新 Token加锁防并发 - 直接修改 PreparedRequest 并返回 def__init__(self,token_url:str,client_id:str,client_secret:str,scope:str,refresh_margin:int60):self.token_urltoken_url self.client_idclient_id self.client_secretclient_secret self.scopescope self.refresh_marginrefresh_margin self._access_tokenNoneself._expires_at0self._lockthreading.Lock()def__call__(self,r):核心方法——在每次请求前被 Session.send() 调用ifself._should_refresh():self._refresh_token_sync()r.headers[Authorization]fBearer{self._access_token}returnrdef_should_refresh(self)-bool:returnself._access_tokenisNoneortime.time()self._expires_atdef_refresh_token_sync(self):withself._lock:# 双重检查可能已被其他线程刷新ifnotself._should_refresh():returnprint([OAuth2] 正在刷新 Token...)resprequests.post(self.token_url,data{grant_type:client_credentials,client_id:self.client_id,client_secret:self.client_secret,scope:self.scope,},timeout10)resp.raise_for_status()dataresp.json()self._access_tokendata[access_token]expires_indata.get(expires_in,3600)self._expires_attime.time()expires_in-self.refresh_marginprint(f[OAuth2] Token 刷新成功, 有效期{expires_in}s)propertydeftoken_info(self)-dict:ifself._access_tokenisNone:return{status:no_token}remainingself._expires_at-time.time()return{status:activeifremaining0elseexpired,remaining_s:round(remaining,0),}# 使用演示 print( OAuth2 AutoRefresh 认证器 )print(用法: session.auth AutoRefreshOAuth2Auth(...))print( - 首次请求自动获取 Token)print( - Token 过期前 margin 秒自动刷新)print( - 并发安全threading.Lock 双重检查)print( - 基于 AuthBase 协议零侵入)可能遇到的坑及解决方法坑1Basic Auth 的 latin1 编码限制# HTTPBasicAuth 用 latin1 编码用户名和密码非 UTF-8# 中文用户名会抛出 UnicodeEncodeError# 解决: 如果必须用中文自定义 AuthBase 使用 UTF-8 编码坑2Digest Auth 的 qop 兼容性# 不同服务器对 qop (quality of protection) 的支持不同# 有的服务器要求 qopauth有的不支持 qop# HTTPDigestAuth 源码中有复杂的 qop 协商逻辑坑3auth 在 Session.send() 中被调用两次# 第一次: 无凭据请求 - 401# 第二次: 带凭据请求 - 200# 如果你的 AuthBase 有副作用如计数、日志会被调用两次测试验证importpytestimportrequestsfromrequests.authimportHTTPBasicAuth,AuthBaseclassTestAuthInternals:验证 auth 内部机制deftest_basic_auth_encodes_correctly(self):fromrequests.authimport_basic_auth_str auth_str_basic_auth_str(user,pass)assertauth_str.startswith(Basic )importbase64 decodedbase64.b64decode(auth_str.split()[1]).decode(latin1)assertdecodeduser:passdeftest_auth_modifies_prepared_request(self):srequests.Session()reqrequests.Request(GET,https://httpbin.org/headers)preps.prepare_request(req)authHTTPBasicAuth(testuser,testpass)prepauth(prep)assertAuthorizationinprep.headers s.close()deftest_custom_auth_returns_modified_request(self):classDummyAuth(AuthBase):def__call__(self,r):r.headers[X-Auth-Dummy]yesreturnr srequests.Session()s.authDummyAuth()resps.get(https://httpbin.org/headers)assertresp.json()[headers][X-Auth-Dummy]yess.close()4. 项目总结核心源码要点类/函数关键逻辑复杂度AuthBase抽象基类单方法__call__(self, r) - r低HTTPBasicAuthBase64 编码 user:passlatin1 编码低HTTPDigestAuth解析 401 - 计算 MD5 - 构建 Digest 头高_basic_auth_str()Base64 编码辅助函数低extract_cookies_to_jar()从 Response 提取 Cookie中适用场景理解 requests 认证体系的扩展机制实现自定义认证协议HMAC、OAuth2、AWS SigV4排查 401 重试和跨域认证剥离问题学习 Challenge-Response 协议的设计模式注意事项Basic Auth 使用 latin1 编码非 UTF-8中文用户名可能报错auth 在 Session.send() 中被调用可能存在副作用跨域重定向时 Authorization 头会被剥离安全设计HTTPDigestAuth 依赖 MD5——在现代安全标准中已不推荐常见踩坑经验案例一Basic Auth 用户名含特殊字符。用户名中有冒号:Base64 编码时user:pass中的:导致解析歧义——服务端解析出的用户名是user 冒号前的部分密码包含了冒号后的所有内容。根因username 中的:未被转义。修复username 中的:替换为%3A或避免在用户名中使用冒号。案例二OAuth2 Token 刷新时序竞争。多线程下 10 个请求同时检测到 Token 过期触发 10 次刷新请求——认证服务器短暂过载。虽然加了线程锁但_should_refresh()中的time.time() self._expires_at在锁外判断10 个线程都拿到了 True 然后排队等锁。根因双重检查的第一次检查未完全防护。修复在锁内再次检查。思考题源码题阅读auth.py中HTTPDigestAuth.build_digest_header()的源码找出它是如何处理qopauth-int带请求体完整性保护的 qop与qopauth的差异的设计题设计一个CompositeAuth——组合多个 AuthBase按顺序依次修改 PreparedRequest。例如先注入 API Key再计算 HMAC 签名签名需要 API Key 作为输入。如何保证顺序正确和独立可替换延伸阅读与资源Milvus向量数据库实战修炼从 0 到 1精通向量检索与生产落地后端工程师的 AI 转型第一课Ollama 与私有化大模型实战10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析