FastAPI+Python实现S3服务：PUT请求签名验证失败求助

AI 小北

2026-6-20

问题

使用FastAPI和Python实现S3服务器时，GET、DELETE请求的签名验证可正常工作，但PUT请求的计算签名与提供的签名不匹配。

签名验证代码

async def verify_signature(request: Request):
  try:
    authorization_header = request.headers.get("Authorization")
    amz_date = request.headers.get("x-amz-date")
    if not authorization_header or not amz_date:
        logging.error("Missing authorization or x-amz-date header")
        return False
    logging.debug(f"\n\n=======================================================\n")
    logging.debug(f"Request URL:\n{request.url}")
    logging.debug(f"Authorization Header:\n{authorization_header}")
    logging.debug(f"x-amz-date:\n{amz_date}")

    auth_parts = authorization_header.split(", ")
    credential_part = auth_parts[0]
    signed_headers_part = auth_parts[1]
    signature_part = auth_parts[2]

    credential_scope = (
        credential_part.split(" ")[1].split("Credential=")[1].split("/")[1:]
    )
    credential_scope = "/".join(credential_scope)

    signed_headers = signed_headers_part.split("SignedHeaders=")[-1].split(";")
    provided_signature = signature_part.split("Signature=")[-1]

    logging.debug(f"Signed Headers:\n{signed_headers}")
    logging.debug(f"Credential Scope:\n{credential_scope}")

    headers_dict = {k.lower(): v for k, v in request.headers.items()}
    sorted_headers = {
        k: headers_dict[k] for k in sorted(headers_dict) if k in signed_headers
    }

    logging.debug(f"Headers Dict:\n{headers_dict}")
    logging.debug(f"Sorted Headers:\n{sorted_headers}")

    canonical_uri = request.url.path
    canonical_querystring = request.url.query
    if False and headers_dict.get("x-amz-content-sha256") == "UNSIGNED-PAYLOAD":
        payload_hash = (
            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        )
    else:
        payload_hash = hashlib.sha256(await request.body()).hexdigest()

    sorted_headers["x-amz-content-sha256"] = payload_hash
    canonical_headers = "".join([f"{k}:{v}\n" for k, v in sorted_headers.items()])

    canonical_request = "\n".join(
        [
            request.method,
            canonical_uri,
            canonical_querystring,
            canonical_headers,
            ";".join(signed_headers),
            payload_hash,
        ]
    )

    logging.debug(f"Canonical Request:\n{canonical_request}")

    string_to_sign = "\n".join(
        [
            ALGORITHM,
            amz_date,
            credential_scope,
            hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(),
        ]
    )

    logging.debug(f"String to Sign:\n{string_to_sign}")

    date_stamp = credential_scope.split("/")[0]
    signing_key = get_signature_key(
        AWS_SECRET_ACCESS_KEY,
        date_stamp,
        AWS_REGION,
        SERVICE,
    )

    signature = hmac.new(
        signing_key, string_to_sign.encode("utf-8"), hashlib.sha256
    ).hexdigest()

    logging.debug(f"Calculated Signature: {signature}")
    logging.debug(f"Provided Signature: {provided_signature}")

    is_valid = provided_signature == signature
    if not is_valid:
        logging.error("Signatures do not match")
    return is_valid

  except Exception as e:
    logging.error(f"Verification failed: {e}")
    return False

测试代码

def test():
    from io import BytesIO

    import boto3

    session = boto3.Session(
        AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, region_name=AWS_REGION
    )
    client = session.client(
        "s3", endpoint_url=ENDPOINT
    )
    f = BytesIO(b"salam2")
    f.seek(0)
    put_res = client.put_object(Bucket="mybucket", Key="myfile2.txt", Body=f)
    print(put_res)
    get_res = client.get_object(Bucket="mybucket", Key="myfile2.txt")
    print(get_res["Body"].read())
    del_res = client.delete_object(Bucket="mybucket", Key="myfile2.txt")
    print(del_res)

已参考S3 REST API签名v4官方文档及第三方实现说明。

解决方案

PUT请求签名不匹配的核心问题在于Payload哈希处理逻辑错误，以及对已签名请求头的不当修改，修复步骤如下：

1. 修正Payload哈希获取逻辑

S3签名v4规范中，PUT请求的x-amz-content-sha256头由客户端（如boto3）预先计算并发送，签名验证时必须直接使用该头的原始值，而非自行计算。自行计算可能因请求体读取方式、编码差异导致哈希值与客户端不一致。

替换原代码中Payload哈希的计算逻辑：

