diff --git a/__pycache__/config.cpython-312.pyc b/__pycache__/config.cpython-312.pyc new file mode 100644 index 0000000..a9ec32d Binary files /dev/null and b/__pycache__/config.cpython-312.pyc differ diff --git a/__pycache__/crypto_utils.cpython-312.pyc b/__pycache__/crypto_utils.cpython-312.pyc new file mode 100644 index 0000000..cc4d9d8 Binary files /dev/null and b/__pycache__/crypto_utils.cpython-312.pyc differ diff --git a/__pycache__/main.cpython-312.pyc b/__pycache__/main.cpython-312.pyc new file mode 100644 index 0000000..55bbde4 Binary files /dev/null and b/__pycache__/main.cpython-312.pyc differ diff --git a/config.py b/config.py new file mode 100644 index 0000000..a0780ab --- /dev/null +++ b/config.py @@ -0,0 +1,3 @@ +APP_ID = "fXYWvGfq04uMScP" +TOKEN = "EWRotyMHfTLAlYmCHpdH8AZcuZPBHn" +ENCODING_AES_KEY = "1wJZiJm8URfRjkHu1QfIVDeNFc2jbzzeCZqvhQecsCx" diff --git a/crypto_utils.py b/crypto_utils.py new file mode 100644 index 0000000..33e3eb5 --- /dev/null +++ b/crypto_utils.py @@ -0,0 +1,62 @@ +import base64 +import hashlib +from typing import Tuple + +from Crypto.Cipher import AES + + +BLOCK_SIZE = 32 # PKCS5/PKCS7 padding 使用 32 字节块大小 + + +def pkcs5_unpadding(data: bytes) -> bytes: + pad = data[-1] + if pad < 1 or pad > 32: + return data + return data[:-pad] + + +def pkcs5_padding(data: bytes) -> bytes: + amount_to_pad = BLOCK_SIZE - (len(data) % BLOCK_SIZE) + return data + bytes([amount_to_pad]) * amount_to_pad + + +def get_aes_key_and_iv(encoding_aes_key: str) -> Tuple[bytes, bytes]: + """ + encoding_aes_key 是平台给的 43 位字符串，需要先补一个 '=' 再 base64 解码。 + 得到 32 字节 key，前 16 字节为 iv。 + """ + aes_key = base64.b64decode(encoding_aes_key + "=") + if len(aes_key) != 32: + raise ValueError("encodingAESKey 无效，解码后长度不是 32 字节") + iv = aes_key[:16] + return aes_key, iv + + +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) + cipher = AES.new(aes_key, AES.MODE_CBC, iv) + decrypted = cipher.decrypt(cipher_bytes) + decrypted = pkcs5_unpadding(decrypted) + return decrypted.decode("utf-8") + + +def aes_encrypt_base64(plaintext: str, encoding_aes_key: str) -> str: + aes_key, iv = get_aes_key_and_iv(encoding_aes_key) + data = plaintext.encode("utf-8") + padded = pkcs5_padding(data) + cipher = AES.new(aes_key, AES.MODE_CBC, iv) + encrypted = cipher.encrypt(padded) + return base64.b64encode(encrypted).decode("utf-8") + + +def calc_signature(token: str, timestamp: int, skill_name: str, intent_name: str, query: str) -> str: + """ + 文档签名算法： + str = token + time.Now.Unix + skill_name + intent_name + query + signature = MD5(str) 小写 hex + 参考：https://developers.weixin.qq.com/doc/aispeech/confapi/thirdapi/thirdapi.html + """ + s = f"{token}{timestamp}{skill_name}{intent_name}{query}" + return hashlib.md5(s.encode("utf-8")).hexdigest() + diff --git a/main.py b/main.py new file mode 100644 index 0000000..597f8c4 --- /dev/null +++ b/main.py @@ -0,0 +1,577 @@ +import json +import logging +import base64 +from datetime import datetime, timezone +from typing import Optional + +import httpx +from fastapi import FastAPI, Request, HTTPException +from fastapi.responses import PlainTextResponse, HTMLResponse + +from config import APP_ID, TOKEN, ENCODING_AES_KEY +from crypto_utils import aes_decrypt_base64, aes_encrypt_base64, calc_signature + + +logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") +logger = logging.getLogger(__name__) + +app = FastAPI(title="WeChat Dialog ThirdAPI Demo") + + +async def handle_business(data: dict) -> str: + """ + 业务处理钩子： + - data 为解密后的请求 JSON（XwBrainThirdApiReqInfo 扩展结构） + - 返回值为要展示给用户的短文本答案 + 后续如果要接数据库、HTTP 接口，只需要改这里的逻辑即可。 + """ + query = data.get("Query", "") + skill_name = data.get("SkillName", "") + intent_name = data.get("IntentName", "") + user_id = data.get("UserId", "") + + # 如果想返回普通文本，可以使用下面的示例文案： + # parts = [ + # f"技能：{skill_name or '未提供'}", + # f"意图：{intent_name or '未提供'}", + # f"用户ID：{user_id or '未提供'}", + # f"用户问题：{query or '空'}", + # ] + # return "；".join(parts) + + # 当前示例：返回一个 H5 富文本结构（平台会对 short_answer 做 JSON.parse） + h5_msg = { + "news": { + "articles": [ + { + "title": "实时更新：新型肺炎疫情最新动态", + "description": "腾讯新闻第一时间同步全国新型肺炎疫情动态，欢迎关注、转发", + "url": "https://news.qq.com/zt2020/page/feiyan.htm", + "picurl": "http://mmbiz.qpic.cn/mmbiz_jpg/W3gQtpV3j8D8kZRqfpTJlfVqubwgFQf47H0GWlGV6leaDF80ZpdtuFhQVsCsM3YKmwkujXzdjR2k6aWfA41ic7Q/0?wx_fmt=jpeg", + "type": "h5", + } + ] + } + } + # 按文档要求，answer/short_answer 是字符串，内部再 JSON.parse 才是对象 + return json.dumps(h5_msg, ensure_ascii=False) + + +def _decode_jwt_payload_without_verify(token: str) -> dict: + """ + 仅用于本地调试分析 JWT 的 payload（不做签名校验）。 + """ + try: + parts = token.split(".") + if len(parts) != 3: + return {} + payload = parts[1] + payload += "=" * (-len(payload) % 4) # base64url 补齐 + data = base64.urlsafe_b64decode(payload.encode("utf-8")).decode("utf-8") + obj = json.loads(data) + return obj if isinstance(obj, dict) else {} + except Exception: + return {} + + +async def query_private_messages( + authtoken: str, + wxbot_bid: str, + page: int = 0, + size: int = 30, + filter_value: int = 0, + request_id: str = "local-debug", +) -> dict: + """ + 转发请求到微信对话开放平台的私信列表接口（你在 Charles 抓包的接口）。 + """ + url = "https://chatbot.weixin.qq.com/miniopenai/manualservice/getaccessstautuslist" + body = { + "bid": wxbot_bid, + "filter": filter_value, + "order": {"page": page, "size": size}, + "base": {"requestid": request_id}, + } + headers = { + "authtoken": authtoken, + "wxbot_bid": wxbot_bid, + "content-type": "application/json", + "charset": "utf-8", + } + # 调试场景下常见抓包证书/公司代理证书导致校验失败，这里关闭 verify 以保证接口可用。 + # 生产环境建议改为 verify=True 并配置受信任证书链。 + async with httpx.AsyncClient(timeout=10, verify=False) as client: + resp = await client.post(url, json=body, headers=headers) + resp.raise_for_status() + return resp.json() + + +async def _handle_wechat_request(request: Request, app_id: Optional[str]) -> str: + """ + 公共处理逻辑：便于同时支持 "/" 和 "/wechat/thirdapi" 两种回调路径。 + """ + # 1. app_id 校验 + # 说明：有些平台配置下不会显式在 URL 带 app_id，这种情况下我们只做日志，不强制拦截，避免 400。 + if app_id is None: + logger.warning("请求中未携带 app_id（将继续处理，仅做告警）") + elif app_id != APP_ID: + logger.warning(f"app_id 不匹配，期望={APP_ID}, 实际={app_id}（将继续处理，仅做告警）") + + # 2. 获取原始 body（加密后的 base64 字符串） + raw_body_bytes = await request.body() + cipher_b64 = raw_body_bytes.decode("utf-8").strip() + if not cipher_b64: + raise HTTPException(status_code=400, detail="empty body") + + # 打印原始 body，便于对照平台文档排查（是否真的是 base64 + AES 加密数据） + logger.info(f"收到原始 body（未解密，前 500 字符）：{cipher_b64[:500]}") + + # 平台当前实际发送的是 {"encrypted":"..."} 这一层 JSON，需要先解析再取字段 + if cipher_b64.startswith("{"): + try: + outer = json.loads(cipher_b64) + if isinstance(outer, dict) and "encrypted" in outer: + cipher_b64 = str(outer["encrypted"]).replace("\n", "").strip() + except Exception: + # 如果解析失败，就按原始字符串继续处理（保持兼容） + logger.warning("外层 JSON 解析失败，按原始 body 作为密文处理") + + # 3. 解密 + try: + decrypted_json_str = aes_decrypt_base64(cipher_b64, ENCODING_AES_KEY) + except Exception: + # 常见情况：非第三方服务接口的探测/噪音流量，或使用了不同的 AESKey。 + logger.error( + "AES 解密失败，可能为非本机器人第三方服务请求或密钥不匹配（忽略该请求）。" + ) + raise HTTPException(status_code=400, detail="decrypt error") + + # 4. 解析 JSON + try: + data = json.loads(decrypted_json_str) + except json.JSONDecodeError: + logger.error(f"解密结果不是合法 JSON: {decrypted_json_str}") + raise HTTPException(status_code=400, detail="invalid json in decrypted body") + + # 5. 打印请求内容到终端日志（分析、显示，带中文注释） + logger.info("收到微信第三方服务请求(已解密)：") + logger.info( + "请求概览：RequestId=%s，用户ID=%s，问题=%s，技能=%s，意图=%s", + data.get("RequestId", ""), + data.get("UserId", ""), + data.get("Query", ""), + data.get("SkillName", ""), + data.get("IntentName", ""), + ) + logger.info("完整请求JSON：%s", json.dumps(data, ensure_ascii=False, indent=2)) + + # 6. （可选）签名校验 + try: + timestamp = data.get("Timestamp") + skill_name = data.get("SkillName", "") + intent_name = data.get("IntentName", "") + query = data.get("Query", "") + provided_sig = data.get("Signature", "") + + if timestamp is not None and provided_sig: + expected_sig = calc_signature(TOKEN, timestamp, skill_name, intent_name, query) + if expected_sig != provided_sig: + logger.warning( + f"签名校验失败，expected={expected_sig}, provided={provided_sig}" + ) + # 这里直接拒绝请求，你也可以根据需要改成仅告警 + raise HTTPException(status_code=400, detail="invalid signature") + except HTTPException: + raise + except Exception: + logger.exception("签名校验异常（忽略或按需处理）") + + # 7. 根据业务逻辑构造应答 + try: + answer_text = await handle_business(data) + except Exception: + logger.exception("业务处理异常，将返回兜底文案") + answer_text = "系统繁忙，请稍后再试。" + + resp_obj = { + # 文本类型，内容为可被 JSON.parse 的 H5 结构字符串 + "answer_type": "text", + "text_info": { + "short_answer": answer_text + } + } + + resp_json_str = json.dumps(resp_obj, ensure_ascii=False) + + # 8. 加密响应 JSON + try: + encrypted_resp = aes_encrypt_base64(resp_json_str, ENCODING_AES_KEY) + except Exception as e: + logger.exception("响应 AES 加密失败") + raise HTTPException(status_code=500, detail="encrypt error") from e + + # 按文档要求：如果配置了加密，这里返回的是加密后的 base64 字符串 + return encrypted_resp + + +@app.post("/", response_class=PlainTextResponse) +async def wechat_root(request: Request, app_id: Optional[str] = None): + """ + 有些配置示例直接写根路径，例如： + url: http://example.com/ + 这种情况下，微信会 POST 到 "/"，所以这里也做同样处理。 + """ + return await _handle_wechat_request(request, app_id) + + +@app.post("/wechat/thirdapi", response_class=PlainTextResponse) +async def wechat_thirdapi(request: Request, app_id: Optional[str] = None): + """ + 显式的 /wechat/thirdapi 路径，同样处理。 + """ + return await _handle_wechat_request(request, app_id) + + +@app.get("/health") +def health_check(): + return {"status": "ok"} + + +@app.get("/debug/echo") +async def debug_echo(request: Request, app_id: Optional[str] = None): + """ + 调试用接口：回显请求头和 app_id，方便排查签名和路由问题。 + """ + return { + "app_id": app_id, + "headers": dict(request.headers), + } + + +@app.get("/private-message", response_class=HTMLResponse) +def private_message_page(): + """ + 私信查询页面（本地调试）。 + """ + return """ + + + + + + 私信查询 + + + +
+
+

