Django序列化器如何优雅解析App Store Server Notifications V2的JWS数据

1. 安装依赖 #

你需要提前安装JWS处理依赖：

pip install python-jose[cryptography]

2. 实现通用JWS解码工具 #

先写一个通用的Apple JWS解码函数，统一处理签名验证和解码逻辑，支持缓存Apple公钥避免重复请求：

from jose import jwt, jws
from jose.exceptions import JWSVerificationError
import requests

# 缓存Apple公钥，有效期可自行设置为1天
APPLE_PUBLIC_KEYS = None
APPLE_PUBLIC_KEY_URL = "https://appleid.apple.com/auth/keys"

def get_apple_public_keys():
    global APPLE_PUBLIC_KEYS
    if not APPLE_PUBLIC_KEYS:
        resp = requests.get(APPLE_PUBLIC_KEY_URL)
        resp.raise_for_status()
        APPLE_PUBLIC_KEYS = resp.json()["keys"]
    return APPLE_PUBLIC_KEYS

def decode_appstore_jws(jws_str: str) -> dict:
    # 获取JWS头中的kid，匹配对应Apple公钥
    header = jws.get_unverified_header(jws_str)
    kid = header["kid"]
    public_keys = get_apple_public_keys()
    target_key = next((k for k in public_keys if k["kid"] == kid), None)
    if not target_key:
        raise ValueError(f"未找到匹配的Apple公钥，kid: {kid}")
    # 验证签名并解码payload
    try:
        payload = jwt.decode(
            jws_str, 
            target_key, 
            algorithms=["ES256"], 
            options={"verify_aud": False}
        )
        return payload
    except JWSVerificationError:
        raise ValueError("JWS签名验证失败，请求可能为伪造")

3. 自定义DRF JWS序列化字段 #

扩展DRF的基础字段，实现自动解码JWS串的能力，无需在业务逻辑中手动处理：

from rest_framework import serializers
from .utils import decode_appstore_jws

class JWSDecoderField(serializers.Field):
    def to_internal_value(self, data):
        if not isinstance(data, str):
            raise serializers.ValidationError("JWS字段必须为字符串格式")
        try:
            return decode_appstore_jws(data)
        except Exception as e:
            raise serializers.ValidationError(f"JWS解码失败: {str(e)}")

4. 编写分层序列化器 #

按照解码后的结构化数据层级，编写对应的序列化器，嵌套的JWS字段直接使用上面自定义的JWSDecoderField即可：

# 续购信息解码后的序列化器
class RenewalInfoSerializer(serializers.Serializer):
    autoRenewProductId = serializers.CharField()
    autoRenewStatus = serializers.IntegerField()
    expirationIntent = serializers.IntegerField()
    gracePeriodExpiresDate = serializers.IntegerField()
    isInBillingRetryPeriod = serializers.BooleanField()
    # 其他字段按需补充

# 交易信息解码后的序列化器
class TransactionInfoSerializer(serializers.Serializer):
    appAccountToken = serializers.CharField()
    bundleId = serializers.CharField()
    expiresDate = serializers.IntegerField()
    inAppOwnershipType = serializers.CharField()
    isUpgraded = serializers.BooleanField()
    # 其他字段按需补充

# 通知data部分序列化器
class NotificationDataSerializer(serializers.Serializer):
    appAppleId = serializers.IntegerField()
    bundleId = serializers.CharField()
    bundleVersion = serializers.CharField()
    environment = serializers.CharField()
    # 嵌套JWS字段直接使用自定义字段，自动完成解码
    signedRenewalInfo = JWSDecoderField(child=RenewalInfoSerializer())
    signedTransactionInfo = JWSDecoderField(child=TransactionInfoSerializer())

# 最外层通知序列化器
class AppStoreNotificationSerializer(serializers.Serializer):
    notificationType = serializers.CharField()
    subtype = serializers.CharField(required=False, allow_null=True)
    notificationUUID = serializers.UUIDField()
    data = NotificationDataSerializer()

# 如果需要直接存入数据库，改为继承ModelSerializer，关联对应模型即可

5. 视图层实现 #

完全遵循DRF的常规开发流程，几乎不需要额外改造：

from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from .serializers import AppStoreNotificationSerializer
from .utils import decode_appstore_jws
from .models import AppStoreNotification

class AppStoreNotificationView(APIView):
    # App Store官方通知不需要鉴权，按需关闭即可
    authentication_classes = []
    permission_classes = []

    def post(self, request):
        # App Store请求的body是raw JWS串，Content-Type为application/jwt
        raw_jws = request.body.decode()
        # 先解码最外层JWS
        try:
            outer_payload = decode_appstore_jws(raw_jws)
        except Exception as e:
            return Response({"error": str(e)}, status=status.HTTP_400_BAD_REQUEST)
        # 常规序列化校验
        serializer = AppStoreNotificationSerializer(data=outer_payload)
        if not serializer.is_valid():
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
        # 直接保存到数据库即可，所有嵌套字段已经完成解码
        AppStoreNotification.objects.create(**serializer.validated_data)
        # 必须返回200状态码，否则App Store会重复推送通知
        return Response(status=status.HTTP_200_OK)

注意事项 #

• 生产环境必须开启签名验证，禁止使用无验证的解码方式，避免恶意伪造请求造成业务损失
• Apple公钥可设置缓存时间（建议1天），减少不必要的外部请求，提升接口响应速度
• 字段可根据Apple官方文档按需补充，不需要的字段无需在序列化器中声明，会自动忽略

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