114 lines
4.3 KiB
Python
114 lines
4.3 KiB
Python
"""支付适配层公共能力：
|
|
|
|
- 密钥加载与 PEM 规整（兼容支付宝常见的「无头无尾裸 base64」私钥/公钥）
|
|
- RSA2(SHA256withRSA) 签名与验签（支付宝 RSA2 / 微信 V3 共用同一原语）
|
|
- 微信 V3 回调 AES-256-GCM 解密
|
|
|
|
三渠道统一把解析结果交给 payment_service.settle_order 幂等入账。
|
|
"""
|
|
import base64
|
|
import json
|
|
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import padding, rsa
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
|
|
|
|
class PaymentConfigError(Exception):
|
|
"""支付渠道密钥/参数不可用（配置缺失或格式错误）"""
|
|
|
|
|
|
# ---------- 密钥加载 ----------
|
|
|
|
def _normalize_pem(key: str, label: str) -> bytes:
|
|
"""把裸 base64 或带换行内容规整为标准 PEM；已带头尾则原样返回"""
|
|
key = (key or "").strip()
|
|
if not key:
|
|
raise PaymentConfigError(f"{label} 未配置")
|
|
if key.startswith("-----BEGIN"):
|
|
return key.encode()
|
|
body = "".join(key.split())
|
|
chunks = "\n".join(body[i:i + 64] for i in range(0, len(body), 64))
|
|
return f"-----BEGIN {label}-----\n{chunks}\n-----END {label}-----\n".encode()
|
|
|
|
|
|
def load_private_key(key: str) -> rsa.RSAPrivateKey:
|
|
pem = _normalize_pem(key, "RSA PRIVATE KEY")
|
|
try:
|
|
loaded = serialization.load_pem_private_key(pem, password=None)
|
|
except Exception as exc: # noqa: BLE001
|
|
# 支付宝密钥常见为 PKCS8，裸 base64 时按 PKCS8 再试一次
|
|
try:
|
|
pem8 = _normalize_pem(key, "PRIVATE KEY")
|
|
loaded = serialization.load_pem_private_key(pem8, password=None)
|
|
except Exception as exc2: # noqa: BLE001
|
|
raise PaymentConfigError(f"私钥格式错误: {exc2}") from exc
|
|
if not isinstance(loaded, rsa.RSAPrivateKey):
|
|
raise PaymentConfigError("私钥不是 RSA 类型")
|
|
return loaded
|
|
|
|
|
|
def load_public_key(key: str) -> rsa.RSAPublicKey:
|
|
pem = _normalize_pem(key, "PUBLIC KEY")
|
|
try:
|
|
loaded = serialization.load_pem_public_key(pem)
|
|
except Exception as exc: # noqa: BLE001
|
|
try:
|
|
pem_cert = _normalize_pem(key, "CERTIFICATE")
|
|
loaded = serialization.load_pem_public_key(pem_cert)
|
|
except Exception as exc2: # noqa: BLE001
|
|
raise PaymentConfigError(f"公钥格式错误: {exc2}") from exc
|
|
if not isinstance(loaded, rsa.RSAPublicKey):
|
|
raise PaymentConfigError("公钥不是 RSA 类型")
|
|
return loaded
|
|
|
|
|
|
# ---------- RSA2 签名 / 验签 ----------
|
|
|
|
def rsa2_sign(content: str, private_key: str) -> str:
|
|
"""SHA256withRSA 签名，返回 base64"""
|
|
key = load_private_key(private_key)
|
|
signature = key.sign(content.encode("utf-8"), padding.PKCS1v15(), hashes.SHA256())
|
|
return base64.b64encode(signature).decode()
|
|
|
|
|
|
def rsa2_verify(content: str, signature_b64: str, public_key: str) -> bool:
|
|
"""SHA256withRSA 验签；内容/签名异常一律返回 False（不抛错，交由调用方拒绝入账）"""
|
|
try:
|
|
key = load_public_key(public_key)
|
|
key.verify(
|
|
base64.b64decode(signature_b64),
|
|
content.encode("utf-8"),
|
|
padding.PKCS1v15(),
|
|
hashes.SHA256(),
|
|
)
|
|
return True
|
|
except Exception: # noqa: BLE001
|
|
return False
|
|
|
|
|
|
def build_sign_content(params: dict) -> str:
|
|
"""待验签/待签名串：剔除 sign/sign_type 与空值，按 key 字典序，k=v 以 & 连接"""
|
|
items = sorted(
|
|
(k, v) for k, v in params.items()
|
|
if k not in ("sign", "sign_type") and v not in (None, "")
|
|
)
|
|
return "&".join(f"{k}={v}" for k, v in items)
|
|
|
|
|
|
# ---------- 微信 V3 回调解密 ----------
|
|
|
|
def wechat_decrypt_resource(resource: dict, api_v3_key: str) -> dict | None:
|
|
"""AES-256-GCM 解密回调 resource，返回业务明文 dict；失败返回 None"""
|
|
try:
|
|
ciphertext = base64.b64decode(resource.get("ciphertext", ""))
|
|
nonce = resource.get("nonce", "").encode("utf-8")
|
|
associated = (resource.get("associated_data") or "").encode("utf-8")
|
|
if not ciphertext or not nonce:
|
|
return None
|
|
aesgcm = AESGCM(api_v3_key.encode("utf-8"))
|
|
plain = aesgcm.decrypt(nonce, ciphertext, associated)
|
|
return json.loads(plain.decode("utf-8"))
|
|
except Exception: # noqa: BLE001
|
|
return None
|