私信查询调试台

+

用于调试微信对话开放平台私信列表接口，支持一键查询、Token 过期分析与结果可视化。

+ + + + +
+
+ + +
+
+ + +
+
+
+ + +
+
提示：查询失败时，先检查 authtoken 是否过期、wxbot_bid 是否与当前机器人一致。
+
+ +
+

等待操作...

+ 
{
+  "message": "点击上方按钮开始查询"
+}
+
+
+ + + + +""" + + +@app.post("/api/private-message/query") +async def api_private_message_query(payload: dict): + """ + 查询私信列表接口（调试版）。 + """ + authtoken = str(payload.get("authtoken", "")).strip() + wxbot_bid = str(payload.get("wxbot_bid", "")).strip() + page = int(payload.get("page", 0)) + size = int(payload.get("size", 30)) + filter_value = int(payload.get("filter", 0)) + request_id = str(payload.get("requestid", "local-debug")).strip() or "local-debug" + + if not authtoken or not wxbot_bid: + raise HTTPException(status_code=400, detail="authtoken 和 wxbot_bid 不能为空") + + try: + result = await query_private_messages( + authtoken=authtoken, + wxbot_bid=wxbot_bid, + page=page, + size=size, + filter_value=filter_value, + request_id=request_id, + ) + except Exception as e: + logger.error("私信查询失败：%s", str(e)) + raise HTTPException(status_code=502, detail=f"private-message query failed: {e}") + + contacts = result.get("contacts") or [] + return { + "ok": True, + "count": result.get("count"), + "contacts_preview": [ + { + "nick": c.get("nick"), + "userId": c.get("userId"), + "lastMsg": (c.get("lastMsg") or {}).get("content"), + "lastActiveTime": c.get("lastActiveTime"), + } + for c in contacts + ], + "raw": result, + } + + +@app.post("/api/private-message/analyze") +async def api_private_message_analyze(payload: dict): + """ + 分析 authtoken 的过期时间和基本信息（基于 JWT payload，不校验签名）。 + """ + authtoken = str(payload.get("authtoken", "")).strip() + if not authtoken: + raise HTTPException(status_code=400, detail="authtoken 不能为空") + + claims = _decode_jwt_payload_without_verify(authtoken) + now_ts = int(datetime.now(timezone.utc).timestamp()) + exp = int(claims.get("exp", 0)) if str(claims.get("exp", "")).isdigit() else 0 + iat = int(claims.get("iat", 0)) if str(claims.get("iat", "")).isdigit() else 0 + sign_time_ms = claims.get("signetime") + + expires_in_seconds = exp - now_ts if exp else None + if expires_in_seconds is None: + status = "无法解析过期时间" + elif expires_in_seconds <= 0: + status = "已过期" + elif expires_in_seconds < 3600: + status = "即将过期（1小时内）" + else: + status = "有效" + + report = { + "status": status, + "now_utc": datetime.now(timezone.utc).isoformat(), + "claims": claims, + "analysis": { + "iat_utc": datetime.fromtimestamp(iat, tz=timezone.utc).isoformat() if iat else None, + "exp_utc": datetime.fromtimestamp(exp, tz=timezone.utc).isoformat() if exp else None, + "expires_in_seconds": expires_in_seconds, + "expires_in_hours": round(expires_in_seconds / 3600, 2) if isinstance(expires_in_seconds, int) else None, + "signetime_ms": sign_time_ms, + "summary": "该分析基于 JWT payload 解码，不包含签名有效性校验。", + }, + } + return report + diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..659d2f1 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +fastapi +uvicorn +pycryptodome +httpx diff --git a/run.py b/run.py new file mode 100644 index 0000000..529ed34 --- /dev/null +++ b/run.py @@ -0,0 +1,6 @@ +import uvicorn + + +if __name__ == "__main__": + # 如果 8000 被其他程序占用，可以改成 8001 + uvicorn.run("main:app", host="0.0.0.0", port=8001, reload=False) \ No newline at end of file diff --git a/token_client.py b/token_client.py new file mode 100644 index 0000000..f6661e6 --- /dev/null +++ b/token_client.py @@ -0,0 +1,53 @@ +import hashlib +import json +import time +import uuid +from typing import Optional + +import httpx + +from config import APP_ID, TOKEN + + +TOKEN_URL = "https://openaiapi.weixin.qq.com/v2/token" + + +def _md5_hex(data: str) -> str: + return hashlib.md5(data.encode("utf-8")).hexdigest() + + +async def fetch_access_token(account: Optional[str] = None) -> str: + """ + 调用微信智能对话 /v2/token 接口，换取 AccessToken。 + 文档参考：https://developers.weixin.qq.com/doc/aispeech/confapi/dialog/token.html + """ + timestamp = int(time.time()) + nonce = uuid.uuid4().hex[:16] + + body_obj = {} + if account: + body_obj["account"] = account + body_str = json.dumps(body_obj, separators=(",", ":"), ensure_ascii=False) + + # sign = md5(Token + str(unix_timestamp) + nonce + md5(body)) + body_md5 = _md5_hex(body_str) if body_str else _md5_hex("") + sign_src = f"{TOKEN}{timestamp}{nonce}{body_md5}" + sign = _md5_hex(sign_src) + + headers = { + "X-APPID": APP_ID, + "request_id": str(uuid.uuid4()), + "timestamp": str(timestamp), + "nonce": nonce, + "sign": sign, + "content-type": "application/json", + } + + async with httpx.AsyncClient(timeout=5) as client: + resp = await client.post(TOKEN_URL, headers=headers, content=body_str.encode("utf-8")) + resp.raise_for_status() + data = resp.json() + if data.get("code") != 0: + raise RuntimeError(f"fetch_access_token error: {data}") + return data["data"]["access_token"] +