Django/utils/wechat_v3.py

import logging
import re
import time
import hashlib
import base64
import json
import os
from urllib.parse import urlparse

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

from django.conf import settings

logger = logging.getLogger(__name__)


def _get_v3_config(club_id=None):
    from jituan.services.wechat_pay import get_wechat_v3_config
    return get_wechat_v3_config(club_id)


def load_private_key(club_id=None):
    """加载私钥（支持 PKCS#1 和 PKCS#8）。"""
    cfg = _get_v3_config(club_id)
    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, club_id=None):
    """生成 V3 接口的 Authorization 头。"""
    cfg = _get_v3_config(club_id)
    private_key = load_private_key(club_id)
    mchid = cfg['MCHID']
    serial_no = (cfg.get('CERT_SERIAL_NO') or '').strip().upper()
    logger.warning(
        '[WX_V3_SIGN] club=%s mchid=%s serial=%s key_path=%s',
        club_id, mchid, serial_no, cfg.get('PRIVATE_KEY_PATH'),
    )

    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]) + '\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 verify_wechat_sign(headers, body, club_id=None):
    """验证微信回调签名（需提前下载平台证书）。"""
    import 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

    if not re.match(r'^[A-Za-z0-9-]+$', serial):
        return False

    try:
        ts = int(headers.get('Wechatpay-Timestamp', '0'))
        if abs(time.time() - ts) > 300:
            return False
    except (ValueError, TypeError):
        return False

    cfg = _get_v3_config(club_id)
    message = '\n'.join([timestamp, nonce, body]) + '\n'
    cert_path = f"{cfg['PLATFORM_CERT_DIR']}/{serial}.pem"
    if not os.path.exists(cert_path):
        return False

    with open(cert_path, 'rb') as f:
        pub_key = rsa.PublicKey.load_pkcs1_openssl_pem(f.read())

    signature_bytes = base64.b64decode(signature)
    try:
        rsa.verify(message.encode('utf-8'), signature_bytes, pub_key)
        return True
    except rsa.VerificationError:
        return False


def resolve_callback_club_id(headers, body):
    """
    多商户回调：依次用各俱乐部平台证书验签，返回匹配的 club_id。
    验签失败返回 None。
    """
    from jituan.constants import CLUB_ID_DEFAULT
    from jituan.models import Club

    if verify_wechat_sign(headers, body, club_id=CLUB_ID_DEFAULT):
        return CLUB_ID_DEFAULT

    for club in Club.query.filter(status=1).exclude(club_id=CLUB_ID_DEFAULT):
        if verify_wechat_sign(headers, body, club_id=club.club_id):
            return club.club_id
    return None


def decrypt_callback_ciphertext(associated_data, nonce, ciphertext, club_id=None):
    """解密回调数据（使用 AES-GCM）。"""
    cfg = _get_v3_config(club_id)
    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')