Python 签名与验签示例

运行环境

Python 3.10+，执行 pip install cryptography requests。回调接入片段另需 Flask。

统一规则

这些示例用于默认商户 API 和默认平台回调，不适用于另行约定的定制签名格式。中英文页面引用同一份可执行代码。

有请求体的 API 请求，使用商户私钥签名：

path + lowercaseHex(SHA512(rawBody)) + Timestamp

无请求体的 API 请求，使用商户私钥签名：

path + Timestamp

默认异步回调，使用平台公钥验签：

lowercaseHex(SHA512(rawBody)) + Timestamp

API 同步响应封装，使用平台公钥验签：

lowercaseHex(SHA512(平台序列化的 data 字节)) + timestamp

算法统一为 RSA PKCS#1 v1.5 + SHA-512，签名使用标准 Base64，不是 RSA-PSS，也不是对原文做 RSA 加密。
Timestamp 使用秒级 Unix 时间戳；验签使用收到的时间戳，不要重新生成。Header 名称不区分大小写，但 Base64 签名值区分大小写。
请求 path 是接口路径，不包含协议、域名或查询字符串。回调原文不拼路径、URL 或分隔符。
JSON 只序列化一次：签名和发送必须使用同一份 UTF-8 字节。字段顺序不要求预先排序，但空格、转义、换行、数字表示和顺序必须与发送内容完全一致。
无请求体使用 null / nil / None；它与“存在但长度为零的请求体”不同。回调始终对收到的完整 body 求摘要。
私钥支持未加密 PKCS#1、PKCS#8；公钥支持 X.509 SPKI、PKCS#1；支持 PEM、纯 Base64 和转义换行。证书、加密私钥、SSH 格式及非 RSA 密钥不在本示例支持范围内。
商户请求用商户私钥；验签必须使用后台配置的 平台公钥。不能使用商户自己的公钥，也不能信任请求中携带的公钥。不要发送私钥给技术支持；泄露后应更换。

失败处理

签名失败抛出异常或返回 error，调用方必须停止发送，不能用空签名继续请求。验签成功正常返回，失败记录 [payment-verify] FAILED 并抛出异常或返回 error；不得忽略失败后继续入账。

日志包含阶段、字节长度和错误原因，不记录原始请求体、完整签名或密钥。生产环境不要打开 HTTP 正文日志；不要把内部诊断错误原样返回给调用方，并对失败请求和日志做限流。

HTTPS 必须校验证书与主机名。网络超时不能证明订单创建失败，不能盲目自动重新下单；应按商户订单号查询或人工核查。

回调接入

先限制请求体大小，读取原始字节，读取 Timestamp、Signature，验签成功后再解析 JSON。检查订单归属、金额、币种和允许的状态变化，通过事务做持久化幂等处理。只有完成可靠处理后才按协议返回 SUCCESS。重复通知不得重复产生资金变动。

验签函数只负责密码学校验，不等于防重放。时间窗口应结合平台重试规则设置；重放检测、持久化幂等、回调请求限流仍由接入方实现。不要因固定短时间窗口而拒绝正常重试。

同步响应不能当作回调

API 同步响应从 JSON 的 timestamp、sign 字段取值，验签对象是 data 的平台序列化字节，不是整个响应体；回调则从 Header 取值，验签对象是完整 HTTP body。

不要把 data 解析后重新序列化再验签：大整数、1.00、转义或字段顺序可能改变。使用能保留 JSON 原始值文本的解析器提取 data（例如 Go json.RawMessage、C# JsonElement.GetRawText()），并把这些 UTF-8 字节传给示例的 verifyPlatform / VerifyPlatform / verify_platform。data 是字符串时保留 JSON 引号与转义，是 null 时保留字面量 null；不要自行添加换行或空格。缺少签名、字段异常或验签失败必须拒绝作为可信业务结果。

接口路径

同一请求函数可用于以下路径，参数以相应币种文档为准：

方法	路径	用途
POST	`/v1/merchant/pay`	代收下单
POST	`/v1/merchant/pay/order/info`	代收查单
POST	`/v1/merchant/payout`	代付下单
POST	`/v1/merchant/payout/order/info`	代付查单
GET	`/v1/merchant/account`	余额查询，无请求体

签名与验签完整代码

sign.py

import base64
import hashlib
import logging
import re
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa

logger = logging.getLogger("payment-signature")


def _timestamp(value):
    text = str(value)
    if not re.fullmatch(r"[1-9][0-9]{0,18}", text) or int(text) > 9223372036854775807:
        raise ValueError("invalid Timestamp: use Unix seconds")
    return text


def _base64(text, label):
    try:
        result = base64.b64decode(text, validate=True)
        if not result or base64.b64encode(result).decode("ascii") != text:
            raise ValueError()
        return result
    except (ValueError, TypeError):
        raise ValueError(label + ": invalid Base64") from None


