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 依赖,可换成httpx、urllib3等。
pip install cryptography requestspoetry add cryptography requests引入方式(常规对接只需要这两个名字):
from anony_v2 import AnonyV2Client, auth_header初始化客户端需要 3 个凭据(获取方式见凭据与密钥):
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:
| 层级 | 内容 | 何时用 |
|---|---|---|
AnonyV2Client 类 | build_request / verify_response | 常规对接首选。两个方法覆盖"发业务请求"和"处理响应/回调"的完整流程 |
| 模块级函数 | encrypt / decrypt / sign / verify / canonical / new_nonce / auth_header | auth_header 是 getToken 必用;其余是密码学原语,用于自定义流程或联调时逐步比对中间值(如 canonical 串、密文分段) |
模块级函数一览
| 函数 | 签名 | 说明 |
|---|---|---|
auth_header | auth_header(app_id, secret) | 返回 getToken 用的 Authorization 头值:"auth " + md5(APP-ID + "&" + secret)。分隔符是 &,是最容易手写出错的一步,务必用此函数 |
encrypt | encrypt(plaintext, server_public_key_b64) | RSA-OAEP(SHA-1) 加密,明文按 214 字节分段,返回 base64 字符串 |
decrypt | decrypt(cipher_b64, merchant_private_key_b64) | 密文 Base64 解码后按 256 字节分段解密,返回 bytes(需自行 .decode("utf-8")) |
sign | sign(canonical_str, merchant_private_key_b64) | SHA256withRSA 签名,返回 base64 字符串 |
verify | verify(canonical_str, server_public_key_b64, sign_b64) | SHA256withRSA 验签,返回 True / False(不抛异常) |
canonical | canonical(ts, nonce, payload) | 拼接 canonical 串:ts + "\n" + nonce + "\n" + payload |
new_nonce | new_nonce() | 生成 16 字节 CSPRNG 随机数的 hex 串(32 个字符) |
模块级常量:OAEP_MAX_PLAIN = 214(OAEP 明文分段)、RSA_BLOCK = 256(RSA-2048 密文块)、TS_WINDOW = 300(时间窗 ±5 分钟)。
AnonyV2Client 方法一览
| 方法 | 签名 | 说明 |
|---|---|---|
build_request | build_request(data, token) | 把业务 dict 加密并签名,返回 {"body": <密文>, "headers": <5 个请求头>} |
verify_response | verify_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 字段。完整流程:
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"]三个要点:
auth_header(app_id, secret)生成Authorization头,格式为auth+md5(APP-ID + "&" + secret)。- 外层响应用标准 JSON 库解析(
tr.json()),data字段是 base64 密文。 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 库的 body 和 headers:
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.post 用 data=req["body"] 传原始密文字符串(不要用 json=,body 不是 JSON)。
响应 code != 10000 表示失败,message 为原因、data 为空,直接读 body["message"] 即可,无需解密。错误码速查见错误码。
TRON 链 USDC 充值已关闭
向平台的 TRON 地址转入 USDC 不会入账、也不会触发回调,资金无法自动到账。请勿引导你的用户用 TRON-USDC 向 addressType: "trx" 的地址充值;TRON 链充值请使用 USDT 或 TRX。提现币种枚举中同样不包含 TRON-USDC。
各接口的参数与响应字段以接口文档为准:createAddress、createWithdrawOrder、merchantInfo、getDeposits、checkWithdrawOrder。币种/地址类型枚举见接口公共说明。
第三步:处理响应与回调(verify_response)
verify_response(encrypted_body, timestamp, nonce, sign_b64, is_fresh_nonce=None) 是下行方向的唯一入口,同步响应和异步回调共用。内部按固定顺序执行:
- 校验
timestamp为纯数字且在 ±300 秒窗口内,否则抛ValueError("timestamp out of window"); - 若传入了
is_fresh_nonce,调用is_fresh_nonce(nonce),返回假值则抛ValueError("duplicate nonce (replay)"); - 用平台公钥对 canonical 串验签,失败抛
ValueError("signature verify failed"); - 用商户私钥 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 在请求头里:
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"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 接口把加密、签名、防重放、验签、解密整条链路跑通(可反复调用,不产生任何订单):
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 不在白名单。