along_django/utils/wechat_v3.py

# utils/wechat_v3.py
"""
微信支付 V3 签名 / 验签 / 回调解密。
支持传入商户配置 dict；未传时使用 wechat_mch_service 默认配置（表优先，settings 兜底）。
"""
import base64
import hashlib
import logging
import os
import time
from urllib.parse import urlparse

from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from Crypto.Cipher import AES

logger = logging.getLogger('utils.wechat_v3')


def _load_platform_public_key(pem_bytes):
    """加载微信平台证书或公钥 PEM（兼容 CERTIFICATE / PUBLIC KEY / RSA PUBLIC KEY）。"""
    if b'BEGIN CERTIFICATE' in pem_bytes:
        return x509.load_pem_x509_certificate(pem_bytes, default_backend()).public_key()
    return serialization.load_pem_public_key(pem_bytes, backend=default_backend())


def _resolve_cfg(wx_cfg=None):
    if wx_cfg:
        return wx_cfg
    from utils.wechat_mch_service import get_default_wx_cfg
    cfg = get_default_wx_cfg()
    if not cfg:
        raise RuntimeError('无可用微信商户配置')
    return cfg


def load_private_key(wx_cfg=None):
    """加载私钥（支持 PKCS#1 和 PKCS#8）"""
    cfg = _resolve_cfg(wx_cfg)
    key_path = cfg['PRIVATE_KEY_PATH']
    with open(key_path, 'rb') as f:
        key_data = f.read()
    return serialization.load_pem_private_key(
        key_data,
        password=None,
        backend=default_backend(),
    )


def build_authorization(method, url, body, wx_cfg=None):
    """生成V3接口的Authorization头；wx_cfg 指定商户签名配置。"""
    cfg = _resolve_cfg(wx_cfg)
    private_key = load_private_key(cfg)
    mchid = cfg['MCHID']
    serial_no = cfg['CERT_SERIAL_NO']

    parsed = urlparse(url)
    path = parsed.path
    if parsed.query:
        path += '?' + parsed.query

    timestamp = str(int(time.time()))
    nonce = hashlib.md5((timestamp + 'wechat').encode()).hexdigest()[:32]
    message = '\n'.join([method, path, timestamp, nonce, body or '']) + '\n'

    signature = private_key.sign(
        message.encode('utf-8'),
        padding.PKCS1v15(),
        hashes.SHA256(),
    )
    signature_b64 = base64.b64encode(signature).decode('utf-8')
    return (
        f'WECHATPAY2-SHA256-RSA2048 mchid="{mchid}",nonce_str="{nonce}",'
        f'timestamp="{timestamp}",serial_no="{serial_no}",signature="{signature_b64}"'
    )


def _resolve_platform_pub_path(wx_cfg, serial=None):
    """兼容：既可是目录/{serial}.pem，也可是直接指向公钥文件路径。"""
    base = (wx_cfg or {}).get('PLATFORM_PUB_KEY_PATH') or (wx_cfg or {}).get('PLATFORM_CERT_DIR') or ''
    base = str(base).strip()
    if not base:
        return None
    if os.path.isfile(base):
        return base
    if serial:
        candidate = os.path.join(base, f'{serial}.pem')
        if os.path.isfile(candidate):
            return candidate
    return None


def verify_wechat_sign(headers, body, wx_cfg=None):
    """验证微信回调签名（用 cryptography，不依赖 rsa 包）。可指定商户或试全部。"""
    serial = headers.get('Wechatpay-Serial')
    signature = headers.get('Wechatpay-Signature')
    timestamp = headers.get('Wechatpay-Timestamp')
    nonce = headers.get('Wechatpay-Nonce')
    if not all([serial, signature, timestamp, nonce]):
        return False

    message = '\n'.join([timestamp, nonce, body]) + '\n'
    signature_bytes = base64.b64decode(signature)

    cfgs = []
    if wx_cfg:
        cfgs = [wx_cfg]
    else:
        from utils.wechat_mch_service import list_wx_cfgs_for_callback
        cfgs = list_wx_cfgs_for_callback()

    for cfg in cfgs:
        cert_path = _resolve_platform_pub_path(cfg, serial=serial)
        if not cert_path or not os.path.exists(cert_path):
            continue
        try:
            with open(cert_path, 'rb') as f:
                pub_key = _load_platform_public_key(f.read())
            pub_key.verify(
                signature_bytes,
                message.encode('utf-8'),
                padding.PKCS1v15(),
                hashes.SHA256(),
            )
            return True
        except Exception as e:
            logger.debug('回调验签未通过 path=%s err=%s', cert_path, e)
            continue
    return False


def decrypt_callback_ciphertext(associated_data, nonce, ciphertext, wx_cfg=None):
    """用指定商户 APIv3 密钥解密回调。"""
    cfg = _resolve_cfg(wx_cfg)
    api_v3_key = cfg['API_V3_KEY'].encode('utf-8')
    nonce_bytes = nonce.encode('utf-8')
    ciphertext_bytes = base64.b64decode(ciphertext)

    cipher = AES.new(api_v3_key, AES.MODE_GCM, nonce=nonce_bytes)
    if associated_data:
        cipher.update(associated_data.encode('utf-8'))
    plaintext = cipher.decrypt_and_verify(ciphertext_bytes[:-16], ciphertext_bytes[-16:])
    return plaintext.decode('utf-8')


def decrypt_callback_with_merchants(associated_data, nonce, ciphertext):
    """
    回调解密：因密文未解出前无法知道 out_bill_no，
    仅对启用商户密钥做 AES 尝试（通常很少），成功后返回 (plaintext, wx_cfg)。
    业务侧再用 out_bill_no 反查记录上的 mch_id，对齐一致性。
    """
    from utils.wechat_mch_service import list_wx_cfgs_for_callback

    last_err = None
    for cfg in list_wx_cfgs_for_callback():
        try:
            text = decrypt_callback_ciphertext(associated_data, nonce, ciphertext, wx_cfg=cfg)
            return text, cfg
        except Exception as e:
            last_err = e
            continue
    raise ValueError(f'所有商户密钥均无法解密回调: {last_err}')