Python实现苹果支付自动订阅服务端全流程详解

做服务端开发的朋友迟早会撞上一个需求：iOS 端接入苹果的内购自动续订订阅，然后服务端要负责验证订单、记录订阅状态、处理续费通知、同步用户权益。标题里说的"python苹果支付(服务端）自动订阅版"，其实就是苹果App Store的Auto-Renewable Subscription整套服务端逻辑。很多刚接触这个领域的同学以为写个接口接收一下客户端传来的receipt，调一下苹果的验单接口就完事了，真做起来才发现事情远比想象中多——通知要验证、票据要解密、退款要同步、沙箱和生产环境还要来回切换。这篇文章我把整套服务端流程拆开讲，从账号准备、receipt验证、通知处理到状态计算，全部用Python代码演示一遍，适合已经跑通客户端逻辑、正准备补服务端闭环的团队参考。

1. 自动订阅的服务端逻辑到底在哪里

苹果的自动续订订阅和普通的一次性内购最大的区别在于：支付并不是只发生一次。用户首发购买之后，后续的每一期续费都由苹果自动扣款，扣款成功后你的APP和服务端都会收到通知。这意味着你必须有一套持续运行的机制来处理这些持续发生的状态变化。

1.1 客户端传上来的Receipt到底是什么

用户在iOS上完成订阅后，客户端会拿到一个base64编码的transactionReceipt字符串。这串东西在服务端眼里就是一个"凭证"，你可以把它发给苹果的验证接口，换取一份结构化的订单数据。

但是要注意，这个receipt并不是只包含当前这一笔交易，它是一个票据集合——里面可能包含该用户在这台设备上的所有历史交易记录，包括已经过期的、退款的、换购的。苹果返回的验证结果中，receipt.in_app数组和latest_receipt_info数组会给出交易明细，其中latest_receipt_info通常按时间倒序排列，第一条往往就是最新的一笔订阅交易。

1.2 订阅状态为什么不能只靠客户端判断

我见过不少项目最初只在客户端判断expires_date是否大于当前时间，以此决定用户是否有会员权益。这个做法在沙箱测试时看着没问题，上线后就会出现两种情况：

• 用户关闭了自动续订，但客户端缓存里还是"有效期到明年"，导致权益迟迟不消失。
• 用户换设备登录，新设备上没有任何本地票据，客户端根本不知道对方是不是有效订阅者。

正确做法是把订阅状态的判断权放到服务端。客户端只负责上传receipt和过期时间作为参考，服务端以苹果验单结果和通知消息为准，统一计算并存储用户的当前订阅状态。这样不管是换设备、重装APP还是关掉续费，用户的权益判定都走同一套逻辑。

2. 接入前的账号与密钥准备

苹果的服务端接入有两种方式：一种是拿账号密码换令牌去调用App Store Connect API，另一种是直接用密钥做JWT签名。自动续订相关服务端开发中，最常用的是App Store Server Library配合App Store Server API和App Store Server Notifications V2，这套体系的鉴权方式是密钥签名。

2.1 密钥、密钥ID和Issuer ID从哪里拿

登录App Store Connect，进入"用户和访问" -> "集成" -> "App Store Connect API"，这里可以生成一个API密钥。生成时会得到一个.p8文件、一个Key ID和一个Issuer ID，这三个东西是后面服务端所有请求的凭证基础。

其中.p8文件内容形如-----BEGIN PRIVATE KEY-----开头的一段椭圆曲线私钥，苹果用的是ES256签名算法。Key ID和Issuer ID是两串固定字符串，跟某一个开发者账号绑定。这三样信息建议放在环境变量或配置中心，不要硬编码进代码仓库。

提示：.p8密钥文件只能下载一次，弄丢了只能重新生成。生成后记得按环境隔离，生产环境和测试环境建议用不同的密钥，避免权限边界模糊。

2.2 沙箱环境与生产环境的切换逻辑