def _key(text, private):
    if not isinstance(text, str) or len(text) > 16384:
        raise ValueError("key: missing or too large")
    text = text.replace(r"\r\n", "\n").replace(r"\n", "\n").strip()
    if text.startswith("-----BEGIN"):
        match = re.fullmatch(r"-----BEGIN ([A-Z ]+)-----\s*([A-Za-z0-9+/=\s]+)\s*-----END \1-----", text)
        labels = ("PRIVATE KEY", "RSA PRIVATE KEY") if private else ("PUBLIC KEY", "RSA PUBLIC KEY")
        if not match or match[1] not in labels:
            raise ValueError("key: expected one unencrypted RSA PEM block of the correct kind")
        text = match[2]
    der = _base64("".join(text.split()), "key")
    try:
        key = serialization.load_der_private_key(der, password=None) if private else serialization.load_der_public_key(der)
        expected = rsa.RSAPrivateKey if private else rsa.RSAPublicKey
        if not isinstance(key, expected):
            raise ValueError()
        return key
    except (ValueError, TypeError):
        raise ValueError("key: invalid/truncated, encrypted or non-RSA key") from None


def _hash(body):
    if not isinstance(body, bytes):
        raise ValueError("body: expected raw bytes, not parsed JSON")
    return hashlib.sha512(body).hexdigest()


def _log(operation, body, error):
    logger.error("[payment-%s] FAILED body_bytes=%d error=%s", operation,
                 len(body) if isinstance(body, bytes) else 0, str(error))


def sign_request(path, body, timestamp, merchant_private_key):
    """None means no body; b'' means a present, empty body. Returns Base64 or raises."""
    try:
        if not isinstance(path, str) or not path.startswith("/"):
            raise ValueError("path: expected API path")
        origin = path + ("" if body is None else _hash(body)) + _timestamp(timestamp)
        key = _key(merchant_private_key, True)
        signature = key.sign(origin.encode("utf-8"), padding.PKCS1v15(), hashes.SHA512())
        return base64.b64encode(signature).decode("ascii")
    except Exception as error:
        _log("sign", body, error)
        raise


def verify_platform(body, timestamp, signature, platform_public_key):
    """Exact callback bytes, or exact signed data bytes for API envelopes. NO path."""
    try:
        origin = _hash(body) + _timestamp(timestamp)
        key = _key(platform_public_key, False)
        sig = _base64(signature, "Signature")
        expected = (key.key_size + 7) // 8
        if len(sig) != expected:
            raise ValueError(f"Signature length: got {len(sig)}, expected {expected} bytes")
        try:
            key.verify(sig, origin.encode("utf-8"), padding.PKCS1v15(), hashes.SHA512())
        except InvalidSignature:
            raise ValueError("RSA SHA512 verification failed: check platform key, raw bytes, timestamp and formula") from None
    except Exception as error:
        _log("verify", body, error)
        raise

可直接运行的回调验签样例

以下独立脚本包含完整测试平台公钥、原始 Body、Timestamp 和对应 Signature，无需下单、联网或提供私钥。仅依赖 cryptography，不依赖 sign.py。

pip install cryptography
python verify_callback_demo.py

把下方完整代码放在 verify_callback_demo.py 中运行：

"""Offline callback verification vector. Test public key only; never use in production."""
import base64
import hashlib

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa


PLATFORM_PUBLIC_KEY = b"""-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAskopsusffzyPW77qfwfp
/daafyzflakpa3YoGRS+RrjDyNSsJnszOSPgyjsJvaHhh3qBd4qsH4N0+G15mP8c
9s8dhz0VWySwu+Xxo+pJBjurnkGpDBclTgecL5nKDj3SwB6gJ7LR5i0Ey+DGURvL
MUEVJO2UW6ZzcpPKYpbJGsNiQa1XsB9k/Wx22MEZlbAuFu2C4dOLJkApjAOULcnF
Ik80/yFvXIvqZIfLNwtffPlkXXw/I2sxP13Gne0edQvPOAJCyUV6UmDKNILLS8X0
nNi6ZzaVSbfe0BKB/0Yq5pUktn00Ridt86oktSG7wmUmsUFL11WnGpEVN2q3m8Bv
vwIDAQAB
-----END PUBLIC KEY-----
"""
RAW_BODY = (
    b'{"transactionId":"TEST-PKR-001","merchantSerialNo":"TEST-MERCHANT-001",'
    b'"currency":"pkr","method":"BANK","amount":"100.00",'
    b'"orderAmount":"100.00","status":"20","fee":"1.00"}'
)
TIMESTAMP = "1792132760"
SIGNATURE = "XfXOGFZPUgEckD3m1+ncrXC0KiXIHxpoTjls16ULOO5gByIMCjn4LLB/+85kNzEUbq2vfnj7NRIbS/vxoG6y/N/ht/xVP89HkwWaVGYJPF5R4g46VnGatRtHSPfENm8sJXTyyqNUnHBecSxVF4k5JRlItpue8IGKmtfpzYlqlaUylDSuhAHo7ZX7JVGihSJmZxN7fHy8d+KiDV8tF3cKswvvq9aefbSWq8nG9Lw+536dey5ZV9YUxfaI+OxrG9+N0F2jR7zvKI/EiilIrO5mLNd2aN7hrdJbXY6PQOrWiY9m0+DRMQxURnGGyupLPYuFbr6ApP3jb6VfMaoUS1n7tw=="
EXPECTED_BODY_SHA512 = "e246541235790a61cd2d10a2748e817aaefb0b147c25456420f380859a7a4e43e491f67b8bb234cd0a5e08d1dda00c7cb66c66d86fae3ec23d3ba4f2627891b6"