# 替换原有的payload_hash计算代码
content_sha256 = headers_dict.get("x-amz-content-sha256")
if content_sha256 == "UNSIGNED-PAYLOAD":
    payload_hash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
elif content_sha256:
    payload_hash = content_sha256
else:
    # 兼容未发送该头的边缘情况，自行计算哈希
    payload_hash = hashlib.sha256(await request.body()).hexdigest()

2. 禁止修改已签名的请求头值

原代码中sorted_headers["x-amz-content-sha256"] = payload_hash会覆盖请求头中的原始值，导致Canonical Headers与客户端签名时使用的头数据不一致，直接删除该行代码：

# 删除此行：sorted_headers["x-amz-content-sha256"] = payload_hash

sorted_headers已从原始请求头中获取了x-amz-content-sha256的值（若它在signed_headers列表中），无需手动修改。

3. 确保Canonical Headers与Signed Headers顺序一致

AWS要求Canonical Headers必须按字母顺序排序，同时signed_headers参数也必须是按字母顺序排列的头名称列表。原代码中sorted_headers已按key排序，这部分逻辑正确，只需确保从Authorization头解析出的signed_headers列表符合排序规则（boto3默认会按此规则发送）。

完整修复后的签名验证代码

async def verify_signature(request: Request):
  try:
    authorization_header = request.headers.get("Authorization")
    amz_date = request.headers.get("x-amz-date")
    if not authorization_header or not amz_date:
        logging.error("Missing authorization or x-amz-date header")
        return False
    logging.debug(f"\n\n=======================================================\n")
    logging.debug(f"Request URL:\n{request.url}")
    logging.debug(f"Authorization Header:\n{authorization_header}")
    logging.debug(f"x-amz-date:\n{amz_date}")

    auth_parts = authorization_header.split(", ")
    credential_part = auth_parts[0]
    signed_headers_part = auth_parts[1]
    signature_part = auth_parts[2]

    credential_scope = (
        credential_part.split(" ")[1].split("Credential=")[1].split("/")[1:]
    )
    credential_scope = "/".join(credential_scope)

    signed_headers = signed_headers_part.split("SignedHeaders=")[-1].split(";")
    provided_signature = signature_part.split("Signature=")[-1]

    logging.debug(f"Signed Headers:\n{signed_headers}")
    logging.debug(f"Credential Scope:\n{credential_scope}")

    headers_dict = {k.lower(): v for k, v in request.headers.items()}
    sorted_headers = {
        k: headers_dict[k] for k in sorted(headers_dict) if k in signed_headers
    }

    logging.debug(f"Headers Dict:\n{headers_dict}")
    logging.debug(f"Sorted Headers:\n{sorted_headers}")

    canonical_uri = request.url.path
    canonical_querystring = request.url.query
    
    # 修正Payload哈希获取逻辑
    content_sha256 = headers_dict.get("x-amz-content-sha256")
    if content_sha256 == "UNSIGNED-PAYLOAD":
        payload_hash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
    elif content_sha256:
        payload_hash = content_sha256
    else:
        payload_hash = hashlib.sha256(await request.body()).hexdigest()

    # 移除对已签名头的修改
    canonical_headers = "".join([f"{k}:{v}\n" for k, v in sorted_headers.items()])

    canonical_request = "\n".join(
        [
            request.method,
            canonical_uri,
            canonical_querystring,
            canonical_headers,
            ";".join(signed_headers),
            payload_hash,
        ]
    )

    logging.debug(f"Canonical Request:\n{canonical_request}")

    string_to_sign = "\n".join(
        [
            ALGORITHM,
            amz_date,
            credential_scope,
            hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(),
        ]
    )

    logging.debug(f"String to Sign:\n{string_to_sign}")

    date_stamp = credential_scope.split("/")[0]
    signing_key = get_signature_key(
        AWS_SECRET_ACCESS_KEY,
        date_stamp,
        AWS_REGION,
        SERVICE,
    )

    signature = hmac.new(
        signing_key, string_to_sign.encode("utf-8"), hashlib.sha256
    ).hexdigest()

    logging.debug(f"Calculated Signature: {signature}")
    logging.debug(f"Provided Signature: {provided_signature}")

    is_valid = provided_signature == signature
    if not is_valid:
        logging.error("Signatures do not match")
    return is_valid

  except Exception as e:
    logging.error(f"Verification failed: {e}")
    return False

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

FastAPI+Python实现S3服务：PUT请求签名验证失败求助

签名验证代码 #

测试代码 #

1. 修正Payload哈希获取逻辑 #

2. 禁止修改已签名的请求头值 #

3. 确保Canonical Headers与Signed Headers顺序一致 #