如何结合requests_aws4auth与aiohttp实现AWS S3认证请求?
解决aiohttp结合AWS4Auth调用S3 API的签名问题
问题分析
你当前的代码存在两个核心错误:
- 错误读取signing_key属性作为请求头:
signing_key是AWS4Auth内部用于生成签名的派生密钥,并非请求需要的Authorization、x-amz-date等头字段,直接将其属性塞入请求头会导致签名验证失败(400错误)。 - 复用其他请求的签名头:AWS签名v4的每个签名都与请求方法、资源路径、时间戳、请求内容哈希等强绑定,GET请求的签名头无法复用给PUT请求,会触发403权限错误。
正确解决方案
不需要重写AWS签名v4实现,直接利用requests_aws4auth的sign方法为每个aiohttp请求生成专属签名头即可。
方案1:手动生成签名头(简单直接)
self.aws4auth = AWS4Auth(access_key, secret_key, 'us-east-1', 's3') with open(file_path, 'rb') as file: content = file.read() url = f'{self.host}/{bucket}/{object_name}' # 为当前PUT请求生成完整的签名头 signed_headers = self.aws4auth.sign( method='PUT', url=url, body=content ) async with aiohttp.ClientSession() as session: async with session.put(url, data=content, headers=signed_headers) as resp: print(resp.status) print(await resp.text())
方案2:用aiohttp拦截器自动签名(适合批量请求)
如果需要多次发起请求,可以自定义拦截器,在每个请求发送前自动生成签名:
import aiohttp from requests_aws4auth import AWS4Auth class AWS4AuthInterceptor: def __init__(self, aws4auth): self.aws4auth = aws4auth async def on_request_start(self, session, trace_config_ctx, params): # 提取请求的核心信息 method = params.method url = str(params.url) body = params.data # 生成签名头并合并到请求中 signed_headers = self.aws4auth.sign(method=method, url=url, body=body) params.headers.update(signed_headers) # 使用示例 self.aws4auth = AWS4Auth(access_key, secret_key, 'us-east-1', 's3') # 创建拦截器 trace_config = aiohttp.TraceConfig() trace_config.on_request_start.append(AWS4AuthInterceptor(self.aws4auth).on_request_start) async with aiohttp.ClientSession(trace_configs=[trace_config]) as session: with open(file_path, 'rb') as file: content = file.read() url = f'{self.host}/{bucket}/{object_name}' async with session.put(url, data=content) as resp: print(resp.status) print(await resp.text())
关键说明
- AWS签名v4的签名是请求专属的,必须针对每个请求的方法、URL、内容等单独生成,不能复用其他请求的签名头。
requests_aws4auth的sign方法已经封装了完整的签名逻辑,直接调用即可生成符合要求的请求头,无需自行实现签名流程。
内容的提问来源于stack exchange,提问作者Brian Postow
相关产品推荐
相关产品推荐

