217 lines
7.9 KiB
Python
217 lines
7.9 KiB
Python
"""支付渠道加密能力测试：RSA2 签名/验签与微信 V3 AES-GCM 解密均为离线可验证。
|
||
|
||
真实商户密钥接入前，用自签密钥对完成「本地签名 → 验签」往返，确保：
|
||
- 支付宝 order_string 可被支付宝公钥验签通过（签名串构造正确）
|
||
- 回调验签能正确拒绝篡改报文
|
||
- 微信 V3 回调解密可还原明文
|
||
"""
|
||
import base64
|
||
import json
|
||
import uuid
|
||
from urllib.parse import parse_qs
|
||
|
||
import pytest
|
||
from cryptography.hazmat.primitives import serialization
|
||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||
|
||
from app.config import settings
|
||
from app.payment import alipay, base, wechat
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def keypair():
|
||
"""生成一对 RSA 密钥（模拟商户私钥 / 平台公钥）"""
|
||
key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||
priv = key.private_bytes(
|
||
encoding=serialization.Encoding.PEM,
|
||
format=serialization.PrivateFormat.PKCS8,
|
||
encryption_algorithm=serialization.NoEncryption(),
|
||
).decode()
|
||
pub = key.public_key().public_bytes(
|
||
encoding=serialization.Encoding.PEM,
|
||
format=serialization.PublicFormat.SubjectPublicKeyInfo,
|
||
).decode()
|
||
return priv, pub
|
||
|
||
|
||
def _pem_body(pem: str, label: str) -> str:
|
||
"""去掉 PEM 头尾与换行，得到裸 base64（支付宝密钥常见下发形式）"""
|
||
body = "\n".join(
|
||
line for line in pem.splitlines() if not line.startswith("-----")
|
||
)
|
||
return body.replace("\n", "")
|
||
|
||
|
||
# ---------- base：密钥加载 / 签名原语 ----------
|
||
|
||
def test_load_private_key_accepts_pem_and_bare_base64(keypair):
|
||
priv, _ = keypair
|
||
assert base.load_private_key(priv) is not None
|
||
# 裸 base64（PKCS8）也应可加载
|
||
assert base.load_private_key(_pem_body(priv, "PRIVATE KEY")) is not None
|
||
|
||
|
||
def test_load_public_key_accepts_pem(keypair):
|
||
_, pub = keypair
|
||
assert base.load_public_key(pub) is not None
|
||
|
||
|
||
def test_load_private_key_empty_raises():
|
||
with pytest.raises(base.PaymentConfigError):
|
||
base.load_private_key("")
|
||
|
||
|
||
def test_rsa2_sign_verify_roundtrip(keypair):
|
||
priv, pub = keypair
|
||
content = "app_id=2021000000&method=alipay.trade.app.pay&version=1.0"
|
||
signature = base.rsa2_sign(content, priv)
|
||
assert base.rsa2_verify(content, signature, pub) is True
|
||
# 篡改内容必须验签失败
|
||
assert base.rsa2_verify(content + "&x=1", signature, pub) is False
|
||
# 篡改签名必须验签失败（非法 base64 也不得抛异常）
|
||
assert base.rsa2_verify(content, "not-a-signature", pub) is False
|
||
|
||
|
||
def test_build_sign_content_sorts_and_filters():
|
||
params = {"b": "2", "a": "1", "sign": "xxx", "sign_type": "RSA2", "empty": ""}
|
||
assert base.build_sign_content(params) == "a=1&b=2"
|
||
|
||
|
||
# ---------- 支付宝 ----------
|
||
|
||
def test_alipay_dev_mock_when_not_configured(monkeypatch):
|
||
monkeypatch.setattr(settings, "alipay_app_id", "")
|
||
result = alipay.create_app_pay_params("R123", 100)
|
||
assert result["dev_mock"] is True
|
||
|
||
|
||
def test_alipay_order_string_signable_and_verifiable(monkeypatch, keypair):
|
||
priv, pub = keypair
|
||
monkeypatch.setattr(settings, "alipay_app_id", "2021000000000000")
|
||
monkeypatch.setattr(settings, "alipay_private_key", priv)
|
||
monkeypatch.setattr(settings, "alipay_public_key", pub)
|
||
|
||
result = alipay.create_app_pay_params("R20261015001", 1234)
|
||
assert "dev_mock" not in result
|
||
params = {k: v[0] for k, v in parse_qs(result["order_string"]).items()}
|
||
|
||
# 关键参数齐备
|
||
assert params["app_id"] == "2021000000000000"
|
||
assert params["method"] == alipay.METHOD_APP_PAY
|
||
assert params["sign_type"] == "RSA2"
|
||
# 金额按分转元、两位小数
|
||
biz = json.loads(params["biz_content"])
|
||
assert biz["total_amount"] == "12.34"
|
||
assert biz["out_trade_no"] == "R20261015001"
|
||
|
||
# order_string 中的参数可被公钥验签通过（即签名串构造正确）
|
||
assert base.rsa2_verify(base.build_sign_content(params), params["sign"], pub) is True
|
||
|
||
|
||
def test_alipay_verify_callback_accepts_valid_and_rejects_tampered(monkeypatch, keypair):
|
||
priv, pub = keypair
|
||
monkeypatch.setattr(settings, "alipay_public_key", pub)
|
||
|
||
form = {
|
||
"out_trade_no": "R20261015001",
|
||
"trade_no": "2026101522001400000001",
|
||
"trade_status": "TRADE_SUCCESS",
|
||
"total_amount": "12.34",
|
||
}
|
||
form["sign"] = base.rsa2_sign(base.build_sign_content(form), priv)
|
||
form["sign_type"] = "RSA2"
|
||
assert alipay.verify_callback(form) is True
|
||
|
||
# 篡改金额 → 验签失败
|
||
assert alipay.verify_callback({**form, "total_amount": "0.01"}) is False
|
||
# 缺少 sign → 拒绝
|
||
form.pop("sign")
|
||
assert alipay.verify_callback(form) is False
|
||
|
||
|
||
# ---------- 微信 V3 ----------
|
||
|
||
def test_wechat_dev_mock_when_not_configured(monkeypatch):
|
||
monkeypatch.setattr(settings, "wechat_mchid", "")
|
||
import asyncio
|
||
result = asyncio.run(wechat.create_app_pay_params("R123", 100))
|
||
assert result["dev_mock"] is True
|
||
|
||
|
||
def test_wechat_app_params_second_sign_verifiable(monkeypatch, keypair):
|
||
"""未接入网络前，直接验证 App 调起参数的二次签名逻辑"""
|
||
import asyncio
|
||
priv, pub = keypair
|
||
monkeypatch.setattr(settings, "wechat_mchid", "1900000001")
|
||
monkeypatch.setattr(settings, "wechat_app_id", "wx1234567890")
|
||
monkeypatch.setattr(settings, "wechat_api_v3_key", "k" * 32)
|
||
monkeypatch.setattr(settings, "wechat_private_key", priv)
|
||
|
||
async def fake_prepay(order_no, amount_fen):
|
||
return "prepay_test_123"
|
||
|
||
monkeypatch.setattr(wechat, "_create_prepay_id", fake_prepay)
|
||
params = asyncio.run(wechat.create_app_pay_params("R20261015001", 1234))
|
||
|
||
assert params["appid"] == "wx1234567890"
|
||
assert params["partnerid"] == "1900000001"
|
||
assert params["prepayid"] == "prepay_test_123"
|
||
assert params["package"] == "Sign=WXPay"
|
||
|
||
sign_str = (
|
||
f"{params['appid']}\n{params['partnerid']}\n{params['prepayid']}\n"
|
||
f"{params['package']}\n{params['noncestr']}\n{params['timestamp']}\n"
|
||
)
|
||
assert base.rsa2_verify(sign_str, params["sign"], pub) is True
|
||
|
||
|
||
def test_wechat_verify_callback_roundtrip(monkeypatch, keypair):
|
||
"""构造一次完整回调：平台私钥签名 + AES-GCM 加密 → 验签解密还原明文"""
|
||
priv, pub = keypair
|
||
api_v3_key = "0123456789abcdef0123456789abcdef"
|
||
monkeypatch.setattr(settings, "wechat_mchid", "1900000001")
|
||
monkeypatch.setattr(settings, "wechat_app_id", "wx1234567890")
|
||
monkeypatch.setattr(settings, "wechat_api_v3_key", api_v3_key)
|
||
monkeypatch.setattr(settings, "wechat_private_key", priv)
|
||
monkeypatch.setattr(settings, "wechat_platform_cert", pub)
|
||
|
||
plain = {
|
||
"out_trade_no": "R20261015001",
|
||
"transaction_id": "4200001234202610150001",
|
||
"trade_state": "SUCCESS",
|
||
"amount": {"total": 1234, "currency": "CNY"},
|
||
}
|
||
nonce = uuid.uuid4().hex[:12]
|
||
ciphertext = AESGCM(api_v3_key.encode()).encrypt(
|
||
nonce.encode(), json.dumps(plain).encode(), b"transaction"
|
||
)
|
||
body = json.dumps({
|
||
"resource": {
|
||
"algorithm": "AEAD_AES_256_GCM",
|
||
"ciphertext": base64.b64encode(ciphertext).decode(),
|
||
"nonce": nonce,
|
||
"associated_data": "transaction",
|
||
}
|
||
}).encode()
|
||
|
||
timestamp, header_nonce = "1792000000", uuid.uuid4().hex
|
||
message = f"{timestamp}\n{header_nonce}\n{body.decode()}\n"
|
||
headers = {
|
||
"wechatpay-timestamp": timestamp,
|
||
"wechatpay-nonce": header_nonce,
|
||
"wechatpay-signature": base.rsa2_sign(message, priv),
|
||
}
|
||
|
||
result = wechat.verify_callback(headers, body)
|
||
assert result is not None
|
||
assert result["out_trade_no"] == "R20261015001"
|
||
assert result["amount"]["total"] == 1234
|
||
|
||
# 篡改 timestamp → 验签失败
|
||
assert wechat.verify_callback({**headers, "wechatpay-timestamp": "1"}, body) is None
|
||
|
||
|
||
def test_wechat_decrypt_rejects_bad_ciphertext():
|
||
assert base.wechat_decrypt_resource({"ciphertext": "!!", "nonce": "abc"}, "k" * 32) is None
|