root/Neo-ZQYY

Fork 0

Files

Neo 6e20987d2f 微信小程序页面迁移校验之前 P5任务处理之前

2026-03-09 01:19:21 +08:00

16 KiB

Raw Permalink Blame History

微信小程序 API 安全（证书 + 签名 + 加密）使用说明

校验依据：https://developers.weixin.qq.com/miniprogram/dev/server/getting_started/api_signature.html

背景

微信小程序部分敏感 API（如获取手机号等）支持服务通信二次加密和签名，可有效防止数据篡改与泄露。开发者在小程序管理后台「开发 → 开发管理 → 开发设置 → API 安全」进行密钥配置。

你拿到的四样材料

材料	微信后台对应位置	用途
开放平台证书编号（SN）	API 安全页面显示	放在请求头 `Wechatmp-Serial` 中标识证书
`.cer` 平台证书文件	配置完公钥后下载	验证微信 API 响应的签名（RSAwithSHA256）
对称密钥（AES256 Key）	API 对称密钥处配置	加密请求数据 + 解密响应数据（AES256_GCM）
非对称密钥（RSA 私钥）	API 非对称密钥处配置	对请求签名（RSAwithSHA256 + PSS 填充）

支持的算法组合

微信支持以下算法，在管理后台配置：

类型	可选算法
加密算法	AES256_GCM、SM4_GCM
签名算法	RSAwithSHA256、SM2withSM3

以下以 AES256_GCM + RSAwithSHA256 为例说明。

完整流程概览

请求：原始数据 → AES256_GCM 加密 → RSAwithSHA256 签名 → 发送
响应：接收 → RSAwithSHA256 验签 → AES256_GCM 解密 → 原始数据

配置步骤

1. 将密钥文件放到服务器

New-Item -ItemType Directory -Path D:\NeoZQYY\secrets -Force

将以下文件放入 D:\NeoZQYY\secrets\：

wx_api_private_key.pem — 应用私钥（RSA）
wx_api_platform_cert.cer — 平台证书（微信的公钥，用于验签响应）

2. 配置环境变量

在 apps/backend/.env.local 中添加：

# API 安全 - 对称密钥编号（微信后台 API 安全页面显示的 SN）
WX_API_SYM_SN=你的对称密钥编号

# API 安全 - 对称密钥（AES256 Key，Base64 编码的 32 字节密钥）
WX_API_SYMMETRIC_KEY=你的对称密钥明文

# API 安全 - 非对称密钥编号
WX_API_ASYM_SN=你的非对称密钥编号

# API 安全 - RSA 私钥文件路径
WX_API_PRIVATE_KEY_PATH=D:/NeoZQYY/secrets/wx_api_private_key.pem

# API 安全 - 平台证书文件路径（微信的公钥，用于验签响应）
WX_API_PLATFORM_CERT_PATH=D:/NeoZQYY/secrets/wx_api_platform_cert.cer

# API 安全 - 平台证书编号（微信后台显示，非证书内序列号）
WX_API_PLATFORM_CERT_SN=平台证书编号

3. 后端工具类

在 apps/backend/app/utils/ 下新建 wx_api_sign.py：

"""
微信小程序 API 安全 — 请求加密 + 签名 + 响应验签 + 解密

校验依据：
https://developers.weixin.qq.com/miniprogram/dev/server/getting_started/api_signature.html

流程：
1. 请求加密：AES256_GCM 加密请求数据
2. 请求签名：RSAwithSHA256 + PSS 填充（salt=32）对密文签名
3. 响应验签：用平台证书公钥验证响应签名
4. 响应解密：AES256_GCM 解密响应数据
"""
import base64
import json
import os
import time
from pathlib import Path

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding as asym_padding
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.x509 import load_pem_x509_certificate

from app.config import get