苹果验单接口有生产线和沙箱线，前者是buy.itunes.apple.com，后者是sandbox.itunes.apple.com。客户端在TestFlight和Xcode调试时拿到的receipt，只能发到沙箱线去验证，正式上架后App Store下载的包发的是生产线的请求。

很多团队在服务端实现时会写成：先请求生产环境，如果返回21007（沙箱receipt被发到生产），就再请求沙箱环境；反过来如果生产receipt发到沙箱，会返回21008。这套逻辑能跑通，但要注意一个细节：必须先生产后沙箱，因为生产环境下如果遇到21007说明这是沙箱票据，此时降级到沙箱请求是正常流程；如果反过来先沙箱后生产，线上receipt可能会在沙箱接口被判定为非法。

我这里列一个常用的状态码参照表，验单接口直接按这组数字判断：

3. 用Python实现对接收据验证

验单是整个订阅系统的地基。这一步做扎实了，后面解析订单、计算权益、记录流水都会顺畅很多。

3.1 构建验单请求与解析响应体

Python这边用requests就够了。核心函数逻辑是按照上面说过的先生产后沙箱策略去调用苹果接口。

import requests import base64 import json import time import logging REQUEST_TIMEOUT = 30 MAX_RETRY = 3 PRODUCTION_URL = "https://buy.itunes.apple.com/verifyReceipt" SANDBOX_URL = "https://sandbox.itunes.apple.com/verifyReceipt" # 这个从App Store Connect后台获取 SHARED_SECRET = "your_shared_secret" logger = logging.getLogger(__name__) def verify_receipt(receipt_str: str, is_sandbox: bool = False) -> dict: """ :param receipt_str: 客户端上传的base64 receipt字符串 :param is_sandbox: 是否强制走沙箱环境 :return: 苹果完整响应字典 """ payload = { "receipt-data": receipt_str, "password": SHARED_SECRET, "exclude-old-transactions": False } url = SANDBOX_URL if is_sandbox else PRODUCTION_URL for attempt in range(MAX_RETRY): try: resp = requests.post(url, json=payload, timeout=REQUEST_TIMEOUT) data = resp.json() status = data.get("status") if status == 21007 and not is_sandbox: # 沙箱票据打到了生产，切换沙箱重新验证 logger.info("receipt is sandbox, retry on sandbox environment") return verify_receipt(receipt_str, is_sandbox=True) if status == 21008 and is_sandbox: # 生产票据打到了沙箱，切回生产重试 logger.warning("production receipt sent to sandbox, switch to production") return verify_receipt(receipt_str, is_sandbox=False) if status == 0: return data # 21005 属于服务器暂时繁忙，可以做退避重试 if status in (21005,) and attempt < MAX_RETRY - 1: time.sleep(2 ** attempt) continue logger.error(f"verify_receipt failed, status={status}") return data except (requests.RequestException, json.JSONDecodeError) as exc: logger.warning(f"network error on attempt {attempt + 1}: {exc}") if attempt < MAX_RETRY - 1: time.sleep(2 ** attempt) else: raise return {}

这里有几个细节容易忽略。第一，exclude-old-transactions字段，如果你只想拿当前订阅周期的最新数据，可以考虑设为true，但如果你需要历史退款记录或者想对比整条订阅链，就设成false。第二，请求超时一定要设置，苹果接口偶尔会慢，不设超时的服务端代码在生产环境容易被拖死。

3.2 把响应解析成我们关心的订阅状态

苹果验单成功返回的JSON结构大致长这样：

{ "status": 0, "environment": "Production", "receipt": { "receipt_type": "Production", "bundle_id": "com.example.app", "in_app": [] }, "latest_receipt_info": [ { "product_id": "com.example.sub.monthly", "transaction_id": "1000000123456789", "original_transaction_id": "1000000123456788", "purchase_date": "2024-01-01 00:00:00 Etc/GMT", "expires_date": "2024-02-01 00:00:00 Etc/GMT", "is_trial_period": "false", "cancellation_date": "", "cancellation_reason": "" } ], "pending_renewal_info": [ { "product_id": "com.example.sub.monthly", "auto_renew_status": "1", "expiration_intent": "1" } ] }

