From d695f87e33bf73f2219419909cf52fa949376b44 Mon Sep 17 00:00:00 2001 From: eric Date: Tue, 31 Mar 2026 10:45:13 -0500 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=85=B3=E9=94=AE=E8=AF=8D=E5=9B=9E?= =?UTF-8?q?=E5=A4=8D=E4=BC=98=E5=85=88=E5=B9=B6=E5=90=8E=E5=8F=B0=E8=A7=A6?= =?UTF-8?q?=E5=8F=91B=E5=90=8C=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将A回调链路调整为先返回关键词回复，再以后台任务触发B接口同步，避免B接口耗时或失败影响用户主回复；同时修正微信消息体加解密格式并固定B接口凭据来源。 Made-with: Cursor --- config.py | 5 +---- crypto_utils.py | 25 +++++++++++++++++++++---- main.py | 49 +++++++++++++++++++++---------------------------- 3 files changed, 43 insertions(+), 36 deletions(-) diff --git a/config.py b/config.py index 7138423..23f72ca 100644 --- a/config.py +++ b/config.py @@ -10,7 +10,4 @@ PB_COLLECTION = "wechat_private_users" # A 回调触发 B 接口同步的默认配置 PM_SYNC_AUTHTOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJyaWQiOjE5MzE1MjcsIm9wZW5pZCI6Im85VS04NW1DSXZ0eDZ0a0pteTQyMkkzcUNHVWciLCJzaWduZXRpbWUiOjE3NzQ5MzM0NDQ4MTgsImlhdCI6MTc3NDkzMzQ0NCwiZXhwIjoxNzc3NTI1NDQ0fQ.iC9vdjBqa2LGlqA6F-Dg4GcfDyv3Rp8JwpBaO3j_eE4" -PM_SYNC_DEFAULT_WXBOT_BID = "4888c6ab5cb70dcc3adea1e2d2ff0201" -PM_SYNC_BID_BY_CHANNEL = { - "wx5cee92c2c93fb480": "4888c6ab5cb70dcc3adea1e2d2ff0201", -} +PM_SYNC_DEFAULT_WXBOT_BID = "75021053883a9090c67d5becde7f4b0a" diff --git a/crypto_utils.py b/crypto_utils.py index 33e3eb5..096b69c 100644 --- a/crypto_utils.py +++ b/crypto_utils.py @@ -1,5 +1,7 @@ import base64 import hashlib +import os +import struct from typing import Tuple from Crypto.Cipher import AES @@ -34,16 +36,31 @@ def get_aes_key_and_iv(encoding_aes_key: str) -> Tuple[bytes, bytes]: def aes_decrypt_base64(cipher_b64: str, encoding_aes_key: str) -> str: aes_key, iv = get_aes_key_and_iv(encoding_aes_key) - cipher_bytes = base64.b64decode(cipher_b64) + # 回调密文偶尔会带换行/空白，先规整再补齐 base64 padding。 + compact = "".join(str(cipher_b64 or "").split()) + compact += "=" * (-len(compact) % 4) + cipher_bytes = base64.b64decode(compact) cipher = AES.new(aes_key, AES.MODE_CBC, iv) decrypted = cipher.decrypt(cipher_bytes) decrypted = pkcs5_unpadding(decrypted) - return decrypted.decode("utf-8") + # 微信消息体格式：16B随机串 + 4B网络序长度 + 明文 + appid + if len(decrypted) < 20: + raise ValueError("解密后消息长度非法") + content = decrypted[16:] + msg_len = struct.unpack("!I", content[:4])[0] + msg_start = 4 + msg_end = msg_start + msg_len + if msg_end > len(content): + raise ValueError("解密后消息长度字段越界") + msg = content[msg_start:msg_end] + return msg.decode("utf-8") -def aes_encrypt_base64(plaintext: str, encoding_aes_key: str) -> str: +def aes_encrypt_base64(plaintext: str, encoding_aes_key: str, app_id: str) -> str: aes_key, iv = get_aes_key_and_iv(encoding_aes_key) - data = plaintext.encode("utf-8") + raw = plaintext.encode("utf-8") + # 微信消息体格式：16B随机串 + 4B网络序长度 + 明文 + appid + data = os.urandom(16) + struct.pack("!I", len(raw)) + raw + app_id.encode("utf-8") padded = pkcs5_padding(data) cipher = AES.new(aes_key, AES.MODE_CBC, iv) encrypted = cipher.encrypt(padded) diff --git a/main.py b/main.py index f070e8b..0ea0396 100644 --- a/main.py +++ b/main.py @@ -6,7 +6,7 @@ from datetime import datetime, timezone from typing import Optional import httpx -from fastapi import FastAPI, Request, HTTPException +from fastapi import FastAPI, Request, HTTPException, BackgroundTasks from fastapi.responses import PlainTextResponse, HTMLResponse from config import ( @@ -19,7 +19,6 @@ from config import ( PB_COLLECTION, PM_SYNC_AUTHTOKEN, PM_SYNC_DEFAULT_WXBOT_BID, - PM_SYNC_BID_BY_CHANNEL, ) from crypto_utils import aes_decrypt_base64, aes_encrypt_base64, calc_signature @@ -240,10 +239,8 @@ async def _sync_user_profile_from_b(data: dict) -> None: return channel_id = str(data.get("ChannelId", "") or data.get("channelId", "")).strip() - bid_map = PM_SYNC_BID_BY_CHANNEL if isinstance(PM_SYNC_BID_BY_CHANNEL, dict) else {} - bid_map = {str(k): str(v) for k, v in bid_map.items()} - - wxbot_bid = bid_map.get(channel_id) or (PM_SYNC_DEFAULT_WXBOT_BID or "").strip() + # B 接口配置与 A 回调路由解耦：固定使用 B 的默认凭据。 + wxbot_bid = (PM_SYNC_DEFAULT_WXBOT_BID or "").strip() if not wxbot_bid: logger.warning("未配置可用 wxbot_bid，跳过 B 接口同步") return @@ -394,7 +391,7 @@ async def query_private_messages( return resp.json() -async def _handle_wechat_request(request: Request, app_id: Optional[str]) -> str: +async def _handle_wechat_request(request: Request, app_id: Optional[str], background_tasks: BackgroundTasks) -> str: """ 公共处理逻辑：便于同时支持 "/" 和 "/wechat/thirdapi" 两种回调路径。 """ @@ -415,11 +412,11 @@ async def _handle_wechat_request(request: Request, app_id: Optional[str]) -> str logger.info(f"收到原始 body（未解密，前 500 字符）：{cipher_b64[:500]}") # 平台当前实际发送的是 {"encrypted":"..."} 这一层 JSON，需要先解析再取字段 - if cipher_b64.startswith("{"): + if "encrypted" in cipher_b64: try: outer = json.loads(cipher_b64) if isinstance(outer, dict) and "encrypted" in outer: - cipher_b64 = str(outer["encrypted"]).replace("\n", "").strip() + cipher_b64 = "".join(str(outer["encrypted"]).split()).strip() except Exception: # 如果解析失败，就按原始字符串继续处理（保持兼容） logger.warning("外层 JSON 解析失败，按原始 body 作为密文处理") @@ -474,20 +471,16 @@ async def _handle_wechat_request(request: Request, app_id: Optional[str]) -> str except Exception: logger.exception("签名校验异常（忽略或按需处理）") - # 7. 根据业务逻辑构造应答 - # 7.1 先做用户资料同步（异常不影响主回复） - try: - await _sync_user_profile_from_b(data) - except Exception: - logger.exception("触发 B 接口同步到 PocketBase 失败（不影响回复）") - - # 7.2 再生成业务回复 + # 7. 根据业务逻辑构造应答（关键词回复优先） try: answer_text = await handle_business(data) except Exception: logger.exception("业务处理异常，将返回兜底文案") answer_text = "系统繁忙，请稍后再试。" + # 7.1 主回复确定后再后台触发 B 同步，避免影响回复链路耗时 + background_tasks.add_task(_sync_user_profile_from_b, data) + resp_obj = { # 文本类型，内容为可被 JSON.parse 的 H5 结构字符串 "answer_type": "text", @@ -500,7 +493,7 @@ async def _handle_wechat_request(request: Request, app_id: Optional[str]) -> str # 8. 加密响应 JSON try: - encrypted_resp = aes_encrypt_base64(resp_json_str, ENCODING_AES_KEY) + encrypted_resp = aes_encrypt_base64(resp_json_str, ENCODING_AES_KEY, APP_ID) except Exception as e: logger.exception("响应 AES 加密失败") raise HTTPException(status_code=500, detail="encrypt error") from e @@ -510,21 +503,21 @@ async def _handle_wechat_request(request: Request, app_id: Optional[str]) -> str @app.post("/", response_class=PlainTextResponse) -async def wechat_root(request: Request, app_id: Optional[str] = None): +async def wechat_root(request: Request, background_tasks: BackgroundTasks, app_id: Optional[str] = None): """ 有些配置示例直接写根路径，例如： url: http://example.com/ 这种情况下，微信会 POST 到 "/"，所以这里也做同样处理。 """ - return await _handle_wechat_request(request, app_id) + return await _handle_wechat_request(request, app_id, background_tasks) @app.post("/wechat/thirdapi", response_class=PlainTextResponse) -async def wechat_thirdapi(request: Request, app_id: Optional[str] = None): +async def wechat_thirdapi(request: Request, background_tasks: BackgroundTasks, app_id: Optional[str] = None): """ 显式的 /wechat/thirdapi 路径，同样处理。 """ - return await _handle_wechat_request(request, app_id) + return await _handle_wechat_request(request, app_id, background_tasks) @app.get("/health") @@ -807,18 +800,18 @@ def private_message_page():