Webhooks是第三方实时通知客户事件发生的常用方法。FastAPI是 Python 最好的 API 框架之一,添加对 webhook 的支持既快速又简单。

webhook 的常用签名算法是 SHA256 HMAC 签名,但很容易编写一个常用函数来支持 md5、sha1 和 sha256。

import hmac
from hashlib import md5, sha1, sha256
from functools import partial


def compare_signature(digestmod, key: bytes, message: bytes, expected_signature: str) -> bool:
    mac = hmac.new(key, message, digestmod)
    return hmac.compare_digest(mac.hexdigest(), expected_signature)


compare_md5_signature = partial(compare_signature, md5)
compare_sha1_signature = partial(compare_signature, sha1)
compare_sha256_signature = partial(compare_signature, sha256)

最重要的部分是确保使用hmac.compare_digest函数而不是简单的字符串比较(==),这可以防止定时攻击。

from fastapi import Request, Header, FastAPI, Depends, HTTPException

app = FastAPI()

# 32 bytes or more for the best security
WEBHOOK_SECRET = b"443c9cec-9dd6-11e9-b610-3c15c2eb7774"


async def _validate_webhook_signature(request: Request):
    webhook_signature = request.headers.get("webhook-signature")
    if not webhook_signature:
        raise HTTPException(400, "Missing webhook signature header")

    request_body = await request.body()
    # CHANGE: compare_sha256_signature for the signature your integration uses
    if not compare_sha256_signature(WEBHOOK_SECRET, request_body, webhook_signature):
        raise HTTPException(401, "Invalid webhook signature")


@app.post("/my-webhook", dependencies=[Depends(_validate_webhook_signature)])
def my_webhook():
    # ... process data
    return {}

创建_validate_webhook_signature函数后,可以在以后定义的任何 webhook 路由上使用它。 Webhook 签名的总体思路是确保向您的 Webhook 发送数据的任何第三方服务都是他们自称的身份,而不是恶意行为者。

要记住的最重要的概念是 webhook 签名可以替代您的正常应用程序身份验证,但不能替代授权。可以在身份验证页面上的附录 I中了解更多信息。