Django 接收 Square 支付 Webhook 时返回 401

2026-11-04排障赵师傅

结论

这个 401 Unauthorized 很可能来自应用自身的签名校验。Square Webhook 不需要额外的登录认证。

当前实现有两处签名校验错误：

请求头名称不对。Square Webhook 签名通常位于 x-square-hmacsha256-signature，代码读取的却是 X-Square-Signature，很可能只能得到空字符串。
参与签名的内容不完整。Square 使用的是：

notification_url + raw_request_body

当前代码只对 request_body 计算 HMAC，因此无法匹配 Square 发来的签名。

Square 不会携带 Django 用户凭证、JWT 或 Basic Auth。Webhook URL 应允许匿名访问，并通过签名确认请求是否来自 Square。

正确的签名计算方式

签名应按下面的方式计算：

Base64(HMAC-SHA256(signature_key, notification_url + raw_request_body))

这里需要注意：

signature_key 是对应 Webhook subscription 的 Signature Key，不是 Square Access Token。
notification_url 必须与 Square Developer Dashboard 中配置的 URL 完全一致。
raw_request_body 必须使用未经重新序列化的原始请求体。
签名请求头是 x-square-hmacsha256-signature。

路径末尾的 / 也会参与计算，所以下面两个地址会生成不同的签名：

https://api.mydomain.com/api/wallet/webhooks/square
https://api.mydomain.com/api/wallet/webhooks/square/

推荐实现

Square SDK 43.x 提供了 Webhook 签名验证工具。具体导入路径要以当前安装包为准，该版本通常可以使用 square.utils.webhooks_helper.verify_signature。

import json

from django.conf import settings
from django.utils.decorators import method_decorator
from django.views.decorators.csrf import csrf_exempt

from rest_framework import status
from rest_framework.permissions import AllowAny
from rest_framework.response import Response
from rest_framework.views import APIView

from square.utils.webhooks_helper import verify_signature


SQUARE_WEBHOOK_NOTIFICATION_URL = (
    "https://api.mydomain.com/api/wallet/webhooks/square/"
)


@method_decorator(csrf_exempt, name="dispatch")
class SquareWebhookView(APIView):
    authentication_classes = []
    permission_classes = [AllowAny]

    def post(self, request):
        signature = request.headers.get(
            "x-square-hmacsha256-signature",
            "",
        )

        raw_body = request.body.decode("utf-8")

        is_valid = verify_signature(
            request_body=raw_body,
            signature_header=signature,
            signature_key=settings.SQUARE_WEBHOOK_SIGNATURE_KEY,
            notification_url=SQUARE_WEBHOOK_NOTIFICATION_URL,
        )

        if not is_valid:
            return Response(
                {"error": "Invalid webhook signature"},
                status=status.HTTP_401_UNAUTHORIZED,
            )

        try:
            event = json.loads(raw_body)
        except json.JSONDecodeError:
            return Response(
                {"error": "Invalid JSON"},
                status=status.HTTP_400_BAD_REQUEST,
            )

        event_type = event.get("type")
        payment = (
            event.get("data", {})
            .get("object", {})
            .get("payment", {})
        )

        if (
            event_type == "payment.updated"
            and payment.get("status") == "COMPLETED"
        ):
            payment_id = payment.get("id")

            if not payment_id:
                return Response(
                    {"error": "Payment ID is missing"},
                    status=status.HTTP_400_BAD_REQUEST,
                )

            process_completed_square_payment(payment_id)

        return Response(status=status.HTTP_200_OK)

process_completed_square_payment() 应该是一个普通业务函数。不要在 Webhook view 中手工构造 DRF Request，然后直接调用另一个 view 的 post()。

例如：

from decimal import Decimal

from django.db import transaction

from square import Square
from square.environment import SquareEnvironment

from users.models import CustomUser
from wallet.models import Wallet


@transaction.atomic
def process_completed_square_payment(payment_id):
    client = Square(
        token=settings.SQUARE_ACCESS_TOKEN,
        environment=SquareEnvironment.PRODUCTION,
    )

    payment_response = client.payments.get_payment(
        payment_id=payment_id,
    )

    if payment_response.errors:
        raise ValueError(
            f"Unable to retrieve Square payment: "
            f"{payment_response.errors}"
        )

    payment = payment_response.payment

    if payment.status != "COMPLETED":
        return

    order_id = payment.order_id
    if not order_id:
        raise ValueError("Square payment does not contain an order ID")

    order_response = client.orders.retrieve_order(
        order_id=order_id,
        location_id=settings.SQUARE_LOCATION_ID,
    )

    if order_response.errors:
        raise ValueError(
            f"Unable to retrieve Square order: "
            f"{order_response.errors}"
        )

    metadata = order_response.order.metadata or {}

    user_id = metadata.get("user_id")
    amount_value = metadata.get("amount")
    plan_value = metadata.get("plan")

    if not all((user_id, amount_value, plan_value)):
        raise ValueError("Required order metadata is missing")

    user = CustomUser.objects.get(id=user_id)
    wallet, _ = Wallet.objects.get_or_create(user=user)

    amount = Decimal(amount_value)
    plan = int(plan_value)

    # 在这里创建 Transaction、Subscription 等业务记录。
    # 应使用 payment_id 或 event_id 做唯一约束，防止重复处理。

