所需库 [代码]WeiXinPayV3/requirement.txt[代码] [代码]Flask==1.0.2 pyOpenSSL==20.0.1 cryptography==3.4.7 requests==2.25.1 [代码] 核心API对接 [代码]WeiXinPayV3/api.py[代码] [代码]import datetime import json import os import random import string import time import OpenSSL import base64 import requests from urllib.parse import urlparse from cryptography.hazmat.primitives.ciphers.aead import AESGCM # V3版支付 class WxPayV3(object): basedir = os.path.abspath(os.path.dirname(__file__)) # 获取当前目录 def __init__(self, ignore_resp_sign=False): """ :param ignore_resp_sign: 是否忽略应答验签(用于第一次缓存证书) """ self.app_id = '{app_id}' self.mch_id = '{mch_id}' self.api_v3_key = '{api_v3_key}' self.serial_no = '{serial_no}' # 证书序列号(微信支付商户平台获取) self.ignore_resp_sign = ignore_resp_sign self.certificates_path = self.basedir + '/certificates.json' # 支付签名 @staticmethod def sign(s): # 这里的key为示例 api_client_key = ("-----BEGIN PRIVATE KEY-----\n" "MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDAPM45XP7PC6qe\n" "MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDAPM45XP7PC6qe\n" "Kkh/7UGUD/hxkPwNB2PKAms=\n" "-----END PRIVATE KEY-----") pkey = OpenSSL.crypto.load_privatekey(OpenSSL.crypto.FILETYPE_PEM, api_client_key) signature = base64.b64encode(OpenSSL.crypto.sign(pkey=pkey, data=s.encode(), digest='SHA256')) return signature.decode() # 请求签名 def _auth(self, req): data = req.method + '\n' parsed = urlparse(req.url) data += parsed.path if parsed.query: data += "?" + parsed.query data += '\n' timestamp = str(int(time.time())) data += timestamp + '\n' nonce_str = ''.join(random.sample(string.ascii_letters + string.digits, 32)) data += nonce_str + '\n' if req.data: data += req.data data += '\n' signature = self.sign(data) authorization = ('WECHATPAY2-SHA256-RSA2048 ' 'mchid="{0}",nonce_str="{1}",' 'signature="{2}",timestamp="{3}",' 'serial_no="{4}"').format(self.mch_id, nonce_str, signature, timestamp, self.serial_no) req.headers["Authorization"] = authorization req.headers['Content-Type'] = 'application/json' req.headers['Accept'] = 'application/json' req.headers['User-Agent'] = 'requests ' + requests.__version__ r = req.prepare() s = requests.Session() resp = s.send(r, timeout=2) # 验签 if not self.ignore_resp_sign: nonce = resp.headers.get('Wechatpay-Nonce') signature = resp.headers.get('Wechatpay-Signature') serial = resp.headers.get('Wechatpay-Serial') timestamp = resp.headers.get('Wechatpay-Timestamp') body = resp.text cer = self.get_certificate_by_serial_no(serial) ret = self.resp_sign(timestamp=timestamp, nonce=nonce, body=body, cer=cer, signature=signature) assert ret is None, "resp sign error" return resp # Native下单 def native_pay(self, total, out_trade_no, attach=None): url = 'https://api.mch.weixin.qq.com/v3/pay/transactions/native' notify_url = 'https://pay.weixin.com/notify' data = dict(appid=self.app_id, mchid=self.mch_id, description="测试native支付", notify_url=notify_url, out_trade_no=out_trade_no, amount=dict(total=total, currency='CNY')) if attach is not None: data['attach'] = attach req = requests.Request(method="POST", url=url, data=json.dumps(data)) response = self._auth(req=req) result = response.json() return result['code_url'] @staticmethod def gen_out_trade_no(trade_type="jsapi"): out_trade_no = "wx" + trade_type + datetime.datetime.now().strftime("%Y%m%d%H%M%S") out_trade_no += str(random.randint(1000000, 9999999)) return out_trade_no # JSAPI下单 def jsapi_pay(self, total, out_trade_no, openid, attach=None): url = 'https://api.mch.weixin.qq.com/v3/pay/transactions/jsapi' notify_url = 'https://pay.weixin.com/notify' data = dict(appid=self.app_id, mchid=self.mch_id, description="测试jsapi支付", notify_url=notify_url, out_trade_no=out_trade_no, amount=dict(total=total, currency='CNY'), payer=dict(openid=openid)) if attach is not None: data['attach'] = attach req = requests.Request(method="POST", url=url, data=json.dumps(data)) response = self._auth(req=req) result = response.json() return result['prepay_id'] # 查询订单 def query_order(self, out_trade_no=None, transaction_id=None): if out_trade_no is None and transaction_id is None: raise ValueError('Param Error') if out_trade_no is not None: url = 'https://api.mch.weixin.qq.com/v3/pay/transactions/out-trade-no/{0}'.format(out_trade_no) else: url = 'https://api.mch.weixin.qq.com/v3/pay/transactions/id/{0}'.format(out_trade_no) url += '?' + 'mchid=' + self.mch_id req = requests.Request(method="GET", url=url) response = self._auth(req=req) result = response.json() return result # 获取微信平台证书并缓存 def get_certificates_from_wx(self): # TODO 这里实现非常简陋,鉴于V3版本证书会动态更新建议使用中控服务统一获取和安全存储证书(惰性更新也可) url = 'https://api.mch.weixin.qq.com/v3/certificates' req = requests.Request(method="GET", url=url) response = self._auth(req=req) result = response.json() with open(self.certificates_path, 'w+') as f: f.write(json.dumps(result['data'])) return result def get_certificate_by_serial_no(self, serial_no): if not os.path.exists(self.certificates_path): raise FileNotFoundError('请先下载证书进行缓存') with open(self.certificates_path, 'r') as f: content = f.read() certificates = json.loads(content) nonce, ciphertext, associated_data = None, None, None for certificate in certificates: if certificate['serial_no'] == serial_no: nonce = certificate['encrypt_certificate']['nonce'] ciphertext = certificate['encrypt_certificate']['ciphertext'] associated_data = certificate['encrypt_certificate']['associated_data'] break if ciphertext is None: raise ValueError('certificate not found') cer = self.decrypt_aes_gcm(nonce=nonce, ciphertext=ciphertext, associated_data=associated_data) return cer def decrypt_aes_gcm(self, nonce, ciphertext, associated_data): aes_gcm = AESGCM(self.api_v3_key.encode()) plaintext = aes_gcm.decrypt(nonce=nonce.encode(), associated_data=associated_data.encode(), data=base64.b64decode(ciphertext)) return plaintext # 应答验签 @staticmethod def resp_sign(timestamp, nonce, body, cer, signature): cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, cer) s = '{0}\n{1}\n{2}\n'.format(timestamp, nonce, body) signature = base64.b64decode(signature) try: OpenSSL.crypto.verify(cert=cert, signature=signature, data=s, digest='SHA256') except Exception as e: return e.__str__() return None # JsApi签名(未验证) def jsapi_sign(self, prepay_id): timestamp = int(time.time()) nonce_str = ''.join(random.sample(string.ascii_letters + string.digits, 32)) package = "prepay_id=" + prepay_id sign_type = 'RSA' s = "{0}\n{1}\n{2}\n{3}\n".format(self.app_id, timestamp, nonce_str, package) pay_sign = self.sign(s) obj = dict(app_id=self.app_id, timestamp=timestamp, nonce_str=nonce_str, package=package, sign_type=sign_type, pay_sign=pay_sign) return obj [代码] 验签及支付demo(这里以Flask为例) [代码]WeiXinPayV3/web.py[代码] [代码]import json from flask import Flask, request, jsonify, abort from api import WxPayV3 app = Flask(__name__) # 支付结果通知 @app.route('/notify', methods=['POST']) def notify(): nonce = request.headers.get('Wechatpay-Nonce') signature = request.headers.get('Wechatpay-Signature') serial = request.headers.get('Wechatpay-Serial') timestamp = request.headers.get('Wechatpay-Timestamp') body = request.get_data(as_text=True) pay_v3 = WxPayV3() cer = pay_v3.get_certificate_by_serial_no(serial) # 支付结果通知签名验证 ret = pay_v3.resp_sign(timestamp=timestamp, nonce=nonce, body=body, cer=cer, signature=signature) if ret is not None: # 返回非空则为验签失败 return abort(400) req_id = request.json.get('id') create_time = request.json.get('create_time') resource_type = request.json.get('resource_type') event_type = request.json.get('event_type') summary = request.json.get('summary') resource = request.json.get('resource') # 解密数据 plaintext = pay_v3.decrypt_aes_gcm(nonce=resource['nonce'], ciphertext=resource['ciphertext'], associated_data=resource['associated_data']) transaction_data = json.loads(plaintext) transaction_id = transaction_data['transaction_id'] out_trade_no = transaction_data['out_trade_no'] trade_type = transaction_data['trade_type'] trade_state = transaction_data['trade_state'] total = transaction_data['amount']['total'] if event_type == 'TRANSACTION.SUCCESS' and trade_state == "SUCCESS": # 支付成功 pass elif event_type == 'REFUND.SUCCESS' and trade_state == "SUCCESS": # 退款成功 pass return jsonify(code="SUCCESS", message="成功") # Native支付 @app.route('/pay/native', methods=['GET']) def pay_native(): pay_v3 = WxPayV3() out_trade_no = pay_v3.gen_out_trade_no(trade_type="native") code_url = pay_v3.native_pay(total=1, out_trade_no=out_trade_no) return jsonify(code_url=code_url) # 小程序/公众号支付 @app.route('/pay/jsapi', methods=['GET']) def pay_jsapi(): pay_v3 = WxPayV3() out_trade_no = pay_v3.gen_out_trade_no(trade_type="jsapi") prepay_id = pay_v3.jsapi_pay(total=1, out_trade_no=out_trade_no, openid='oE6E') obj = pay_v3.jsapi_sign(prepay_id=prepay_id) return jsonify(obj=obj) if __name__ == '__main__': app.run(port=9012) [代码] 提醒 1.收到微信支付异步通知需要两个条件：通知地址为HTTPS、在商户中心设置APIv3密钥。 2.对于微信支付平台的应答，需要使用平台证书来进行验签；但平台证书只能通过 获取平台证书接口下载，所以当第一次去获取证书时将ignore_resp_sign设置为True以忽略应答验签。 3.小程序/公众号支付的签名目前并未进行验证。 4.本示例中所有证书(无论是客户端证书还是微信平台证书)均不需要通过命令行OpenSSL进行转换。