对服务端来说，最核心的字段是latest_receipt_info里的expires_date和pending_renewal_info里的auto_renew_status。expires_date是当前订阅周期的到期时间，auto_renew_status表示用户是否开启了自动续订。

把这段逻辑封装成一个解析函数，方便上层业务直接拿状态：

from datetime import datetime, timezone from zoneinfo import ZoneInfo def parse_subscription_info(verify_data: dict, product_id: str) -> dict: """ 从苹果验单响应中提炼出订阅状态 :return: 包含 expires_date, auto_renew_status, in_trial 等字段 """ latest_info = verify_data.get("latest_receipt_info", []) target_trx = None # latest_receipt_info 可能包含多个订阅产品，按 product_id 过滤 for item in latest_info: if item.get("product_id") == product_id: target_trx = item break if not target_trx: return {"active": False, "reason": "no_subscription_found"} expires_str = target_trx.get("expires_date") or "" cancellation_date = target_trx.get("cancellation_date") or "" # 苹果返回的时间有时带毫秒，格式形如 2024-01-01 00:00:00 Etc/GMT try: expires_dt = datetime.strptime(expires_str, "%Y-%m-%d %H:%M:%S Etc/GMT") expires_dt = expires_dt.replace(tzinfo=timezone.utc) except ValueError: # 部分老版本接口可能带毫秒 expires_dt = datetime.fromisoformat(expires_str.replace("Z", "+00:00")) # 退款状态：苹果会写入 cancellation_date is_cancelled = bool(cancellation_date) # 是否处于试用期 is_trial = target_trx.get("is_trial_period", "false") == "true" # 查询 pending_renewal_info 中的自动续订状态 auto_renew = False for pending in verify_data.get("pending_renewal_info", []): if pending.get("product_id") == product_id and pending.get("auto_renew_status") == "1": auto_renew = True now = datetime.now(timezone.utc) active = expires_dt > now and not is_cancelled return { "active": active, "expires_date": expires_dt.isoformat(), "auto_renew_status": auto_renew, "is_trial": is_trial, "is_cancelled": is_cancelled, "transaction_id": target_trx.get("transaction_id"), "original_transaction_id": target_trx.get("original_transaction_id"), "product_id": product_id, }

注意，expires_date的时区解析是很多人踩坑的地方。苹果返回的时间字符串是"Etc/GMT"格式，部分版本还可能出现类似"2024-01-01 00:00:00 America/Los_Angeles"的情况，服务端统一存成ISO格式的UTC时间最稳妥。

提示：判断"用户是否还有效"不能只看expires_date这一个字段。退款、争议、购买被撤销等场景下，expires_date可能仍然在未来，但revenue已经没了。是否将退款用户移除权益，需要业务上重新审视，至少服务端要把cancellation_date是否存在当作一个独立信号存下来。

4. 接入App Store Server Notifications V2

验单是"用户主动上报"的场景，但自动续订服务端不能只依赖用户主动上报——用户续费成功、续费失败、退款、关闭续订，这些事件苹果都必须主动通知给你的服务器。这就是App Store Server Notifications V2，简称通知V2。

4.1 通知类型与子类型怎么理解

通知V2的机制是：苹果会给你配置的回调URL发送一个JWS格式的POST请求。JWS负载里包含signedPayload字段，里面是一串签名数据。

通知类型很多，服务端最关心的几个：

看到这个表就知道，通知处理是整个订阅系统的"事件总线"，比验单接口重要得多。客户端可以不主动上报，但通知一个都不能少处理。

4.2 签名验签与解密实现

通知V2的signedPayload是一个JWS格式的字符串，它的header部分会带上x5c，也就是Apple的证书链。服务端需要拿苹果根证书去验签。

Python这边我建议用jwt和cryptography两个库配合。先取出x5c证书链，然后通过cryptography从PEM证书中提取公钥，最后用jwt.decode校验签名。

