将Requests的HTTP Digest Auth转aiohttp认证时遇401错误求助

核心问题在于：requests的HTTPDigestAuth处理的是HTTP摘要认证，而aiohttp的BasicAuth仅支持基础认证，二者是完全不同的认证协议，直接替换必然返回401未授权。

一、原Requests代码逻辑（对比参考） #

原Requests会自动完成摘要认证的握手流程：先发送无认证请求，收到401后提取响应头的摘要参数，生成符合规则的认证头再重新发送请求。示例代码：

import requests
from requests.auth import HTTPDigestAuth

url = "https://your-api-url.com"
payload = {"param": "value"}
resp = requests.post(url, data=payload, auth=HTTPDigestAuth("your-username", "your-password"))

二、aiohttp的两种解决方案 #

方案1：使用第三方库（简单快捷）

aiohttp官方没有内置摘要认证实现，但可以用aiohttp_digest_auth库直接替代HTTPDigestAuth的功能：

1. 安装依赖

pip install aiohttp_digest_auth

2. 异步请求代码

import asyncio
import aiohttp
from aiohttp_digest_auth import DigestAuth

async def send_request():
    url = "https://your-api-url.com"
    payload = {"param": "value"}
    auth = DigestAuth("your-username", "your-password")
    
    async with aiohttp.ClientSession(auth=auth) as session:
        async with session.post(url, data=payload) as resp:
            print(f"状态码: {resp.status}")
            print(f"响应内容: {await resp.text()}")

asyncio.run(send_request())

这个库会自动处理摘要认证的两次请求流程，和Requests的行为完全一致。

方案2：手动实现摘要认证（无第三方依赖）

如果不想引入额外库，可以手动模拟Requests的流程：先发送无认证请求触发401，解析响应头的摘要参数，再生成符合规则的认证头重新请求：

import asyncio
import aiohttp
from hashlib import md5

def build_digest_auth_header(username, password, method, url, auth_params):
    # 从WWW-Authenticate头中提取参数
    realm = auth_params["realm"]
    nonce = auth_params["nonce"]
    qop = auth_params.get("qop")
    algorithm = auth_params.get("algorithm", "MD5")

    # 计算HA1（用户名:域:密码的MD5哈希）
    ha1 = md5(f"{username}:{realm}:{password}".encode()).hexdigest()
    # 计算HA2（请求方法:请求URL的MD5哈希）
    ha2 = md5(f"{method}:{url}".encode()).hexdigest()

    # 生成响应值
    if qop:
        nc = "00000001"  # 首次请求的nonce计数
        cnonce = md5(b"random_seed").hexdigest()[:16]  # 随机客户端nonce
        response = md5(f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}".encode()).hexdigest()
        return (
            f'Digest username="{username}", realm="{realm}", nonce="{nonce}", uri="{url}", '
            f'qop={qop}, nc={nc}, cnonce="{cnonce}", response="{response}", algorithm={algorithm}'
        )
    else:
        response = md5(f"{ha1}:{nonce}:{ha2}".encode()).hexdigest()
        return (
            f'Digest username="{username}", realm="{realm}", nonce="{nonce}", uri="{url}", '
            f'response="{response}", algorithm={algorithm}'
        )

async def send_request():
    url = "https://your-api-url.com"
    payload = {"param": "value"}
    username = "your-username"
    password = "your-password"
    method = "POST"

    async with aiohttp.ClientSession() as session:
        # 第一步：发送无认证请求，获取认证参数
        async with session.post(url, data=payload) as resp:
            if resp.status != 401:
                print(f"状态码: {resp.status}")
                print(f"响应内容: {await resp.text()}")
                return

            # 解析WWW-Authenticate头
            auth_header = resp.headers.get("WWW-Authenticate")
            if not auth_header or not auth_header.startswith("Digest "):
                raise ValueError("目标服务器不支持HTTP摘要认证")
            
            auth_params = {}
            for part in auth_header[7:].split(", "):
                key, val = part.split("=", 1)
                auth_params[key] = val.strip('"')

        # 第二步：生成认证头并重新发送请求
        digest_header = build_digest_auth_header(username, password, method, url, auth_params)
        async with session.post(url, data=payload, headers={"Authorization": digest_header}) as resp:
            print(f"状态码: {resp.status}")
            print(f"响应内容: {await resp.text()}")

asyncio.run(send_request())

关键说明 #

• 摘要认证的核心是根据服务器返回的nonce、realm等参数，结合用户名密码、请求方法和URL，按照MD5哈希规则生成response值，放入Authorization头。
• 第三方库的方式更稳定，无需自己处理哈希计算和参数解析的细节，推荐优先使用。

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