Skip to content

Python SDK

Python SDK 是单文件 anony_v2.py(随对接资料包 sdk/python/ 提供),复制进你的项目即可使用,无需安装额外的 SDK 包。它实现了 v2 协议的全部密码学环节:RSA-OAEP(MGF1 = SHA-1)加密、SHA256withRSA 签名、canonical 串拼接、时间戳/nonce 防重放。协议细节见协议规范

与 PHP / Node SDK 不同,Python SDK 不内置 HTTP 调用:它只负责构造请求(加密 + 签名 + 请求头)和处理响应(验签 + 解密),HTTP 由你自选的库发起。本文以 requests 为例。

依赖

  • cryptography:SDK 的唯一硬依赖,负责 RSA 加解密与签名。
  • requests(或任意 HTTP 库):发起 HTTP 请求用,非 SDK 依赖,可换成 httpxurllib3 等。
bash
pip install cryptography requests
bash
poetry add cryptography requests

引入方式(常规对接只需要这两个名字):

python
from anony_v2 import AnonyV2Client, auth_header

初始化客户端需要 3 个凭据(获取方式见凭据与密钥):

python
c = AnonyV2Client(
    app_id,             # ANONY-APP-ID(商户号)
    server_pub_b64,     # 服务端公钥 public_t,base64(PEM)
    merchant_priv_b64,  # 商户私钥 private_u,base64(PEM)
    # base_url 默认 "https://api.anonypay.io/api/merchant/",可省略
)

密钥格式是 base64(PEM)

构造函数接收的两个密钥都是对整段 PEM 文本再做一次 Base64 的字符串,SDK 内部会先 Base64 解码还原 PEM 再加载。直接传裸 PEM 会加载失败。

模块级函数与 AnonyV2Client 类:何时用哪个

anony_v2.py 暴露两层 API:

层级内容何时用
AnonyV2Clientbuild_request / verify_response常规对接首选。两个方法覆盖"发业务请求"和"处理响应/回调"的完整流程
模块级函数encrypt / decrypt / sign / verify / canonical / new_nonce / auth_headerauth_header 是 getToken 必用;其余是密码学原语,用于自定义流程或联调时逐步比对中间值(如 canonical 串、密文分段)

模块级函数一览