import base64 import json import jwt import requests from cryptography import x509 from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives import serialization # Apple Root CA - G3 证书，建议存为pem文件，这里仅演示加载方式 APPLE_ROOT_CA_PATH = "config/AppleRootCA-G3.pem" # 从苹果下载的根证书 def load_root_ca(path: str): with open(path, "rb") as f: return x509.load_pem_x509_certificate(f.read()) def extract_leaf_certificate(signed_payload: str): """ 从JWS header的x5c字段中解出叶子证书 """ header_part = signed_payload.split(".")[0] # JWS header是base64url编码 header_bytes = base64.urlsafe_b64decode(header_part + "==") header = json.loads(header_bytes) x5c_chain = header.get("x5c", []) if not x5c_chain: raise ValueError("no x5c cert chain found") # x5c 的第一项就是叶子证书 leaf_der = base64.b64decode(x5c_chain[0]) leaf_cert = x509.load_der_x509_certificate(leaf_der) return leaf_cert def verify_signed_payload(signed_payload: str) -> dict: """ 验证通知V2的JWS并返回解析后的data """ root_ca = load_root_ca(APPLE_ROOT_CA_PATH) leaf_cert = extract_leaf_certificate(signed_payload) # 验链：从叶子证书往根证书方向验证 public_key = leaf_cert.public_key() header_part = signed_payload.split(".")[0] header_bytes = base64.urlsafe_b64decode(header_part + "==") header = json.loads(header_bytes) algorithm = header.get("alg", "ES256") # 使用ec公钥来decode decoded = jwt.decode( signed_payload, public_key, algorithms=[algorithm], options={"verify_aud": False}, ) return decoded

这里我要特别提醒：通知V2和早期的通知V1完全不同。V1用的是明文JSON，随便一个POST都能伪造；V2是JWS签名包裹，必须验签，否则随便一个懂HTTP的人就能给你的服务器发一条"用户已续费一年"的假通知。

验签通过后，decoded里实际上是一个{ "notificationType": "...", "subtype": "...", "data": {...}, "signedDate": ... }的结构。

4.3 通知处理必须幂等

通知V2有个特性：苹果可能重复发送同一条通知。不管是网络重试还是苹果内部重放，如果你的处理逻辑里"给用户加200天会员"会重复执行，那你就亏大了。

所以通知处理的第一步不是立刻改数据库，而是先查transaction_id+notification_type是否已经处理过。一般方案是在订阅订单表上建一个unique constraint，处理前先查询或插入。

from datetime import datetime def process_notification(payload: dict, order_repo, event_repo): """ payload 是验签之后解析出来的通知数据 order_repo: 订单/订阅表仓储 event_repo: 事件流水表仓储 """ notification_type = payload.get("notificationType") subtype = payload.get("subtype", "") data = payload.get("data", {}) signed_date = payload.get("signedDate") transaction_id = data.get("transactionId") original_transaction_id = data.get("originalTransactionId") product_id = data.get("productId") if not transaction_id: return {"code": "IGNORED", "message": "missing transaction id"} # 幂等控制：同一transaction + 同一通知类型只能处理一次 event_key = f"{transaction_id}_{notification_type}_{subtype}" if event_repo.exists(event_key): return {"code": "DUPLICATED", "message": "event already processed"} # 找出这笔交易对应的本地订单 order = order_repo.find_by_original_transaction_id(original_transaction_id) if not order: # 可能是首次订阅时通知先到，落本地订单前要兜底 order = order_repo.create_from_notification(data) if notification_type == "SUBSCRIBED": order_repo.activate_subscription(order, data) elif notification_type == "DID_RENEW": order_repo.extend_subscription(order, data) elif notification_type == "DID_FAIL_TO_RENEW": order_repo.mark_retry_schedule(order, data) elif notification_type == "DID_EXPIRE": order_repo.expire_subscription(order) elif notification_type == "CANCEL": order_repo.revoke_subscription(order) elif notification_type == "REFUND": order_repo.revoke_refund(order, data) # 记录处理流水，保证幂等 event_repo.save(event_key, payload) return {"code": "OK"}

