153 lines
5.0 KiB
Python
153 lines
5.0 KiB
Python
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)
|