class WxApiSecurity:
    """微信 API 安全：加密 + 签名 + 验签 + 解密"""

    def __init__(self):
        self.app_id: str = get("WX_APP_ID", "")
        self.sym_sn: str = get("WX_API_SYM_SN", "")
        self.asym_sn: str = get("WX_API_ASYM_SN", "")
        self.platform_cert_sn: str = get("WX_API_PLATFORM_CERT_SN", "")

        # 对称密钥（AES256，Base64 编码）
        sym_key_b64 = get("WX_API_SYMMETRIC_KEY", "")
        self.sym_key = base64.b64decode(sym_key_b64) if sym_key_b64 else None

        # 加载 RSA 私钥（应用私钥，用于签名请求）
        key_path = get("WX_API_PRIVATE_KEY_PATH", "")
        if key_path and Path(key_path).exists():
            with open(key_path, "rb") as f:
                self.private_key = serialization.load_pem_private_key(
                    f.read(), password=None
                )
        else:
            self.private_key = None

        # 加载平台证书（微信公钥，用于验签响应）
        cert_path = get("WX_API_PLATFORM_CERT_PATH", "")
        if cert_path and Path(cert_path).exists():
            with open(cert_path, "rb") as f:
                cert = load_pem_x509_certificate(f.read())
                self.platform_public_key = cert.public_key()
        else:
            self.platform_public_key = None

    # ========== 请求加密（AES256_GCM）==========

    def encrypt_request(
        self, url_path: str, req_data: dict, timestamp: int
    ) -> dict:
        """
        加密请求数据。

        参数：
            url_path: 完整 API URL（含协议，不含 query 参数），
                      如 "https://api.weixin.qq.com/wxa/business/getuserphonenumber"
            req_data: 原始请求参数字典
            timestamp: 时间戳

        返回：
            加密后的请求体 {"iv": "...", "data": "...", "authtag": "..."}
        """
        if not self.sym_key:
            raise RuntimeError("对称密钥未配置")

        # 在原数据内添加安全字段
        req_data["_n"] = base64.b64encode(os.urandom(16)).decode()
        req_data["_appid"] = self.app_id
        req_data["_timestamp"] = timestamp

        plaintext = json.dumps(req_data, ensure_ascii=False).encode("utf-8")

        # GCM 额外认证数据（AAD）：urlpath|appid|timestamp|sn
        aad = f"{url_path}|{self.app_id}|{timestamp}|{self.sym_sn}".encode()

        # 12 字节随机 IV
        iv = os.urandom(12)
        aesgcm = AESGCM(self.sym_key)
        # encrypt 返回 ciphertext + tag（最后 16 字节是 tag）
        ct_with_tag = aesgcm.encrypt(iv, plaintext, aad)
        ciphertext = ct_with_tag[:-16]
        authtag = ct_with_tag[-16:]

        return {
            "iv": base64.b64encode(iv).decode(),
            "data": base64.b64encode(ciphertext).decode(),
            "authtag": base64.b64encode(authtag).decode(),
        }

    # ========== 请求签名（RSAwithSHA256 + PSS）==========

    def sign_request(
        self, url_path: str, post_data: str, timestamp: int
    ) -> dict[str, str]:
        """
        生成签名请求头。

        参数：
            url_path: 完整 API URL（含协议，不含 query 参数）
            post_data: 加密后的请求体 JSON 字符串
            timestamp: 时间戳（与加密时一致）

        返回：
            签名相关的请求头字典
        """
        if not self.private_key:
            raise RuntimeError("RSA 私钥未配置")

        # 待签名串格式：urlpath\nappid\ntimestamp\npostdata
        # 字段之间用换行符分隔，末尾无额外换行符
        payload = f"{url_path}\n{self.app_id}\n{timestamp}\n{post_data}"

        # RSAwithSHA256 + PSS 填充，salt 长度为 32
        signature = self.private_key.sign(
            payload.encode("utf-8"),
            asym_padding.PSS(
                mgf=asym_padding.MGF1(hashes.SHA256()),
                salt_length=32,
            ),
            hashes.SHA256(),
        )

        return {
            "Wechatmp-Appid": self.app_id,
            "Wechatmp-TimeStamp": str(timestamp),
            "Wechatmp-Signature": base64.b64encode(signature).decode(),
        }

    # ========== 响应验签 ==========

    def verify_response(
        self, url_path: str, resp_data: str, timestamp: str,
        signature_b64: str, serial: str
    ) -> bool:
        """
        验证微信 API 响应签名。

        参数：
            url_path: 请求的 API URL
            resp_data: 响应体字符串
            timestamp: 响应头 Wechatmp-TimeStamp
            signature_b64: 响应头 Wechatmp-Signature（Base64）
            serial: 响应头 Wechatmp-Serial（平台证书编号）
        """
        if not self.platform_public_key:
            raise RuntimeError("平台证书未配置")

        # 待验签串：urlpath\nappid\ntimestamp\nrespdata
        payload = f"{url_path}\n{self.app_id}\n{timestamp}\n{resp_data}"
        signature = base64.b64decode(signature_b64)

        try:
            self.platform_public_key.verify(
                signature,
                payload.encode("utf-8"),
                asym_padding.PSS(
                    mgf=asym_padding.MGF1(hashes.SHA256()),
                    salt_length=asym_padding.PSS.AUTO,
                ),
                hashes.SHA256(),
            )
            return True
        except Exception:
            return False

    # ========== 响应解密（AES256_GCM）==========

    def decrypt_response(
        self, resp_body: dict, url_path: str, timestamp: str
    ) -> dict:
        """
        解密微信 API 响应数据。

        参数：
            resp_body: 响应体 {"iv": "...", "data": "...", "authtag": "..."}
            url_path: 请求的 API URL
            timestamp: 响应头 Wechatmp-TimeStamp
        """
        if not self.sym_key:
            raise RuntimeError("对称密钥未配置")

        iv = base64.b64decode(resp_body["iv"])
        ciphertext = base64.b64decode(resp_body["data"])
        authtag = base64.b64decode(resp_body["authtag"])

        # AAD 格式与请求时一致
        aad = f"{url_path}|{self.app_id}|{timestamp}|{self.sym_sn}".encode()

        aesgcm = AESGCM(self.sym_key)
        plaintext = aesgcm.decrypt(iv, ciphertext + authtag, aad)
        return json.loads(plaintext)

    # ========== 便捷方法：加密 + 签名一步到位 ==========

    def prepare_request(
        self, url_path: str, req_data: dict
    ) -> tuple[str, dict[str, str]]:
        """
        一步完成加密 + 签名，返回 (加密后的body字符串, 请求头字典)。
        """
        timestamp = int(time.time())

        # 1. 加密
        encrypted = self.encrypt_request(url_path, req_data, timestamp)
        body_str = json.dumps(encrypted, ensure_ascii=False)

        # 2. 签名
        headers = self.sign_request(url_path, body_str, timestamp)
        headers["Content-Type"] = "application/json"

        return body_str, headers


