Last active
August 13, 2026 03:27
-
-
Save Wxh16144/08e20f0e8ac1402014f427d5e6c0bfd4 to your computer and use it in GitHub Desktop.
微信公众号 API 最小验证脚本
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| # -*- coding: utf-8 -*- | |
| """ | |
| 微信公众号 API 最小验证脚本 (单文件 / 零第三方依赖 / 直接运行) | |
| 功能: | |
| GET /health 健康检查: 返回 ok, 用于确认服务是否启动成功 | |
| GET /wechat 服务器配置验证: 校验 signature 后原样返回 echostr | |
| POST /wechat 消息回调(安全模式): 验签 -> AES-256-CBC 解密 -> 解析消息 | |
| -> 组织被动回复 -> 加密返回; 无需回复时返回 success | |
| 运行: | |
| python3 wechat_verify.py # 监听 0.0.0.0:8099 | |
| python3 wechat_verify.py --selftest # 内置 AES 已知答案自检(不出网) | |
| 公众号后台「设置与开发 - 基本配置 - 服务器配置」: | |
| URL(服务器地址) = http://<公网IP或域名>:8099/wechat | |
| Token(令牌) = 下方 WECHAT_TOKEN | |
| 消息加解密方式 = 安全模式 | |
| EncodingAESKey = 下方 WECHAT_ENCODING_AES_KEY | |
| 说明: 只使用 Python 标准库, 服务器上无需 pip install 任何包; | |
| AES-256-CBC 为纯 Python 实现, 仅用于功能验证, 生产请用系统 OpenSSL。 | |
| """ | |
| import base64 | |
| import hashlib | |
| import os | |
| import re | |
| import struct | |
| import sys | |
| import time | |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer | |
| from urllib.parse import urlparse, parse_qs | |
| # ===================== 配置区: 改成你自己的 ===================== | |
| WECHAT_TOKEN = 'your_token_here' # 公众号后台填写的 Token | |
| WECHAT_ENCODING_AES_KEY = 'your_encoding_aes_key_here_43_chars_total' # EncodingAESKey(43位) | |
| PORT = 8099 # 监听端口 | |
| # ============================================================== | |
| # ---------------- 纯 Python AES-256-CBC (无任何第三方依赖) ---------------- | |
| def _gf_mul(a, b): | |
| """GF(2^8) 多项式乘法 (模 0x11B)""" | |
| r = 0 | |
| for _ in range(8): | |
| if b & 1: | |
| r ^= a | |
| hi = a & 0x80 | |
| a = (a << 1) & 0xFF | |
| if hi: | |
| a ^= 0x1B | |
| b >>= 1 | |
| return r | |
| def _gf_inv(x): | |
| """GF(2^8) 乘法逆元: x^254""" | |
| if x == 0: | |
| return 0 | |
| r, base, e = 1, x, 254 | |
| while e: | |
| if e & 1: | |
| r = _gf_mul(r, base) | |
| base = _gf_mul(base, base) | |
| e >>= 1 | |
| return r | |
| def _rotl8(v, n): | |
| return ((v << n) | (v >> (8 - n))) & 0xFF | |
| # S 盒 / 逆 S 盒: 由 GF 逆元 + 仿射变换生成 (FIPS-197) | |
| SBOX = [0] * 256 | |
| INV_SBOX = [0] * 256 | |
| for _x in range(256): | |
| _b = _gf_inv(_x) | |
| _s = _b ^ _rotl8(_b, 1) ^ _rotl8(_b, 2) ^ _rotl8(_b, 3) ^ _rotl8(_b, 4) ^ 0x63 | |
| SBOX[_x] = _s | |
| for _i in range(256): | |
| INV_SBOX[SBOX[_i]] = _i | |
| def _expand_key(key): | |
| """AES-256 密钥扩展: 32 字节 -> 240 字节轮密钥""" | |
| if len(key) != 32: | |
| raise ValueError('AES-256 密钥必须为 32 字节') | |
| rcon = [0, 1, 2, 4, 8, 16, 32, 64, 128, 27, 54] | |
| words = [list(key[i:i + 4]) for i in range(0, 32, 4)] | |
| for i in range(8, 60): | |
| t = list(words[i - 1]) | |
| if i % 8 == 0: | |
| t = t[1:] + t[:1] # RotWord | |
| t = [SBOX[b] for b in t] # SubWord | |
| t[0] ^= rcon[i // 8] | |
| elif i % 8 == 4: | |
| t = [SBOX[b] for b in t] # AES-256 特有的一次 SubWord | |
| words.append([words[i - 8][j] ^ t[j] for j in range(4)]) | |
| return [b for w in words for b in w] | |
| def _add_rk(s, rk): | |
| return [s[i] ^ rk[i] for i in range(16)] | |
| def _shift_rows(s): | |
| for r in range(1, 4): | |
| row = [s[r + 4 * c] for c in range(4)] | |
| row = row[r:] + row[:r] | |
| for c in range(4): | |
| s[r + 4 * c] = row[c] | |
| def _inv_shift_rows(s): | |
| for r in range(1, 4): | |
| row = [s[r + 4 * c] for c in range(4)] | |
| row = row[-r:] + row[:-r] | |
| for c in range(4): | |
| s[r + 4 * c] = row[c] | |
| def _mix_columns(s): | |
| for c in range(4): | |
| a = s[4 * c:4 * c + 4] | |
| s[4 * c] = _gf_mul(a[0], 2) ^ _gf_mul(a[1], 3) ^ a[2] ^ a[3] | |
| s[4 * c + 1] = a[0] ^ _gf_mul(a[1], 2) ^ _gf_mul(a[2], 3) ^ a[3] | |
| s[4 * c + 2] = a[0] ^ a[1] ^ _gf_mul(a[2], 2) ^ _gf_mul(a[3], 3) | |
| s[4 * c + 3] = _gf_mul(a[0], 3) ^ a[1] ^ a[2] ^ _gf_mul(a[3], 2) | |
| def _inv_mix_columns(s): | |
| for c in range(4): | |
| a = s[4 * c:4 * c + 4] | |
| s[4 * c] = _gf_mul(a[0], 14) ^ _gf_mul(a[1], 11) ^ _gf_mul(a[2], 13) ^ _gf_mul(a[3], 9) | |
| s[4 * c + 1] = _gf_mul(a[0], 9) ^ _gf_mul(a[1], 14) ^ _gf_mul(a[2], 11) ^ _gf_mul(a[3], 13) | |
| s[4 * c + 2] = _gf_mul(a[0], 13) ^ _gf_mul(a[1], 9) ^ _gf_mul(a[2], 14) ^ _gf_mul(a[3], 11) | |
| s[4 * c + 3] = _gf_mul(a[0], 11) ^ _gf_mul(a[1], 13) ^ _gf_mul(a[2], 9) ^ _gf_mul(a[3], 14) | |
| def _encrypt_block(block, rk): | |
| s = list(block) | |
| s = _add_rk(s, rk[0]) | |
| for r in range(1, 14): | |
| s = [SBOX[b] for b in s] | |
| _shift_rows(s) | |
| _mix_columns(s) | |
| s = _add_rk(s, rk[r]) | |
| s = [SBOX[b] for b in s] | |
| _shift_rows(s) | |
| return bytes(_add_rk(s, rk[14])) | |
| def _decrypt_block(block, rk): | |
| s = list(block) | |
| s = _add_rk(s, rk[14]) | |
| for r in range(13, 0, -1): | |
| _inv_shift_rows(s) | |
| s = [INV_SBOX[b] for b in s] | |
| s = _add_rk(s, rk[r]) | |
| _inv_mix_columns(s) | |
| _inv_shift_rows(s) | |
| s = [INV_SBOX[b] for b in s] | |
| return bytes(_add_rk(s, rk[0])) | |
| def _xor_bytes(a, b): | |
| return bytes(x ^ y for x, y in zip(a, b)) | |
| def aes_cbc_encrypt(plain, key, iv): | |
| rk = [_expand_key(key)[16 * i:16 * i + 16] for i in range(15)] | |
| prev, out = iv, b'' | |
| for i in range(0, len(plain), 16): | |
| enc = _encrypt_block(_xor_bytes(plain[i:i + 16], prev), rk) | |
| out += enc | |
| prev = enc | |
| return out | |
| def aes_cbc_decrypt(cipher, key, iv): | |
| rk = [_expand_key(key)[16 * i:16 * i + 16] for i in range(15)] | |
| prev, out = iv, b'' | |
| for i in range(0, len(cipher), 16): | |
| block = cipher[i:i + 16] | |
| out += _xor_bytes(_decrypt_block(block, rk), prev) | |
| prev = block | |
| return out | |
| # ---------------- 微信安全模式: 验签 / 加解密 ---------------- | |
| def _key_and_iv(): | |
| key = base64.b64decode(WECHAT_ENCODING_AES_KEY + '=') # 43位 Base64 补 '=' -> 32 字节 | |
| return key, key[:16] | |
| def wx_verify_signature(signature, timestamp, nonce, encrypt=None): | |
| """SHA1(sort(token, timestamp, nonce[, encrypt]))""" | |
| parts = [WECHAT_TOKEN, timestamp, nonce] | |
| if encrypt: | |
| parts.append(encrypt) | |
| parts.sort() | |
| return hashlib.sha1(''.join(parts).encode('utf-8')).hexdigest() == signature | |
| def wx_msg_signature(timestamp, nonce, encrypt): | |
| parts = [WECHAT_TOKEN, timestamp, nonce, encrypt] | |
| parts.sort() | |
| return hashlib.sha1(''.join(parts).encode('utf-8')).hexdigest() | |
| def wx_decrypt(encrypt): | |
| """密文 -> 明文 XML (随机串16 + 长度4 + XML + receiveId)""" | |
| key, iv = _key_and_iv() | |
| plain = aes_cbc_decrypt(base64.b64decode(encrypt), key, iv) | |
| pad = plain[-1] | |
| if not (1 <= pad <= 32): | |
| raise ValueError('PKCS7 填充无效') | |
| plain = plain[:-pad] | |
| if len(plain) < 20: | |
| raise ValueError('明文长度不足') | |
| msg_len = struct.unpack('>I', plain[16:20])[0] | |
| if msg_len > len(plain) - 20: | |
| raise ValueError('消息长度越界') | |
| return plain[20:20 + msg_len].decode('utf-8') | |
| def wx_encrypt(xml_text, receive_id, timestamp, nonce): | |
| """明文 XML -> 安全模式密文, 返回 base64 密文""" | |
| key, iv = _key_and_iv() | |
| msg = xml_text.encode('utf-8') | |
| plain = os.urandom(16) + struct.pack('>I', len(msg)) + msg + receive_id.encode('utf-8') | |
| pad = 32 - len(plain) % 32 | |
| plain += bytes([pad]) * pad | |
| return base64.b64encode(aes_cbc_encrypt(plain, key, iv)).decode('ascii') | |
| # ---------------- XML 解析 / 被动回复构建 ---------------- | |
| def extract_tag(xml, tag): | |
| """提取单个 XML 标签内容 (支持 CDATA)""" | |
| m = re.search(r'<%s>\s*(?:<!\[CDATA\[([\s\S]*?)\]\]>|([\s\S]*?))\s*</%s>' % (tag, tag), xml) | |
| if not m: | |
| return None | |
| return (m.group(1) if m.group(1) is not None else (m.group(2) or '')).strip() | |
| def build_text_reply(to_user, from_user, content): | |
| return ( | |
| '<xml>' | |
| '<ToUserName><![CDATA[%s]]></ToUserName>' | |
| '<FromUserName><![CDATA[%s]]></FromUserName>' | |
| '<CreateTime>%d</CreateTime>' | |
| '<MsgType><![CDATA[text]]></MsgType>' | |
| '<Content><![CDATA[%s]]></Content>' | |
| '</xml>') % (to_user, from_user, int(time.time()), content) | |
| def build_encrypted_reply(encrypt, msg_signature, timestamp, nonce): | |
| return ( | |
| '<xml>' | |
| '<Encrypt><![CDATA[%s]]></Encrypt>' | |
| '<MsgSignature><![CDATA[%s]]></MsgSignature>' | |
| '<TimeStamp>%s</TimeStamp>' | |
| '<Nonce><![CDATA[%s]]></Nonce>' | |
| '</xml>') % (encrypt, msg_signature, timestamp, nonce) | |
| def handle_message(plain_xml): | |
| """解析解密后的明文消息, 返回回复 XML; 无需回复返回 None""" | |
| from_user = extract_tag(plain_xml, 'FromUserName') or '' | |
| to_user = extract_tag(plain_xml, 'ToUserName') or '' | |
| msg_type = extract_tag(plain_xml, 'MsgType') or '' | |
| if msg_type == 'text': | |
| content = extract_tag(plain_xml, 'Content') or '' | |
| print('[recv text] openid=%s content=%s' % (from_user, content)) | |
| return build_text_reply(from_user, to_user, '收到: ' + content) | |
| if msg_type == 'event': | |
| event = extract_tag(plain_xml, 'Event') or '' | |
| print('[recv event] openid=%s event=%s' % (from_user, event)) | |
| if event in ('subscribe', 'SCAN'): | |
| return build_text_reply(from_user, to_user, '欢迎关注! 发送任意文本测试回复。') | |
| return None # 其他消息/事件: 静默返回 success | |
| # ---------------- HTTP 服务 ---------------- | |
| class WeChatHandler(BaseHTTPRequestHandler): | |
| server_version = 'WeChatVerify/1.0' | |
| def _is_wechat_route(self): | |
| return urlparse(self.path).path == '/wechat' | |
| def _is_health_route(self): | |
| return urlparse(self.path).path == '/health' | |
| def _query(self): | |
| return {k: v[0] for k, v in parse_qs(urlparse(self.path).query).items()} | |
| def _send(self, text, code=200, ctype='text/plain'): | |
| data = text.encode('utf-8') | |
| self.send_response(code) | |
| self.send_header('Content-Type', '%s; charset=utf-8' % ctype) | |
| self.send_header('Content-Length', str(len(data))) | |
| self.end_headers() | |
| self.wfile.write(data) | |
| def do_GET(self): | |
| if self._is_health_route(): | |
| return self._send('ok') # 健康检查: 服务已启动 | |
| if not self._is_wechat_route(): | |
| return self._send('not found', 404) | |
| q = self._query() | |
| signature, timestamp = q.get('signature', ''), q.get('timestamp', '') | |
| nonce, echostr = q.get('nonce', ''), q.get('echostr', '') | |
| if not (signature and timestamp and nonce and echostr): | |
| return self._send('invalid params', 400) | |
| if wx_verify_signature(signature, timestamp, nonce): | |
| return self._send(echostr) # 验证通过: 原样返回 echostr | |
| return self._send('invalid signature', 403) | |
| def do_POST(self): | |
| if self._is_health_route(): | |
| return self._send('ok') | |
| if not self._is_wechat_route(): | |
| return self._send('not found', 404) | |
| q = self._query() | |
| signature = q.get('signature') or q.get('msg_signature') or '' | |
| timestamp, nonce = q.get('timestamp', ''), q.get('nonce', '') | |
| body = self.rfile.read(int(self.headers.get('Content-Length') or 0)).decode('utf-8', 'replace') | |
| encrypt = extract_tag(body, 'Encrypt') | |
| if not encrypt: | |
| return self._send('missing encrypt', 400) | |
| if not wx_verify_signature(signature, timestamp, nonce, encrypt): | |
| return self._send('invalid signature', 403) | |
| try: | |
| plain = wx_decrypt(encrypt) | |
| except Exception as e: | |
| print('[error] 解密失败: %s' % e) | |
| return self._send('decrypt failed', 400) | |
| reply_xml = handle_message(plain) | |
| if reply_xml is None: | |
| return self._send('success') # 空回复 | |
| enc = wx_encrypt(reply_xml, extract_tag(plain, 'ToUserName') or '', timestamp, nonce) | |
| return self._send(build_encrypted_reply(enc, wx_msg_signature(timestamp, nonce, enc), | |
| timestamp, nonce), | |
| ctype='application/xml') | |
| def log_message(self, fmt, *args): | |
| print('[%s] %s' % (self.log_date_time_string(), fmt % args)) | |
| # ---------------- 自检 / 启动 ---------------- | |
| def selftest(): | |
| """AES-256 已知答案测试 (FIPS-197 C.3) + 微信加解密往返""" | |
| key = bytes.fromhex('000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f') | |
| pt = bytes.fromhex('00112233445566778899aabbccddeeff') | |
| ct = bytes.fromhex('8ea2b7ca516745bfeafc49904b496089') | |
| rk = [_expand_key(key)[16 * i:16 * i + 16] for i in range(15)] | |
| got = _encrypt_block(pt, rk) | |
| assert got == ct, 'AES-256 加密 KAT 失败: %s' % got.hex() | |
| assert _decrypt_block(ct, rk) == pt, 'AES-256 解密 KAT 失败' | |
| global WECHAT_TOKEN, WECHAT_ENCODING_AES_KEY | |
| WECHAT_TOKEN = 'selftest_token' | |
| WECHAT_ENCODING_AES_KEY = base64.b64encode(os.urandom(32)).decode('ascii')[:-1] | |
| xml = ('<xml><ToUserName><![CDATA[gh_selftest]]></ToUserName>' | |
| '<FromUserName><![CDATA[o_selftest_user]]></FromUserName>' | |
| '<CreateTime>1234567890</CreateTime><MsgType><![CDATA[text]]></MsgType>' | |
| '<Content><![CDATA[你好 TrackStack]]></Content><MsgId>1001</MsgId></xml>') | |
| enc = wx_encrypt(xml, 'gh_selftest', '1700000000', 'nonce-1') | |
| assert wx_decrypt(enc) == xml, '微信 Encrypt/Decrypt 往返失败' | |
| print('[ok] AES-256 KAT 通过; 微信 Encrypt/Decrypt 往返通过') | |
| def main(): | |
| if len(sys.argv) > 1 and sys.argv[1] == '--selftest': | |
| selftest() | |
| return | |
| if 'your_' in WECHAT_TOKEN: | |
| print('[warn] 请先修改文件顶部 WECHAT_TOKEN') | |
| if len(WECHAT_ENCODING_AES_KEY) != 43: | |
| print('[warn] EncodingAESKey 长度应为 43, 当前 %d; POST 加解密暂不可用' % len(WECHAT_ENCODING_AES_KEY)) | |
| print('微信验证服务启动: 0.0.0.0:%d/wechat' % PORT) | |
| print('健康检查: curl http://<IP>:%d/health -> ok' % PORT) | |
| print('公众号后台 URL 填写: http://<公网IP或域名>:%d/wechat' % PORT) | |
| ThreadingHTTPServer(('0.0.0.0', PORT), WeChatHandler).serve_forever() | |
| if __name__ == '__main__': | |
| main() |
Author
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
微信公众号 API 最小验证脚本
单文件、零第三方依赖、开箱即跑的微信公众平台服务器验证脚本（纯 Python 标准库实现）。
功能
/healthok，确认服务是否启动成功/wechatsignature后原样返回echostr/wechatsuccess使用
公众号后台配置
设置与开发 → 基本配置 → 服务器配置：
http://<公网IP或域名>:8099/wechatWECHAT_TOKEN一致WECHAT_ENCODING_AES_KEY一致说明
pip install任何包，直接拷过去即可运行crypto等官方实现验证消息处理
脚本内置基础消息处理，方便逐项验证：
收到: <内容>success