不使用 SDK 工具时的写法

如果当前安装的 SDK 中没有上述导入路径，也可以按照同样的规则自行验证：

import base64
import hashlib
import hmac


def is_valid_square_signature(
    signature,
    request_body,
    signature_key,
    notification_url,
):
    message = f"{notification_url}{request_body}"

    computed_signature = base64.b64encode(
        hmac.new(
            signature_key.encode("utf-8"),
            message.encode("utf-8"),
            hashlib.sha256,
        ).digest()
    ).decode("utf-8")

    return hmac.compare_digest(signature, computed_signature)

调用时必须传入固定的公开通知地址：

signature = request.headers.get(
    "x-square-hmacsha256-signature",
    "",
)

raw_body = request.body.decode("utf-8")

if not is_valid_square_signature(
    signature=signature,
    request_body=raw_body,
    signature_key=settings.SQUARE_WEBHOOK_SIGNATURE_KEY,
    notification_url=SQUARE_WEBHOOK_NOTIFICATION_URL,
):
    return Response(
        {"error": "Invalid webhook signature"},
        status=401,
    )

不要先执行 json.loads()，再用 json.dumps() 生成待签名内容。重新序列化可能改变空格、字段顺序或字符转义，导致签名不匹配。

还需要检查的配置

Webhook URL 必须完全一致

可以把 Square Dashboard 中配置的地址直接保存为配置项：

SQUARE_WEBHOOK_NOTIFICATION_URL = (
    "https://api.mydomain.com/api/wallet/webhooks/square/"
)

不要依赖 request.build_absolute_uri() 推断签名地址。应用部署在 Nginx、负载均衡器或 CDN 后方时，Django 看到的协议和主机名可能不同于公网 URL。

同时检查 URL 是否发生重定向，包括：

HTTP 跳转到 HTTPS；
无 / 的地址跳转到带 / 的地址；
主域名跳转到另一个域名；
CDN 或网关改写路径。

Square 中配置的 URL 最好直接命中最终 endpoint。

Signature Key 必须属于当前 Webhook subscription

开发环境和生产环境的配置不能混用。请逐项确认：

Webhook subscription 来自正确的 Square application；
使用的是该 subscription 页面显示的 Signature Key；
事件发往生产 URL 时，后端加载的是生产配置；
环境变量中没有多余的引号、空格或换行。

Django endpoint 应允许匿名访问

建议明确配置：

authentication_classes = []
permission_classes = [AllowAny]

permission_classes = [] 通常也不会要求权限，但显式使用 AllowAny 更容易理解。设置 authentication_classes = [] 还可以避免项目级 DRF authentication class 对 Webhook 请求执行不必要的认证。

csrf_exempt 只会绕过 Django CSRF 校验，不能代替 Square 的签名验证。

确认 401 来自哪一层

如果响应体是：

{"error": "Invalid signature"}

这个 401 就是 SquareWebhookView 返回的。

如果请求没有进入 view，则要检查以下位置：

Nginx、Apache、CDN 或 API Gateway 是否启用了访问认证；
URL 是否被其他 Django route 匹配；
中间件是否要求 JWT、API Key 或登录状态；
WAF 是否拦截了 Square 请求；
域名或路径级别是否配置了 Basic Auth。

Webhook 处理还要考虑重复投递

签名问题解决后，Square 仍可能重复投递同一个事件。创建订阅、交易或余额记录时，应使用 event_id、payment_id 或 order_id 做幂等控制。

常见的处理方式是为支付记录的 square_payment_id 添加唯一约束：

square_payment_id = models.CharField(
    max_length=255,
    unique=True,
)

处理前也可以查询记录是否已经存在：

if Transaction.objects.filter(
    square_payment_id=payment_id
).exists():
    return

数据库唯一约束仍需保留。并发投递时，仅靠先查询再创建可能产生竞态条件。

当前需要修正的是 Square 签名请求头名称，以及签名内容中遗漏的 Webhook notification URL。URL 和 Signature Key 也必须与 Dashboard 中的配置完全一致。