asigv4

Async AWS Signature V4 — framework-agnostic, stdlib-only verify / sign / presign.

• 零运行时依赖：纯 Python 标准库（hmac / hashlib / re / datetime / urllib）
• 全异步 I/O：密钥查找、请求体读取均为 async；签名/预签名为同步纯函数
• 框架无关：核心 verify() 接受原始参数，不绑定任何 Web 框架；body 读取由调用方按自身节奏控制
• 时间完全可控：now 注入签名/校验时间、clock 注入时钟源、max_skew_seconds=None 关闭偏差检查
• Python 3.10+，附 py.typed 类型标记（PEP 561）

安装

pip install asigv4

快速开始

服务端校验

SigV4Verifier.verify() 接受原始参数，调用方从自己框架的 Request 对象中提取即可。 body 接受 bytes 或返回 bytes 的协程，仅在 payload 需要计算哈希时才会被调用 （UNSIGNED-PAYLOAD / STREAMING-* 会跳过，大文件零开销）。

from asigv4 import SigV4Verifier

KEYS = {"AKIAIOSFODNN7EXAMPLE": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"}
verifier = SigV4Verifier(lambda ak: KEYS.get(ak))

# 示例：从任意框架的 request 中提取参数后调用
async def handle(request):
    access_key_id = await verifier.verify(
        method=request.method,
        path=request.url.path,
        headers={k.lower(): v for k, v in request.headers.items()},
        query_params=dict(request.query_params),
        body=request.body,   # bytes 或 callable，仅在需要时才被 await
    )
    return access_key_id

密钥查找回调兼容同步与异步：

async def db_lookup(access_key_id: str) -> str | None:
    row = await pool.fetchrow(
        "SELECT secret_key FROM access_keys WHERE key_id = $1", access_key_id
    )
    return row["secret_key"] if row else None

verifier = SigV4Verifier(db_lookup)   # 直接传 async 函数

生成预签名 URL

from asigv4 import presign_url

url = presign_url(
    method="GET",
    host="bucket.s3.amazonaws.com",
    path="/bucket/key.txt",
    access_key_id="AKIAIOSFODNN7EXAMPLE",
    secret_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    expires=3600,
)
# scheme="http" 可用于本地测试或 TLS 终止代理

客户端签名请求

from asigv4 import sign_request

auth_header = sign_request(
    method="PUT",
    path="/bucket/key.txt",
    headers={"host": "bucket.s3.amazonaws.com"},
    body=b"file content",
    access_key_id="AKIAIOSFODNN7EXAMPLE",
    secret_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
)

时间控制

三档自由控制，覆盖测试注入、时钟漂移补偿、合规审计等场景：

from datetime import datetime, timezone
from asigv4 import SigV4Verifier, sign_request

# 1) 签名时固定 now → 完全确定性（测试 / 互操作向量）
fixed = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
auth = sign_request(method="GET", path="/k", headers={"host": "h"},
                    access_key_id=AK, secret_key=SK, now=fixed)

# 2) 校验时注入 clock → 固定时间源（测试 / 已知时钟漂移补偿）
verifier = SigV4Verifier(lambda ak: KEYS.get(ak), clock=lambda: fixed)
ak = await verifier.verify(method="GET", path="/k",
                           headers=headers, query_params={})

# 3) per-call now 覆盖构造器 clock
ak = await verifier.verify(method="GET", path="/k",
                           headers=headers, query_params={}, now=fixed)

# 4) max_skew_seconds=None 彻底关闭偏差检查（完全自由控制时间）
verifier = SigV4Verifier(lambda ak: KEYS.get(ak), max_skew_seconds=None)

异常

所有错误均为 SigV4Error 子类：

捕获基类 SigV4Error 即可统一处理（推荐，避免泄露 key 是否存在）。

API

详见 sign_design.md。