微信支付 jsapi v3 簽名 和 驗簽

import json
import time
import random
import string
import base64
# 創建訂單
import pprint

import requests
from Cryptodome import Hash
from Cryptodome.PublicKey import RSA
from Cryptodome.Signature import PKCS1_v1_5
from Cryptodome.Hash import SHA256
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

import config
from app import logger


# 簽名驗簽相關
class ToSign:
    # app/handler/wechat_app/
    try:
        with open("app/handler/wechat_app/apiclient_key.pem", 'r') as f:
            private_key = f.read()
    except Exception:
        with open("apiclient_key.pem", 'r') as f:
            private_key = f.read()


    @classmethod
    def set_default(cls):
        cls.timestamp = "%.f" % time.time()  # "%.f" % time.time()  # 時間戳
        cls.nonce_str = "".join(random.sample(string.ascii_letters + string.digits, 16))  # 隨機字符串

    @classmethod
    def set_sign_data(cls, method: str, url: str, body: dict = None):
        """設置默認數據 """
        cls.method = method
        cls.url = url
        if body:
            cls.body = json.dumps(body)  # 轉換為json字符串
        else:
            cls.body = ""

    @classmethod
    def sign_str(cls):
        """生成欲簽名字符串"""
        return str("\n".join([cls.method, cls.url,
                              cls.timestamp, cls.nonce_str,
                              cls.body])+"\n")

    # 簽名
    @classmethod
    def sign(cls, sign_str):
        """簽名 """
        pkey = RSA.importKey(cls.private_key)
        h = SHA256.new(sign_str.encode('utf-8'))
        signature = PKCS1_v1_5.new(pkey).sign(h)
        sign = base64.b64encode(signature).decode()
        return sign

    # 獲取請求頭authorization
    @classmethod
    def authorization_str(cls):
        sign_ = cls.sign(cls.sign_str())
        """拼接header authorization"""
        authorization = 'WECHATPAY2-SHA256-RSA2048 ' \
                        'mchid="{mchid}",' \
                        'nonce_str="{nonce_str}",' \
                        'signature="{sign}",' \
                        'timestamp="{timestamp}",' \
                        'serial_no="{serial_no}"'. \
            format(mchid=config.mchid,
                   nonce_str=cls.nonce_str,
                   sign=sign_,
                   timestamp=cls.timestamp,
                   serial_no=config.serial_no
                   )
        return authorization

    # 驗簽
    @classmethod
    def check_sign(cls, plain_text: str, sign: str, certificate=None) -> bool:  # 明文、 密文
        # base64 解碼
        sign_str = base64.b64decode(sign)
        # 這里采用的是從接獲獲得的證書，微信支付證書
        signature2 = RSA.importKey(certificate)
        verifier = PKCS1_v1_5.new(signature2)
        digest = Hash.SHA256.new()
        digest.update(plain_text.encode("utf8"))
        return verifier.verify(digest, sign_str)


# 解密
def decrypt(nonce, ciphertext, associated_data):
    """
    解密數據獲取平台證書
    https://pay.weixin.qq.com/wiki/doc/apiv3/wechatpay/wechatpay5_1.shtml
    https://pay.weixin.qq.com/wiki/doc/apiv3/wechatpay/wechatpay4_2.shtml
    """
    key = config.Apiv3Key
    key_bytes = str.encode(key)
    nonce_bytes = str.encode(nonce)
    ad_bytes = str.encode(associated_data)
    data = base64.b64decode(ciphertext)
    aesgcm = AESGCM(key_bytes)
    # 解密出來的是加密字符串。取出你想要的數據
    return aesgcm.decrypt(nonce_bytes, data, ad_bytes)


# 獲取微信支付平台證書, 這個證書用於，驗簽 https://pay.weixin.qq.com/wiki/doc/apiv3/wechatpay/wechatpay4_1.shtml
def get_credential():
    # 獲取平台證書序列號
    ToSign.set_default()
    ToSign.set_sign_data("GET", "/v3/certificates")
    authorization = ToSign.authorization_str()
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
        "Authorization": authorization
    }
    res = requests.get("https://api.mch.weixin.qq.com/v3/certificates", headers=headers)
    if res.status_code != 200:
        logger.error("獲取獲取微信支付平台證書失敗")
        logger.error(res.json())
    plain_text, sign, res_data = get_attestation_about(res)

    # 進行解密, 獲取證書
    # 這里可能會有很多證書，選啟用時間最晚的一個
    result_data = max(res_data["data"], key=lambda x: x["effective_time"])
    nonce = result_data["encrypt_certificate"]["nonce"]
    ciphertext = result_data["encrypt_certificate"]["ciphertext"]
    associated_data = result_data["encrypt_certificate"]["associated_data"]
    # 明文串
    str_ = decrypt(nonce=nonce, ciphertext=ciphertext, associated_data=associated_data)
    check_bool = ToSign.check_sign(plain_text=plain_text, sign=sign, certificate=str_)
    if check_bool:
        return str_, result_data
    else:
        logger.error(f"獲取支付證書驗簽失敗,plain_text:{plain_text}, sign: {sign}")
        return "", {}


# 獲取眼前需要的數據
def get_attestation_about(response):
    time_stamp = response.headers.get("Wechatpay-Timestamp")
    nonce = response.headers.get("Wechatpay-Nonce")
    sign = response.headers.get("Wechatpay-Signature")
    logger.info("time_stamp:"+time_stamp)
    logger.info("nonce:"+nonce)
    logger.info("sign:"+sign)
    try:
        req_data = json.dumps(response.json()).replace(" ", "")
    except Exception:
        req_data = response.json
        req_data = json.dumps(req_data, ensure_ascii=False).replace(" ", "")
    plain_text = f"{time_stamp}\n{nonce}\n{req_data}\n"
    return plain_text, sign, json.loads(req_data)  # 加密字符串， 簽名, body->json

調用

# 簽名
ToSign.set_default()
ToSign.set_sign_data("POST", "/v3/pay/transactions/jsapi", data)

authorization_str = ToSign.authorization_str()
headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
        "Authorization": authorization_str
    }
res_data = requests.post(url, json=data, headers=headers)
if 'prepay_id' not in res_data.json():
    return {"success": False, "message": "調取微信支付失敗"}

# 驗簽  https://pay.weixin.qq.com/wiki/doc/apiv3/wechatpay/wechatpay5_1.shtml
"""
1554209980
c5ac7061fccab6bf3e254dcf98995b8c
{"data":[{"serial_no":"5157F09EFDC096DE15EBE81A47057A7232F1B8E1","effective_time":"2018-03-26T11:39:50+08:00","expire_time":"2023-03-25T11:39:50+08:00","encrypt_certificate":{"algorithm":"AEAD_AES_256_GCM","nonce":"d215b0511e9c","associated_data":"certificate","ciphertext":"..."}}]}
"""
ToSign.check_sign(ToSign.sign_str(), sign) # 第一個參數是根據微信返回信息生成的簽名串， 第二個是你從微信得到的簽名


# 

注意：
1.調用ToSign 的時候先調用set_default來進行初始化時間戳和隨機字符串
2. json化是必須加ensure_ascii，否則會進行編碼導致驗簽失敗。