def verify_callback(raw_body, timestamp, signature, platform_public_key):
    """Verify received raw bytes; no path, JSON sorting or re-serialization."""
    key = serialization.load_pem_public_key(platform_public_key)
    if not isinstance(key, rsa.RSAPublicKey):
        raise ValueError("Platform public key must be RSA")
    signature_bytes = base64.b64decode(signature, validate=True)
    expected_bytes = (key.key_size + 7) // 8
    if len(signature_bytes) != expected_bytes:
        raise ValueError(
            f"Signature is {len(signature_bytes)} bytes; "
            f"the {key.key_size}-bit key expects {expected_bytes} bytes"
        )
    body_hash = hashlib.sha512(raw_body).hexdigest()
    signed_text = (body_hash + timestamp).encode("utf-8")
    key.verify(signature_bytes, signed_text, padding.PKCS1v15(), hashes.SHA512())


def main():
    assert hashlib.sha512(RAW_BODY).hexdigest() == EXPECTED_BODY_SHA512
    verify_callback(RAW_BODY, TIMESTAMP, SIGNATURE, PLATFORM_PUBLIC_KEY)
    print("RSA key: 2048 bits")
    print("Signature: 256 bytes")
    print("Callback signature: PASS")

    try:
        verify_callback(RAW_BODY + b" ", TIMESTAMP, SIGNATURE, PLATFORM_PUBLIC_KEY)
    except InvalidSignature:
        print("Changed raw body: rejected (expected)")
    else:
        raise AssertionError("Changed body was accepted")


if __name__ == "__main__":
    main()

预期输出：

RSA key: 2048 bits
Signature: 256 bytes
Callback signature: PASS
Changed raw body: rejected (expected)

样例使用固定测试数据，不代表当前真实订单或线上密钥。生产环境必须使用该商户后台的平台公钥，不能使用商户公钥或本文测试公钥；使用收到的原始请求体字节、Timestamp 和 Signature，不要重新序列化 JSON、增删空格或追加换行。回调签名不包含 URL 或 path。

RSA 签名解码后的字节数应与公钥位数匹配：2048 位对应 256 字节，1024 位对应 128 字节。如果长度不匹配，先核对是否拿错平台公钥；仅改变 PEM 包装或 Base64 格式不能改变密钥位数，也不能修复公私钥不匹配。

HTTP 请求工具

http_util.py

import time
import requests
from sign import sign_request


def send(base_url, path, merchant_code, raw_body, merchant_private_key):
    """raw_body is bytes or None. Never use requests' json= after signing."""
    if not base_url.startswith("https://"):
        raise ValueError("HTTPS is required")
    timestamp = str(int(time.time()))
    signature = sign_request(path, raw_body, timestamp, merchant_private_key)
    headers = {"Content-Type": "application/json", "Merchant-Code": merchant_code,
               "Timestamp": timestamp, "Signature": signature}
    with requests.request("GET" if raw_body is None else "POST", base_url + path,
                          data=raw_body, headers=headers, timeout=(10, 60),
                          allow_redirects=False, verify=True, stream=True) as response:
        if not 200 <= response.status_code < 300:
            raise RuntimeError(f"HTTP {response.status_code}; reconcile before retrying")
        chunks, size = [], 0
        for chunk in response.iter_content(65536):
            size += len(chunk)
            if size > 8 * 1024 * 1024:
                raise RuntimeError("Response exceeds 8 MiB")
            chunks.append(chunk)
        return b"".join(chunks)  # Preserve raw response for envelope verification.

接入示例

以下片段放在接入方应用中使用，payload、可信配置和事务业务处理函数由接入方提供；示例不会自动发起真实下单。

import json
from http_util import send
from sign import verify_platform

body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
response_bytes = send(base_url, "/v1/merchant/pay", merchant_code, body, merchant_private_key)
# GET: send(base_url, "/v1/merchant/account", merchant_code, None, merchant_private_key)
# Verify the signed response data before using it. Do not treat HTTP 200 as payment success.

# Flask example: app.config["MAX_CONTENT_LENGTH"] = 1 << 20
@app.post("/callback")
def callback():
    raw_body = request.get_data(cache=True)
    try:
        verify_platform(raw_body, request.headers.get("Timestamp"),
                        request.headers.get("Signature"), configured_platform_public_key)
    except Exception:
        return "invalid signature", 401
    event = json.loads(raw_body)
    validate_and_apply_idempotently(event)  # Your durable business transaction.
    return "SUCCESS", 200