霖雨寺

app/wechat_pay.py

@@ -0,0 +1,152 @@
+import base64
+import json
+import secrets
+import time
+from pathlib import Path
+from typing import Any
+from urllib import request as urllib_request
+from urllib.error import HTTPError, URLError
+
+from fastapi import HTTPException, status
+
+from app.config import wechat_pay_settings
+
+
+WECHAT_SIGNATURE_TYPE = "WECHATPAY2-SHA256-RSA2048"
+
+
+def ensure_wechat_pay_configured() -> None:
+    if not wechat_pay_settings.is_configured:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="微信支付配置未完成，请先填写 WECHAT_PAY_* 配置。",
+        )
+
+
+def load_private_key():
+    from cryptography.hazmat.primitives import serialization
+
+    key_data = Path(wechat_pay_settings.private_key_path).read_bytes()
+    return serialization.load_pem_private_key(key_data, password=None)
+
+
+def load_public_key():
+    from cryptography.hazmat.primitives import serialization
+
+    if not wechat_pay_settings.public_key_path:
+        return None
+    key_data = Path(wechat_pay_settings.public_key_path).read_bytes()
+    return serialization.load_pem_public_key(key_data)
+
+
+def rsa_sign(message: str) -> str:
+    from cryptography.hazmat.primitives import hashes
+    from cryptography.hazmat.primitives.asymmetric import padding
+
+    signature = load_private_key().sign(
+        message.encode("utf-8"),
+        padding.PKCS1v15(),
+        hashes.SHA256(),
+    )
+    return base64.b64encode(signature).decode("ascii")
+
+
+def rsa_verify(message: str, signature: str) -> bool:
+    from cryptography.exceptions import InvalidSignature
+    from cryptography.hazmat.primitives import hashes
+    from cryptography.hazmat.primitives.asymmetric import padding
+
+    public_key = load_public_key()
+    if public_key is None:
+        return bool(wechat_pay_settings.skip_notify_verify)
+
+    try:
+        public_key.verify(
+            base64.b64decode(signature),
+            message.encode("utf-8"),
+            padding.PKCS1v15(),
+            hashes.SHA256(),
+        )
+    except InvalidSignature:
+        return False
+    return True
+
+
+def build_authorization(method: str, url_path: str, body: str) -> str:
+    timestamp = str(int(time.time()))
+    nonce = secrets.token_hex(16)
+    message = f"{method}\n{url_path}\n{timestamp}\n{nonce}\n{body}\n"
+    signature = rsa_sign(message)
+    return (
+        f'{WECHAT_SIGNATURE_TYPE} mchid="{wechat_pay_settings.mch_id}",'
+        f'nonce_str="{nonce}",signature="{signature}",'
+        f'timestamp="{timestamp}",serial_no="{wechat_pay_settings.mch_serial_no}"'
+    )
+
+
+def post_wechat_json(url_path: str, payload: dict[str, Any]) -> dict[str, Any]:
+    ensure_wechat_pay_configured()
+    body = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
+    url = f"{wechat_pay_settings.api_base_url.rstrip('/')}{url_path}"
+    request = urllib_request.Request(
+        url,
+        data=body.encode("utf-8"),
+        method="POST",
+        headers={
+            "Accept": "application/json",
+            "Authorization": build_authorization("POST", url_path, body),
+            "Content-Type": "application/json",
+            "User-Agent": "py-web-wechat-pay/0.1",
+        },
+    )
+
+    try:
+        with urllib_request.urlopen(request, timeout=10) as response:
+            return json.loads(response.read().decode("utf-8"))
+    except HTTPError as exc:
+        error_body = exc.read().decode("utf-8")
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"微信支付请求失败：{error_body or exc.reason}",
+        ) from exc
+    except URLError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"微信支付网络请求失败：{exc.reason}",
+        ) from exc
+
+
+def build_jsapi_pay_params(prepay_id: str) -> dict[str, str]:
+    timestamp = str(int(time.time()))
+    nonce = secrets.token_hex(16)
+    package = f"prepay_id={prepay_id}"
+    message = f"{wechat_pay_settings.appid}\n{timestamp}\n{nonce}\n{package}\n"
+    return {
+        "appId": wechat_pay_settings.appid,
+        "timeStamp": timestamp,
+        "nonceStr": nonce,
+        "package": package,
+        "signType": "RSA",
+        "paySign": rsa_sign(message),
+    }
+
+
+def decrypt_notify_resource(resource: dict[str, str]) -> dict[str, Any]:
+    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+
+    ciphertext = base64.b64decode(resource["ciphertext"])
+    nonce = resource["nonce"].encode("utf-8")
+    associated_data = resource.get("associated_data", "").encode("utf-8")
+    aesgcm = AESGCM(wechat_pay_settings.api_v3_key.encode("utf-8"))
+    plaintext = aesgcm.decrypt(nonce, ciphertext, associated_data)
+    return json.loads(plaintext.decode("utf-8"))
+
+
+def verify_notify_signature(headers: dict[str, str], body: str) -> bool:
+    timestamp = headers.get("wechatpay-timestamp", "")
+    nonce = headers.get("wechatpay-nonce", "")
+    signature = headers.get("wechatpay-signature", "")
+    if not timestamp or not nonce or not signature:
+        return False
+    message = f"{timestamp}\n{nonce}\n{body}\n"
+    return rsa_verify(message, signature)