# 模块级单例
wx_security = WxApiSecurity()

4. 使用示例：获取手机号

import httpx
import json
from app.utils.wx_api_sign import wx_security


async def get_user_phone_number(access_token: str, code: str) -> dict:
    """
    调用微信接口获取用户手机号（带加密 + 签名）。
    code 来自小程序端 getPhoneNumber 按钮回调。

    接口文档：
    https://developers.weixin.qq.com/miniprogram/dev/server/API/user-info/phone-number/api_getphonenumber.html
    """
    url_path = "https://api.weixin.qq.com/wxa/business/getuserphonenumber"
    full_url = f"{url_path}?access_token={access_token}"

    # 加密 + 签名
    body_str, headers = wx_security.prepare_request(
        url_path, {"code": code}
    )

    async with httpx.AsyncClient() as client:
        resp = await client.post(full_url, content=body_str, headers=headers)

        # 验签（从响应头获取签名信息）
        resp_sig = resp.headers.get("Wechatmp-Signature", "")
        resp_ts = resp.headers.get("Wechatmp-TimeStamp", "")
        resp_serial = resp.headers.get("Wechatmp-Serial", "")

        if resp_sig:
            verified = wx_security.verify_response(
                url_path, resp.text, resp_ts, resp_sig, resp_serial
            )
            if not verified:
                raise ValueError("微信响应签名验证失败")

        # 解密响应
        resp_body = resp.json()
        if "iv" in resp_body and "data" in resp_body:
            return wx_security.decrypt_response(resp_body, url_path, resp_ts)
        else:
            # 未加密的响应（错误码等）
            return resp_body

