Python Requests Session复用HTTPDigestAuth认证异常问题咨询

问题根源 #

requests库自带的HTTPDigestAuth是无状态实现，每次请求都会重新触发完整的认证流程，不会在Session实例中保存nc（nonce计数）、cnonce等关键认证状态，因此即便使用Session，后续请求仍会重复认证，nc值始终停留在1。

解决方案：实现带状态的Digest认证类 #

我们可以基于requests的HTTPDigestAuth扩展，让它在Session内持久化认证状态，自动递增nc并复用已有认证信息，避免重复认证。

代码实现 #

import requests
from requests.auth import HTTPDigestAuth
import hashlib
import time

class StatefulDigestAuth(HTTPDigestAuth):
    def __init__(self, username, password):
        super().__init__(username, password)
        self.nonce_count = 0
        self.cnonce = None
        self.auth_state = {}  # 存储realm/nonce/opaque/qop等状态

    def handle_401(self, r, **kwargs):
        # 解析响应中的Digest认证头，保存状态
        auth_header = r.headers.get('WWW-Authenticate', '')
        if 'Digest' in auth_header:
            self.auth_state = self._parse_auth_header(auth_header)
        
        # 更新认证状态计数与随机值
        self.nonce_count += 1
        self.cnonce = self.cnonce or hashlib.md5(str(time.time()).encode()).hexdigest()[:16]
        
        # 重新构建认证头并发送请求
        req = r.request
        req.headers['Authorization'] = self._build_auth_header(req.method, req.url)
        r.content
        r.close()
        new_r = r.connection.send(req, **kwargs)
        new_r.history.append(r)
        new_r.request = req
        return new_r

    def _parse_auth_header(self, header):
        # 把Digest头解析为键值对字典
        auth_dict = {}
        for part in header.split(','):
            if '=' in part:
                key, val = part.strip().split('=', 1)
                auth_dict[key.lower()] = val.strip('"')
        return auth_dict

    def _build_auth_header(self, method, url):
        # 计算Digest认证所需的HA1/HA2/response值
        ha1 = self._calc_ha1()
        ha2 = self._calc_ha2(method, url)
        response = self._calc_response(ha1, ha2)
        
        # 拼接完整的认证头字符串
        auth_parts = [
            f'Digest username="{self.username}"',
            f'realm="{self.auth_state["realm"]}"',
            f'nonce="{self.auth_state["nonce"]}"',
            f'uri="{url.split("?")[0]}"',
            f'response="{response}"',
            f'cnonce="{self.cnonce}"',
            f'nc={self.nonce_count:08x}',
            f'qop={self.auth_state.get("qop", "auth")}'
        ]
        if self.auth_state.get('opaque'):
            auth_parts.append(f'opaque="{self.auth_state["opaque"]}"')
        if self.auth_state.get('algorithm'):
            auth_parts.append(f'algorithm={self.auth_state["algorithm"]}')
        return ', '.join(auth_parts)

    def _calc_ha1(self):
        # 计算HA1值，兼容MD5和MD5-sess算法
        base = f"{self.username}:{self.auth_state['realm']}:{self.password}".encode()
        ha1 = hashlib.md5(base).hexdigest()
        if self.auth_state.get('algorithm') == 'MD5-sess':
            ha1 = hashlib.md5(f"{ha1}:{self.auth_state['nonce']}:{self.cnonce}".encode()).hexdigest()
        return ha1

    def _calc_ha2(self, method, url):
        # 计算HA2值，兼容auth和auth-int模式
        if self.auth_state.get('qop') == 'auth-int':
            ha2 = hashlib.md5(f"{method}:{url}:".encode()).hexdigest()  # 若有请求体需替换为空字符串
        else:
            ha2 = hashlib.md5(f"{method}:{url.split('?')[0]}".encode()).hexdigest()
        return ha2

    def _calc_response(self, ha1, ha2):
        # 计算最终的response签名值
        qop = self.auth_state.get('qop')
        if qop:
            return hashlib.md5(f"{ha1}:{self.auth_state['nonce']}:{self.nonce_count:08x}:{self.cnonce}:{qop}:{ha2}".encode()).hexdigest()
        else:
            return hashlib.md5(f"{ha1}:{self.auth_state['nonce']}:{ha2}".encode()).hexdigest()

使用方法 #

将原代码中的HTTPDigestAuth替换为自定义的StatefulDigestAuth，配合Session使用即可：

with requests.Session() as session:
    # 初始化带状态的认证实例
    session.auth = StatefulDigestAuth('你的用户名', '你的密码')
    
    # 第一次请求（触发401认证）
    resp1 = session.get('https://你的API地址/端点')
    print(resp1.request.headers['Authorization'])  # 可看到nc=00000001
    
    # 第二次请求（复用认证状态，nc自动递增）
    resp2 = session.get('https://你的API地址/端点')
    print(resp2.request.headers['Authorization'])  # 可看到nc=00000002

简化GET请求流程 #

基于Session和自定义认证类，你可以把请求逻辑封装成简单函数，避免重复编写请求代码：

def call_api(session, url):
    try:
        resp = session.get(url)
        resp.raise_for_status()
        return resp.json()
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {str(e)}")
        return None

# 批量请求示例
with requests.Session() as session:
    session.auth = StatefulDigestAuth('user', 'pass')
    target_urls = [
        'https://api.example.com/resource1',
        'https://api.example.com/resource2',
        'https://api.example.com/resource3'
    ]
    results = [call_api(session, url) for url in target_urls]

验证效果 #

多次请求后查看Authorization头，会发现nc值从00000001开始逐次递增，说明认证状态已被正确维持，不会重复发起认证握手。

内容的提问来源于stack exchange，提问作者The Frog