这段代码兼顾了三个重要设计：幂等、兜底创建、状态流转集中管理。实际项目中，通知处理逻辑会越加越多，建议一开始就把状态流转收敛到独立的service文件里，不要散落在各个视图函数中。

5. 订阅状态持久化与对账思路

通知处理完，用户的关键状态最后都要落到数据库。我见过不少团队把订阅到期时间直接存在用户表的expires_at字段里，简单粗暴，但过一阵子就会出问题——你不知道这个时间是从哪笔交易来的，也不知道某次退款对应的是哪一段权益。

5.1 建议的订单表与对账字段设计

推荐至少维护一张订阅订单表，核心字段：

对账的关键是original_transaction_id和transaction_id的区分。original_transaction_id从首次购买起保持不变，transaction_id每期续费都可能变化。服务端做用户订阅链查询时，永远用original_transaction_id去关联，否则续费通知来了会定位不到历史订阅。

5.2 定期拉取订单状态的兜底方案

只依赖通知还不够，总会有通知丢失、网络故障、服务宕机的情况。所以定期从苹果拉一次订单状态是很有必要的。苹果的App Store Server API提供GET /inAppPurchases/subscriptions/{transactionId}端点，需要带上JWT签名请求。

import requests import jwt import time from datetime import datetime, timedelta, timezone ISSUER_ID = "your_issuer_id" KEY_ID = "your_key_id" BUNDLE_ID = "com.example.app" PRIVATE_KEY_PATH = "config/SubscriptionKey.p8" APP_STORE_API_BASE = "https://api.storekit.itunes.apple.com" def generate_token(): with open(PRIVATE_KEY_PATH, "r") as f: private_key = f.read() now = datetime.now(tz=timezone.utc) headers = { "alg": "ES256", "kid": KEY_ID, "typ": "JWT", } payload = { "iss": ISSUER_ID, "iat": int(now.timestamp()), "exp": int((now + timedelta(minutes=5)).timestamp()), "aud": "appstoreconnect-v1", "bid": BUNDLE_ID, } token = jwt.encode(payload, private_key, algorithm="ES256", headers=headers) return token def fetch_subscription_status(transaction_id: str): token = generate_token() headers = { "Authorization": f"Bearer {token}", } try: resp = requests.get( f"{APP_STORE_API_BASE}/inAppPurchases/subscriptions/{transaction_id}", headers=headers, timeout=10, ) return resp.json() except requests.RequestException as exc: return {"error": str(exc)}

这个兜底拉取建议每天跑一次，或者用户在触发关键业务动作前主动刷新一次。频率不需要太高，因为苹果的Server API也有限流，而且你不是拿它做实时校验，只是兜底纠偏。

6. 实战中最容易翻车的五个细节

这部分不是从文档抄的，是我在实际接入中踩过的坑和帮别人排查过的案例，按翻车频率从高到低整理出来。

6.1 时间戳时区与过期判定

苹果接口返回的时间有两种格式，老版本验单接口常见2024-01-01 00:00:00 Etc/GMT，通知V2里是毫秒级Unix时间戳。如果代码里直接拿字符串比大小，或者按本地时区解析，订阅到期时间就会偏移8小时甚至更多。

统一处理方式：解析成UTC存入数据库，所有比较都用datetime.now(timezone.utc)，前端展示时再转本地时区。

6.2 首次订阅时通知先于客户端上报

有些情况下苹果的SUBSCRIBED通知在客户端调起验单接口之前就到了服务端。这时本地还没有这笔订阅订单，所以通知处理的"兜底创建"逻辑必须提前写好。很多同学只写了更新订单的逻辑，结果首条通知直接return None，用户的权益直到客户端上报才开通，一旦客户端上报失败，用户就白付款了。