函数签名说明
auth_headerauth_header(app_id, secret)返回 getToken 用的 Authorization 头值:"auth " + md5(APP-ID + "&" + secret)分隔符是 &,是最容易手写出错的一步,务必用此函数
encryptencrypt(plaintext, server_public_key_b64)RSA-OAEP(SHA-1) 加密,明文按 214 字节分段,返回 base64 字符串
decryptdecrypt(cipher_b64, merchant_private_key_b64)密文 Base64 解码后按 256 字节分段解密,返回 bytes(需自行 .decode("utf-8")
signsign(canonical_str, merchant_private_key_b64)SHA256withRSA 签名,返回 base64 字符串
verifyverify(canonical_str, server_public_key_b64, sign_b64)SHA256withRSA 验签,返回 True / False(不抛异常)
canonicalcanonical(ts, nonce, payload)拼接 canonical 串:ts + "\n" + nonce + "\n" + payload
new_noncenew_nonce()生成 16 字节 CSPRNG 随机数的 hex 串(32 个字符)

模块级常量:OAEP_MAX_PLAIN = 214(OAEP 明文分段)、RSA_BLOCK = 256(RSA-2048 密文块)、TS_WINDOW = 300(时间窗 ±5 分钟)。

AnonyV2Client 方法一览

方法签名说明
build_requestbuild_request(data, token)把业务 dict 加密并签名,返回 {"body": <密文>, "headers": <5 个请求头>}
verify_responseverify_response(encrypted_body, timestamp, nonce, sign_b64, is_fresh_nonce=None)校验时间窗 → nonce 去重(传入 is_fresh_nonce 时)→ 验签 → 解密,返回业务 dict;任一步失败抛 ValueError

第一步:获取 token(getToken)

getToken 的请求不加密、不签名,仅用 auth_header 做 md5 鉴权;但响应与其它接口一样是加密 + 签名的信封,不是明文 token,必须用 verify_response 验签解密后取 token 字段。完整流程:

python
from anony_v2 import AnonyV2Client, auth_header
import requests

c = AnonyV2Client(app_id, server_pub_b64, merchant_priv_b64)

# 先取 token:请求不加密,仅 md5 鉴权;响应仍需验签+解密
tr = requests.post(c.base_url + "getToken",
                   headers={"ANONY-APP-ID": str(app_id), "Authorization": auth_header(app_id, secret)})
tb = tr.json()
token = c.verify_response(tb["data"], tr.headers["ANONY-TIMESTAMP"],
                          tr.headers["ANONY-NONCE"], tr.headers["ANONY-SIGN"])["token"]

三个要点:

  1. auth_header(app_id, secret) 生成 Authorization 头,格式为 auth + md5(APP-ID + "&" + secret)
  2. 外层响应用标准 JSON 库解析(tr.json()),data 字段是 base64 密文。
  3. verify_response 接收密文与响应头里的 ANONY-TIMESTAMP / ANONY-NONCE / ANONY-SIGN,验签解密后返回 dict,从中取 token

token 有效期(实现细节)

token 采用滑动过期 8 小时 + 绝对上限 24 小时。收到 tokenExpired 错误时重新调用 getToken 即可,建议在业务代码中做好自动刷新。接口详情见 getToken

第二步:发业务请求(build_request)

build_request(data, token) 完成加密 + 时间戳/nonce + 签名,返回可直接交给 HTTP 库的 bodyheaders

python
req = c.build_request({"userOrder": "A1001", "addressType": "trx"}, token)
r = requests.post(c.base_url + "createAddress", data=req["body"], headers=req["headers"])
body = r.json()
if body["code"] == 10000:
    data = c.verify_response(body["data"], r.headers["ANONY-TIMESTAMP"],
                             r.headers["ANONY-NONCE"], r.headers["ANONY-SIGN"])

build_request 内部依次做了:json.dumps(data, ensure_ascii=False) 序列化 → OAEP 加密为 base64 密文 → 生成秒级时间戳与 16 字节 nonce → 对 canonical 串 ts\nnonce\n密文 做 SHA256withRSA 签名。返回的 headers 包含全部 5 个头:ANONY-APP-ID / ANONY-TOKEN / ANONY-TIMESTAMP / ANONY-NONCE / ANONY-SIGN

注意 requests.postdata=req["body"] 传原始密文字符串(不要用 json=,body 不是 JSON)。

响应 code != 10000 表示失败,message 为原因、data 为空,直接读 body["message"] 即可,无需解密。错误码速查见错误码

TRON 链 USDC 充值已关闭

向平台的 TRON 地址转入 USDC 不会入账、也不会触发回调,资金无法自动到账。请勿引导你的用户用 TRON-USDC 向 addressType: "trx" 的地址充值;TRON 链充值请使用 USDT 或 TRX。提现币种枚举中同样不包含 TRON-USDC。

各接口的参数与响应字段以接口文档为准:createAddresscreateWithdrawOrdermerchantInfogetDepositscheckWithdrawOrder。币种/地址类型枚举见接口公共说明

第三步:处理响应与回调(verify_response)

verify_response(encrypted_body, timestamp, nonce, sign_b64, is_fresh_nonce=None) 是下行方向的唯一入口,同步响应和异步回调共用。内部按固定顺序执行:

  1. 校验 timestamp 为纯数字且在 ±300 秒窗口内,否则抛 ValueError("timestamp out of window")
  2. 若传入了 is_fresh_nonce,调用 is_fresh_nonce(nonce),返回假值则抛 ValueError("duplicate nonce (replay)")
  3. 用平台公钥对 canonical 串验签,失败抛 ValueError("signature verify failed")
  4. 用商户私钥 OAEP 解密,json.loads 后返回业务 dict。

is_fresh_nonce:同步响应可省略,回调必须传

is_fresh_nonce 是一个回调函数:收到 nonce,首次见到返回 True,重复返回 False。省略(默认 None)时 verify_response跳过 nonce 去重

  • 处理同步响应(自己刚发出的请求的应答):可以省略。
  • 处理平台回调必须传入,且必须基于原子存储(如 Redis SET key 1 NX EX 900,过期 ≥ 15 分钟)。不传的话,攻击者截获一次合法回调后可以反复重放,让你重复入账。

回调接收示例

回调的 HTTP body 就是 base64 密文本身ANONY-TIMESTAMP / ANONY-NONCE / ANONY-SIGN 在请求头里:

python
import redis
from flask import Flask, request
from anony_v2 import AnonyV2Client

app = Flask(__name__)
rds = redis.Redis()
c = AnonyV2Client(app_id, server_pub_b64, merchant_priv_b64)

def is_fresh_nonce(nonce):
    # 原子去重:首次写入成功返回 True,15 分钟内重复返回 False
    return bool(rds.set("anony:cb:nonce:" + nonce, 1, nx=True, ex=900))

@app.post("/anony/callback")
def anony_callback():
    body = request.get_data(as_text=True)  # 回调体就是密文,不要按 JSON 解析
    try:
        data = c.verify_response(
            body,
            request.headers["ANONY-TIMESTAMP"],
            request.headers["ANONY-NONCE"],
            request.headers["ANONY-SIGN"],
            is_fresh_nonce=is_fresh_nonce,  # 回调必须传,否则可被重放
        )
    except (ValueError, KeyError):
        return "", 400  # 校验失败:返回非 SUCCESS 应答让平台重试,切勿返回 FAIL

    if data["businessType"] == "deposit":
        handle_deposit(data)     # 按 userOrder/apiOrder 幂等入账
    elif data["businessType"] == "withdraw":
        handle_withdraw(data)    # 按 orderStatus 更新订单

    return "SUCCESS"
python
import redis
from fastapi import FastAPI, Request
from fastapi.responses import PlainTextResponse
from anony_v2 import AnonyV2Client

app = FastAPI()
rds = redis.Redis()
c = AnonyV2Client(app_id, server_pub_b64, merchant_priv_b64)

def is_fresh_nonce(nonce):
    # 原子去重:首次写入成功返回 True,15 分钟内重复返回 False
    return bool(rds.set("anony:cb:nonce:" + nonce, 1, nx=True, ex=900))

@app.post("/anony/callback")
async def anony_callback(req: Request):
    body = (await req.body()).decode("utf-8")  # 回调体就是密文,不要按 JSON 解析
    try:
        data = c.verify_response(
            body,
            req.headers["ANONY-TIMESTAMP"],
            req.headers["ANONY-NONCE"],
            req.headers["ANONY-SIGN"],
            is_fresh_nonce=is_fresh_nonce,  # 回调必须传,否则可被重放
        )
    except (ValueError, KeyError):
        # 校验失败:返回非 SUCCESS 应答让平台重试,切勿返回 FAIL
        return PlainTextResponse("", status_code=400)

    if data["businessType"] == "deposit":
        handle_deposit(data)     # 按 userOrder/apiOrder 幂等入账
    elif data["businessType"] == "withdraw":
        handle_withdraw(data)    # 按 orderStatus 更新订单

    return PlainTextResponse("SUCCESS")

回调应答语义(实现细节)

平台以你返回的纯文本 SUCCESS(大写)判定回调成功;对提现类回调,返回 FAIL 表示你拒绝该笔提现(平台会退款处理),其他任何应答都会按退避策略重试(间隔 2s / 30s / 2min / 5min / 15min / 60min,最多 5 次失败后停止)。因此验签/解密失败时绝不能返回 FAIL——返回 HTTP 400 等非 SUCCESS 应答让平台重试即可。

回调的业务字段(充值到账 / 提现状态)详见回调说明

上线前:用 selfcheck 验证整条链路

正式启用前,务必用只读的 v2/selfcheck 接口把加密、签名、防重放、验签、解密整条链路跑通(可反复调用,不产生任何订单):

python
req = c.build_request({"hello": "anony"}, token)
r = requests.post(c.base_url + "v2/selfcheck", data=req["body"], headers=req["headers"])
body = r.json()
data = c.verify_response(body["data"], r.headers["ANONY-TIMESTAMP"],
                         r.headers["ANONY-NONCE"], r.headers["ANONY-SIGN"])
assert data["pong"] is True and data["echo"] == {"hello": "anony"}

能取回 echo 即代表上行(你的加密+签名被服务端接受)与下行(服务端响应被你验签+解密)双向就绪。详见 selfcheck。自检通过后通知平台完成正式启用。

常见问题

  • 解密失败 / 乱码:OAEP 哈希必须是 SHA-1。SDK 内部已写死 padding.OAEP(mgf=MGF1(SHA1()), algorithm=SHA1(), label=None),若你绕过 SDK 自行实现,这是头号互通坑。
  • signatureFailed:canonical 串三段必须用 \n(LF)连接、顺序为 ts, nonce, body;确认服务器 NTP 时间同步(时间戳超 ±5 分钟窗也报这个错)。
  • Oops 纯文本响应:来源 IP 不在白名单。

更多排查见常见问题排查错误码。其他语言 SDK 见 SDK 总览