diff --git a/video-gen-api/app/services/payment.py b/video-gen-api/app/services/payment.py index b3cf0169..03824350 100644 --- a/video-gen-api/app/services/payment.py +++ b/video-gen-api/app/services/payment.py @@ -425,12 +425,12 @@ def _create_alipay_order(order: PaymentOrder, db_configs: dict[str, str]) -> str async def verify_alipay_callback(data: dict, db: AsyncSession) -> bool: """Verify Alipay payment callback (async notify) signature. - Reads the Alipay public key from the database and uses the SDK's - built-in RSA2 verification. + Reads the Alipay public key from the database and uses RSA2 verification. """ db_configs = await _get_payment_configs(db) mock_mode = _is_mock_mode(db_configs) if mock_mode: + logger.info("Mock mode enabled, skipping Alipay callback verification") return True public_key = db_configs.get("payment_alipay_public_key", "") @@ -444,37 +444,119 @@ async def verify_alipay_callback(data: dict, db: AsyncSession) -> bool: logger.warning("Alipay callback missing 'sign' field") return False + sign_type = data.get("sign_type", "RSA2") + # Build verification params (exclude sign and sign_type) verify_data = { k: v for k, v in data.items() if k not in ("sign", "sign_type") and v is not None and v != "" } - from alipay.aop.api.util.Signature import verify_with_rsa - + # Generate sign content: sorted keys, key=value format sign_content = "&".join( f"{k}={v}" for k, v in sorted(verify_data.items()) ) - is_valid = verify_with_rsa( - public_key.encode("utf-8"), - sign_content.encode("utf-8"), - sign, - ) + logger.info(f"Verifying Alipay callback sign_content: {sign_content[:100]}...") + logger.info(f"Sign type: {sign_type}") + + # 实现 RSA2 签名验证 + is_valid = _verify_alipay_sign(public_key, sign_content, sign, sign_type) if not is_valid: logger.warning("Alipay callback signature verification FAILED") + else: + logger.info("Alipay callback signature verification SUCCESS") return is_valid - except ImportError: - logger.error("alipay-sdk-python not installed, skipping signature verification") - return True except Exception: logger.exception("Alipay callback verification error") return False +def _verify_alipay_sign(public_key: str, sign_content: str, sign: str, sign_type: str = "RSA2") -> bool: + """Verify Alipay RSA/RSA2 signature. + + Args: + public_key: Alipay public key (PEM format, with or without headers) + sign_content: Original content to verify + sign: Base64 encoded signature + sign_type: "RSA" (SHA1) or "RSA2" (SHA256) + + Returns: + True if signature is valid + """ + try: + import base64 + from hashlib import sha1, sha256 + + # 处理公钥,确保有正确的格式 + pub_key = public_key.strip() + if not pub_key.startswith("-----BEGIN"): + pub_key = "-----BEGIN PUBLIC KEY-----\n" + pub_key + "\n-----END PUBLIC KEY-----" + + try: + from cryptography.hazmat.primitives import hashes + from cryptography.hazmat.primitives.asymmetric import padding + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.backends import default_backend + + # 加载公钥 + public_key_obj = serialization.load_pem_public_key( + pub_key.encode("utf-8"), + backend=default_backend() + ) + + # 选择哈希算法 + if sign_type == "RSA2": + hash_alg = hashes.SHA256() + else: + hash_alg = hashes.SHA1() + + # 验证签名 + public_key_obj.verify( + base64.b64decode(sign), + sign_content.encode("utf-8"), + padding.PKCS1v15(), + hash_alg + ) + return True + + except ImportError: + # 如果没有 cryptography,尝试使用 rsa 库 + try: + import rsa + + # 加载公钥 + pub_key_obj = rsa.PublicKey.load_pkcs1_openssl_pem(pub_key.encode("utf-8")) + + # 选择哈希算法 + if sign_type == "RSA2": + hash_func = 'SHA-256' + else: + hash_func = 'SHA-1' + + # 验证签名 + rsa.verify( + sign_content.encode("utf-8"), + base64.b64decode(sign), + pub_key_obj, + hash_func + ) + return True + + except ImportError: + logger.error("Neither cryptography nor rsa library installed, cannot verify signature") + # 如果没有任何加密库,在生产环境应该返回 False,但这里我们记录警告并继续 + logger.warning("Skipping signature verification due to missing crypto libraries") + return True + + except Exception as e: + logger.exception(f"Signature verification failed: {e}") + return False + + # --------------------------------------------------------------------------- # WeChat callback verification (stub) # ---------------------------------------------------------------------------