6.3 退款后的用户权益回收延迟

苹果的退款通知到达服务端通常会有一定延迟，极端情况可能拖到24小时。这意味着用户在苹果退款弹窗确认后，APP里的权益可能还会持续一段时间。这不是你服务端代码的问题，是苹果通知机制的特性。但你可以做两件事：

• 配合CANCEL+REFUND通知及时回收。
• 在客户端做一点"软控制"，比如每次启动时对比本地记录和用户手动刷新后的服务端状态。

6.4 宽限期与账单重试

DID_FAIL_TO_RENEW不代表订阅已经结束。苹果会尝试从用户的支付方式里再次扣款，这个阶段订阅仍然有效。如果服务端收到DID_FAIL_TO_RENEW就直接把用户降级，用户后面扣款成功又得恢复，这一来一回体验极差。

正确的做法是把DID_FAIL_TO_RENEW标记为"待重试"，到期时间仍以expires_date为准，等DID_EXPIRE或者GRACE_PERIOD_EXPIRED到了再回收权益。

6.5 验单接口被滥用

验单接口就是服务端暴露给客户端的一个普通HTTP接口。如果不加鉴权和频率限制，任何人都可以伪造一个receipt-data字符串刷你的接口，虽然伪造的票据会验证失败，但大量无效请求会白白消耗你的服务器资源和苹果接口配额。

建议至少加两层防护：

• 客户端登录态校验，未登录用户踢掉。
• 按用户维度限流，比如每5分钟最多验单3次。

from collections import defaultdict import time _verify_quota: dict[str, list[float]] = defaultdict(list) MAX_ATTEMPTS = 3 WINDOW_SECONDS = 300 def check_verify_quota(user_id: str) -> bool: now = time.time() records = _verify_quota[user_id] records = [t for t in records if now - t < WINDOW_SECONDS] if len(records) >= MAX_ATTEMPTS: return False records.append(now) _verify_quota[user_id] = records return True

生产环境可以把这段逻辑换成Redis计数，单机内存版在有多实例部署时会失效。

7. 一套可运行的最小代码骨架

把上面的内容串起来，最小可用的服务端骨架可以分为三层：接口层、验单与通知处理层、数据访问层。我用一个简化版示意：

from flask import Flask, request, jsonify app = Flask(__name__) @app.post("/api/verify_receipt") def api_verify_receipt(): user_id = request.headers.get("X-User-Id") if not user_id: return jsonify({"code": "NO_AUTH"}), 401 if not check_verify_quota(user_id): return jsonify({"code": "RATE_LIMIT"}), 429 body = request.get_json(silent=True) or {} receipt_str = body.get("receipt_data", "") if not receipt_str: return jsonify({"code": "BAD_REQUEST"}), 400 verify_data = verify_receipt(receipt_str) if verify_data.get("status") != 0: return jsonify({"code": "VERIFY_FAILED", "status": verify_data.get("status")}), 400 parsed = parse_subscription_info(verify_data, product_id=body.get("product_id", "")) order_repo.save_from_parse(user_id, parsed) return jsonify(parsed) @app.post("/apple/notifications") def apple_notifications(): signed_payload = request.get_json(silent=True).get("signedPayload", "") payload = verify_signed_payload(signed_payload) result = process_notification(payload, order_repo, event_repo) # 苹果要求返回 200，否则会重发 return jsonify({"code": result["code"]})

这个骨架虽然简单，但已经把限流、验单、通知处理三个核心环节占住了。你只需要把order_repo和event_repo替换为自己的数据库实现，就可以接上真实业务。

文案上我加一个最后的提示：苹果要求在收到通知后尽快返回HTTP 200，如果你要在通知处理逻辑里做大量数据库写操作，建议先同步记录事件到队列，异步慢慢处理，避免因为处理超时被苹果重复推送。实际线上使用中，把通知处理和订单写入链路做异步解耦，能明显减少积压和锁竞争，这是我调试多套订阅系统后最明显的优化收益。