359 lines
16 KiB
Python
359 lines
16 KiB
Python
"""Small adapter around Alipay's official Python SDK.
|
||
|
||
The SDK is imported lazily so local development and the disabled/simulation
|
||
mode do not require credentials or network access.
|
||
"""
|
||
|
||
from decimal import Decimal
|
||
import hashlib
|
||
import json
|
||
from pathlib import Path
|
||
import re
|
||
|
||
from .config import settings
|
||
|
||
|
||
class AlipayTransferError(RuntimeError):
|
||
def __init__(self, message: str, code: str = "", ambiguous: bool = False):
|
||
super().__init__(message)
|
||
self.code = code
|
||
self.ambiguous = ambiguous
|
||
|
||
|
||
def validate_configuration() -> None:
|
||
missing = [
|
||
label for label, value in (
|
||
("ALIPAY_APP_ID", settings.alipay_app_id),
|
||
("ALIPAY_PAYER_ACCOUNT", settings.alipay_payer_account),
|
||
("ALIPAY_APP_PRIVATE_KEY_PATH", settings.alipay_app_private_key_path),
|
||
("ALIPAY_PUBLIC_KEY_PATH 或 ALIPAY_ALIPAY_PUBLIC_CERT_PATH", settings.alipay_public_key_path or settings.alipay_alipay_public_cert_path),
|
||
("ALIPAY_TRANSFER_SCENE_NAME", settings.alipay_transfer_scene_name),
|
||
("ALIPAY_TRANSFER_JOB_TYPE", settings.alipay_transfer_job_type),
|
||
) if not str(value or "").strip()
|
||
]
|
||
if missing:
|
||
raise AlipayTransferError(f"支付宝配置不完整：{', '.join(missing)}")
|
||
if settings.alipay_notify_url and not settings.alipay_notify_url.lower().startswith("https://"):
|
||
raise AlipayTransferError("ALIPAY_NOTIFY_URL 必须使用公网 HTTPS 地址")
|
||
_sdk_key(_read_key(settings.alipay_app_private_key_path, "应用私钥"), private=True)
|
||
if _certificate_mode_enabled():
|
||
app_cert = _load_certificate(settings.alipay_app_cert_path, "应用公钥证书")
|
||
alipay_cert = _load_certificate(settings.alipay_alipay_public_cert_path, "支付宝公钥证书")
|
||
root_certs = _load_certificate_bundle(settings.alipay_root_cert_path, "支付宝根证书")
|
||
_certificate_serial(app_cert)
|
||
_certificate_serial(alipay_cert)
|
||
if not root_certs:
|
||
raise AlipayTransferError("支付宝根证书文件中没有可识别的证书")
|
||
_root_certificate_serial(root_certs)
|
||
private = _sdk_key(_read_key(settings.alipay_app_private_key_path, "应用私钥"), private=True)
|
||
if not _same_rsa_key(private, _certificate_public_key(app_cert)):
|
||
raise AlipayTransferError("应用公钥证书与应用私钥不匹配")
|
||
else:
|
||
_sdk_key(_read_key(settings.alipay_public_key_path, "支付宝公钥"), private=False)
|
||
|
||
|
||
def _certificate_mode_enabled() -> bool:
|
||
return any((settings.alipay_app_cert_path, settings.alipay_alipay_public_cert_path, settings.alipay_root_cert_path))
|
||
|
||
|
||
def _load_certificate(path_value: str, label: str):
|
||
try:
|
||
from cryptography import x509
|
||
except ImportError as exc:
|
||
raise AlipayTransferError("证书模式依赖 cryptography，请安装后重启服务") from exc
|
||
if not path_value or not Path(path_value).is_file():
|
||
raise AlipayTransferError(f"{label}文件不存在，请检查对应 .env 绝对路径")
|
||
try:
|
||
return x509.load_pem_x509_certificate(Path(path_value).read_bytes())
|
||
except Exception as exc:
|
||
raise AlipayTransferError(f"{label}格式无效，应为 PEM 编码的 X.509 .crt 文件") from exc
|
||
|
||
|
||
def _load_certificate_bundle(path_value: str, label: str):
|
||
try:
|
||
from cryptography import x509
|
||
except ImportError as exc:
|
||
raise AlipayTransferError("证书模式依赖 cryptography，请安装后重启服务") from exc
|
||
if not path_value or not Path(path_value).is_file():
|
||
raise AlipayTransferError(f"{label}文件不存在，请检查对应 .env 绝对路径")
|
||
content = Path(path_value).read_bytes()
|
||
blocks = re.findall(rb"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----", content, re.DOTALL)
|
||
certificates = []
|
||
parse_errors = []
|
||
for block in blocks:
|
||
try:
|
||
certificates.append(x509.load_pem_x509_certificate(block))
|
||
except (TypeError, ValueError) as exc:
|
||
# Alipay's root bundle contains RSA, SM2 and ECC certificates.
|
||
# Some cryptography/OpenSSL builds cannot decode the SM2
|
||
# certificate even though the RSA certificates needed for RSA2
|
||
# are valid. Parse each PEM independently so one unsupported
|
||
# non-RSA certificate does not prevent the service from starting.
|
||
parse_errors.append(exc)
|
||
if not certificates:
|
||
cause = parse_errors[0] if parse_errors else None
|
||
raise AlipayTransferError(
|
||
f"{label}格式无效，应为包含有效 RSA 证书的 PEM X.509 证书集合"
|
||
) from cause
|
||
return certificates
|
||
|
||
|
||
def _certificate_serial(certificate) -> str:
|
||
issuer = certificate.issuer.rfc4514_string()
|
||
# Match AntCertificationUtil.getCertSN in Alipay's official Java SDK:
|
||
# MD5(issuer X500 name + the certificate serial number in *decimal*).
|
||
# Using hexadecimal here makes app_cert_sn/alipay_root_cert_sn invalid.
|
||
serial = str(certificate.serial_number)
|
||
return hashlib.md5(f"{issuer}{serial}".encode("utf-8")).hexdigest()
|
||
|
||
|
||
def _root_certificate_serial(certificates) -> str:
|
||
# Alipay's root bundle can contain RSA, SM2 and ECC certificates. The
|
||
# OpenAPI root certificate SN is defined from the RSA certificates only;
|
||
# including every certificate produces isv.invalid-alipay-root-cert-sn.
|
||
serials = [
|
||
_certificate_serial(certificate)
|
||
for certificate in certificates
|
||
# This is the exact RSA signature OID prefix check used by Alipay's
|
||
# AntCertificationUtil.getRootCertSN implementation.
|
||
if certificate.signature_algorithm_oid.dotted_string.startswith("1.2.840.113549.1.1")
|
||
]
|
||
if not serials:
|
||
raise AlipayTransferError("支付宝根证书中没有可用于计算序列号的 RSA 证书")
|
||
return "_".join(serials)
|
||
|
||
|
||
def _certificate_public_key(certificate) -> str:
|
||
from cryptography.hazmat.primitives import serialization
|
||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||
key = certificate.public_key()
|
||
if not isinstance(key, rsa.RSAPublicKey):
|
||
raise AlipayTransferError("支付宝证书必须包含 RSA 公钥")
|
||
return key.public_bytes(
|
||
encoding=serialization.Encoding.PEM,
|
||
format=serialization.PublicFormat.PKCS1,
|
||
).decode("ascii")
|
||
|
||
|
||
def _same_rsa_key(private_key: str, public_key: str) -> bool:
|
||
from Crypto.PublicKey import RSA
|
||
private = RSA.import_key(private_key)
|
||
public = RSA.import_key(public_key)
|
||
private_public = private.publickey()
|
||
return (private_public.n, private_public.e) == (public.n, public.e)
|
||
|
||
|
||
def _read_key(path_value: str, label: str) -> str:
|
||
if not path_value:
|
||
raise AlipayTransferError(f"未配置{label}文件路径")
|
||
path = Path(path_value)
|
||
if not path.is_file():
|
||
raise AlipayTransferError(f"{label}文件不存在：{path}")
|
||
value = path.read_text(encoding="utf-8-sig").strip()
|
||
if not value:
|
||
raise AlipayTransferError(f"{label}文件为空")
|
||
return value
|
||
|
||
|
||
def _sdk_key(value: str, *, private: bool) -> str:
|
||
"""Return a PKCS#1 PEM accepted by the SDK's rsa dependency."""
|
||
try:
|
||
from Crypto.PublicKey import RSA
|
||
normalized = value.strip()
|
||
if "-----BEGIN" not in normalized:
|
||
marker = "PRIVATE KEY" if private else "PUBLIC KEY"
|
||
normalized = f"-----BEGIN {marker}-----\n{normalized}\n-----END {marker}-----"
|
||
key = RSA.import_key(normalized.encode("utf-8"))
|
||
if private and not key.has_private():
|
||
raise ValueError("not a private key")
|
||
if private:
|
||
return key.export_key(format="PEM", pkcs=1).decode("ascii")
|
||
return key.publickey().export_key(format="PEM", pkcs=1).decode("ascii")
|
||
except Exception as exc:
|
||
raise AlipayTransferError("支付宝密钥格式无效，请确认上传的是完整 RSA2 密钥") from exc
|
||
|
||
|
||
def _client():
|
||
try:
|
||
from alipay.aop.api.AlipayClientConfig import AlipayClientConfig
|
||
from alipay.aop.api.DefaultAlipayClient import DefaultAlipayClient
|
||
except ImportError as exc:
|
||
raise AlipayTransferError("未安装支付宝 SDK，请安装 alipay-sdk-python") from exc
|
||
config = AlipayClientConfig()
|
||
config.server_url = settings.alipay_gateway
|
||
config.app_id = settings.alipay_app_id
|
||
config.app_private_key = _sdk_key(_read_key(settings.alipay_app_private_key_path, "应用私钥"), private=True)
|
||
if _certificate_mode_enabled():
|
||
app_cert = _load_certificate(settings.alipay_app_cert_path, "应用公钥证书")
|
||
alipay_cert = _load_certificate(settings.alipay_alipay_public_cert_path, "支付宝公钥证书")
|
||
root_certs = _load_certificate_bundle(settings.alipay_root_cert_path, "支付宝根证书")
|
||
config.alipay_public_key = _certificate_public_key(alipay_cert)
|
||
app_cert_sn = _certificate_serial(app_cert)
|
||
alipay_root_cert_sn = _root_certificate_serial(root_certs)
|
||
|
||
class CertificateAlipayClient(DefaultAlipayClient):
|
||
def _DefaultAlipayClient__get_common_params(self, params):
|
||
common = super()._DefaultAlipayClient__get_common_params(params)
|
||
common["app_cert_sn"] = app_cert_sn
|
||
common["alipay_root_cert_sn"] = alipay_root_cert_sn
|
||
return common
|
||
|
||
client_class = CertificateAlipayClient
|
||
else:
|
||
config.alipay_public_key = _sdk_key(_read_key(settings.alipay_public_key_path, "支付宝公钥"), private=False)
|
||
client_class = DefaultAlipayClient
|
||
config.sign_type = settings.alipay_sign_type
|
||
config.charset = settings.alipay_charset
|
||
config.format = settings.alipay_format
|
||
config.timeout = settings.alipay_http_timeout_seconds
|
||
return client_class(alipay_client_config=config)
|
||
|
||
|
||
def _parse_response(raw, response_cls):
|
||
"""Parse the JSON string returned by DefaultAlipayClient.execute."""
|
||
if isinstance(raw, response_cls):
|
||
response = raw
|
||
else:
|
||
response = response_cls()
|
||
try:
|
||
response.parse_response_content(raw.decode("utf-8") if isinstance(raw, bytes) else raw)
|
||
except Exception as exc:
|
||
raise AlipayTransferError("支付宝返回内容无法解析", ambiguous=True) from exc
|
||
if response.code != "10000" or response.sub_code:
|
||
raise AlipayTransferError(
|
||
response.sub_msg or response.msg or "支付宝请求失败",
|
||
response.sub_code or response.code or "ALIPAY_ERROR",
|
||
)
|
||
return response
|
||
|
||
|
||
def _provider_error_from_exception(exc: Exception) -> AlipayTransferError | None:
|
||
"""Recover Alipay's structured business error from old SDK exceptions.
|
||
|
||
alipay-sdk-python 3.x raises ResponseException before returning an error
|
||
response when the envelope contains ``alipay_cert_sn``. The exception
|
||
still embeds the complete JSON response, so preserve the provider's
|
||
sub-code/message instead of treating a deterministic rejection as an
|
||
ambiguous network failure.
|
||
"""
|
||
message = str(exc)
|
||
json_start = message.find("{")
|
||
if json_start < 0:
|
||
return None
|
||
try:
|
||
payload = json.loads(message[json_start:])
|
||
except (TypeError, ValueError):
|
||
return None
|
||
response = next((
|
||
value for key, value in payload.items()
|
||
if key.endswith("_response") and isinstance(value, dict)
|
||
), None)
|
||
if not response or not response.get("code"):
|
||
return None
|
||
return AlipayTransferError(
|
||
response.get("sub_msg") or response.get("msg") or "支付宝请求失败",
|
||
response.get("sub_code") or response.get("code") or "ALIPAY_ERROR",
|
||
ambiguous=False,
|
||
)
|
||
|
||
|
||
def transfer(*, out_biz_no: str, amount_fen: int, account: str, payee_name: str, order_title: str) -> dict:
|
||
from alipay.aop.api.domain.AlipayFundTransUniTransferModel import AlipayFundTransUniTransferModel
|
||
from alipay.aop.api.domain.Participant import Participant
|
||
from alipay.aop.api.domain.TransferSceneReportInfo import TransferSceneReportInfo
|
||
from alipay.aop.api.request.AlipayFundTransUniTransferRequest import AlipayFundTransUniTransferRequest
|
||
|
||
model = AlipayFundTransUniTransferModel()
|
||
model.out_biz_no = out_biz_no
|
||
model.trans_amount = f"{Decimal(amount_fen) / Decimal(100):.2f}"
|
||
model.product_code = "TRANS_ACCOUNT_NO_PWD"
|
||
model.biz_scene = "DIRECT_TRANSFER"
|
||
model.transfer_scene_name = settings.alipay_transfer_scene_name
|
||
job_type = TransferSceneReportInfo()
|
||
job_type.info_type = "岗位类型"
|
||
job_type.info_content = settings.alipay_transfer_job_type[:32]
|
||
reward_description = TransferSceneReportInfo()
|
||
reward_description.info_type = "报酬说明"
|
||
reward_description.info_content = order_title[:32]
|
||
model.transfer_scene_report_infos = [job_type, reward_description]
|
||
model.order_title = order_title[:128]
|
||
payee = Participant()
|
||
payee.identity = account
|
||
payee.identity_type = "ALIPAY_LOGON_ID"
|
||
payee.name = payee_name
|
||
model.payee_info = payee
|
||
try:
|
||
request = AlipayFundTransUniTransferRequest(biz_model=model)
|
||
if settings.alipay_notify_url:
|
||
request.notify_url = settings.alipay_notify_url
|
||
raw = _client().execute(request)
|
||
from alipay.aop.api.response.AlipayFundTransUniTransferResponse import AlipayFundTransUniTransferResponse
|
||
response = _parse_response(raw, AlipayFundTransUniTransferResponse)
|
||
except Exception as exc:
|
||
if isinstance(exc, AlipayTransferError):
|
||
raise
|
||
provider_error = _provider_error_from_exception(exc)
|
||
if provider_error:
|
||
raise provider_error from exc
|
||
raise AlipayTransferError(f"支付宝转账请求失败：{exc}", ambiguous=True) from exc
|
||
return {
|
||
"out_biz_no": response.out_biz_no or out_biz_no,
|
||
"trade_no": response.pay_fund_order_id or response.order_id or "",
|
||
"status": response.status or "UNKNOWN",
|
||
}
|
||
|
||
|
||
def query(*, out_biz_no: str) -> dict:
|
||
from alipay.aop.api.domain.AlipayFundTransCommonQueryModel import AlipayFundTransCommonQueryModel
|
||
from alipay.aop.api.request.AlipayFundTransCommonQueryRequest import AlipayFundTransCommonQueryRequest
|
||
|
||
model = AlipayFundTransCommonQueryModel()
|
||
model.out_biz_no = out_biz_no
|
||
model.product_code = "TRANS_ACCOUNT_NO_PWD"
|
||
model.biz_scene = "DIRECT_TRANSFER"
|
||
request = AlipayFundTransCommonQueryRequest(biz_model=model)
|
||
try:
|
||
raw = _client().execute(request)
|
||
from alipay.aop.api.response.AlipayFundTransCommonQueryResponse import AlipayFundTransCommonQueryResponse
|
||
response = _parse_response(raw, AlipayFundTransCommonQueryResponse)
|
||
except Exception as exc:
|
||
if isinstance(exc, AlipayTransferError):
|
||
raise
|
||
provider_error = _provider_error_from_exception(exc)
|
||
if provider_error:
|
||
raise provider_error from exc
|
||
raise AlipayTransferError(f"支付宝查询请求失败：{exc}") from exc
|
||
return {
|
||
"out_biz_no": response.out_biz_no or out_biz_no,
|
||
"trade_no": response.pay_fund_order_id or response.order_id or "",
|
||
"status": response.status or response.sub_status or "",
|
||
"fail_reason": response.fail_reason or response.sub_order_fail_reason or response.sub_msg or "",
|
||
}
|
||
|
||
|
||
def verify_notification(params: dict[str, str]) -> bool:
|
||
from alipay.aop.api.util.SignatureUtils import verify_with_rsa
|
||
|
||
signature = params.get("sign")
|
||
if not signature or params.get("sign_type", "RSA2").upper() != "RSA2":
|
||
return False
|
||
if settings.alipay_app_id and params.get("app_id") not in {None, settings.alipay_app_id}:
|
||
return False
|
||
content = "&".join(
|
||
f"{key}={params[key]}" for key in sorted(params)
|
||
if key not in {"sign", "sign_type"}
|
||
)
|
||
try:
|
||
alipay_public_key = (
|
||
_certificate_public_key(_load_certificate(settings.alipay_alipay_public_cert_path, "支付宝公钥证书"))
|
||
if _certificate_mode_enabled()
|
||
else _sdk_key(_read_key(settings.alipay_public_key_path, "支付宝公钥"), private=False)
|
||
)
|
||
return verify_with_rsa(
|
||
alipay_public_key,
|
||
content.encode(settings.alipay_charset),
|
||
signature,
|
||
)
|
||
except Exception:
|
||
return False
|