不使用加密的简单调用（仅签名）

如果某些接口只需要签名不需要加密，可以只用签名部分：

async def call_wx_api_sign_only(
    access_token: str, url_path: str, data: dict
) -> dict:
    """仅签名，不加密请求体"""
    full_url = f"{url_path}?access_token={access_token}"
    timestamp = int(time.time())
    body_str = json.dumps(data, ensure_ascii=False)

    headers = wx_security.sign_request(url_path, body_str, timestamp)
    headers["Content-Type"] = "application/json"

    async with httpx.AsyncClient() as client:
        resp = await client.post(full_url, content=body_str, headers=headers)
        return resp.json()

依赖安装

pip install cryptography httpx

cryptography — RSA 签名（PSS 填充）+ AES256_GCM 加解密 + X.509 证书解析
httpx — 异步 HTTP 客户端

注意：API 安全使用的是 AES256_GCM（不是 CBC），cryptography 库原生支持，不需要 pycryptodome。

请求头说明

Header	说明
`Wechatmp-Appid`	当前小程序的 AppID
`Wechatmp-TimeStamp`	签名时的时间戳
`Wechatmp-Signature`	签名数据（Base64 编码）

响应头说明

Header	说明
`Wechatmp-Appid`	小程序 AppID
`Wechatmp-TimeStamp`	签名时间戳
`Wechatmp-Serial`	平台证书编号（在 MP 管理页获取，非证书内序列号）
`Wechatmp-Signature`	平台证书签名（Base64）
`Wechatmp-Serial-Deprecated`	即将失效的证书编号（仅证书更换周期内出现）
`Wechatmp-Signature-Deprecated`	即将失效的证书签名（仅证书更换周期内出现）

若 Wechatmp-Serial-Deprecated 出现，说明当前使用的平台证书即将过期，请尽快到 MP 后台更新。

签名算法关键细节

签名使用 PSS 填充方式，salt 长度为 32
待签名串格式：urlpath\nappid\ntimestamp\npostdata，字段间用 \n 分隔，末尾无额外换行
urlpath 需要包含 HTTP 协议头（如 https://api.weixin.qq.com/wxa/...），不包含 URL 参数
PSS 签名包含随机因子，每次签名结果都会变化，这是正常的

加密算法关键细节

使用 AES256_GCM 模式（不是 CBC）
IV 为 12 字节随机字符串
AAD（额外认证数据）格式：urlpath|appid|timestamp|sn，用竖线 | 分隔
加密后的 data 明文需要额外包含 _n（随机串）、_appid、_timestamp 三个安全字段
不包含 AccessToken（AccessToken 放在 URL 参数中）

错误码

错误码	说明
40230	缺少 Wechatmp-Serial
40231	缺少 Wechatmp-Timestamp
40232	缺少 Wechatmp-Signature
40233	缺少 Wechatmp-Appid
40234	签名错误
40235	加密错误
40236	无效的 Wechatmp-Appid
40237	Wechatmp-Appid 和 Token 不匹配
40238	开发者未设置对称密钥
40239	开发者未设置公钥
40240	超时的数据

安全注意事项

RSA 私钥文件绝对不能提交到 Git，放在 D:\NeoZQYY\secrets\ 并确保 .gitignore 排除
对称密钥只放在 .env.local 中，不要硬编码
平台证书有有效期，注意响应头中的 Wechatmp-Serial-Deprecated 提示
测试环境和正式环境使用同一套密钥（API 安全是 AppID 级别的）

16 KiB Raw Permalink Blame History Unescape Escape