350 lines
17 KiB
Python
350 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""只追加：扫描 SDK 服务器中能控制手机端微信的真实接口。
|
||
|
||
输出：
|
||
- 开发文档/6、测试/110项功能逐项确认/2026-05-18/微信SDK服务器控制接口扫描_追加版_*.md
|
||
- 开发文档/6、测试/110项功能逐项确认/2026-05-18/微信SDK服务器控制接口矩阵_追加版_*.csv
|
||
|
||
判定原则：
|
||
1. 微信业务接口：接口语义指向好友、消息、群、朋友圈、搜索、资料、二维码、小程序、支付/钱包等微信功能。
|
||
2. 微信支撑接口：亮屏、解锁、打开微信、截图、OCR、前台恢复、Frida/ADB 连接等服务于微信控制链路的接口。
|
||
3. 非核心接口：泛设备、非微信、项目管理、注册中心、经验库等不直接控制手机微信端的接口。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import ast
|
||
import csv
|
||
import json
|
||
import os
|
||
import re
|
||
from dataclasses import dataclass, asdict
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Iterable
|
||
|
||
ROOT = Path.cwd()
|
||
OUT_DIR = ROOT / "开发文档" / "6、测试" / "110项功能逐项确认" / "2026-05-18"
|
||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
TS = datetime.now().strftime("%H%M%S")
|
||
|
||
PY_SOURCES = [
|
||
ROOT / "sdk" / "app" / "routers",
|
||
ROOT / "sdk" / "app" / "services",
|
||
ROOT / "sdk" / "agent" / "hook",
|
||
ROOT / "sdk" / "tests",
|
||
ROOT / "tools",
|
||
ROOT / "工作手机SDK补全包_微信Frida",
|
||
ROOT / "补全包" / "微信Frida_SDK_20260518",
|
||
]
|
||
JS_SOURCES = [
|
||
ROOT / "sdk" / "agent" / "hook",
|
||
ROOT / "工作手机SDK补全包_微信Frida",
|
||
ROOT / "补全包" / "微信Frida_SDK_20260518",
|
||
]
|
||
|
||
BUSINESS_KEYWORDS = {
|
||
"好友与通讯录": ["friend", "contact", "contacts", "address", "通讯录", "好友", "添加好友", "验证", "label", "tag", "标签"],
|
||
"消息与会话": ["message", "msg", "chat", "conversation", "session", "send_text", "sendtext", "send_message", "voice_message", "file", "emoji", "消息", "会话", "聊天", "群发", "撤回"],
|
||
"群与社群": ["group", "room", "chatroom", "member", "群", "社群", "群聊", "群成员", "invite", "kick"],
|
||
"朋友圈": ["moment", "moments", "timeline", "sns", "朋友圈", "动态", "like", "comment", "cover", "privacy"],
|
||
"搜索与发现": ["search", "discover", "scan", "qr", "qrcode", "mini", "program", "搜", "发现", "扫一扫", "二维码", "小程序", "公众号"],
|
||
"资料与账号": ["profile", "account", "me", "self", "user_info", "userinfo", "version", "login", "资料", "账号", "名片", "二维码名片", "微信版本"],
|
||
"钱包与交易安全闸": ["wallet", "payment", "pay", "red_packet", "redpacket", "transfer", "transaction", "收款", "支付", "钱包", "红包", "转账", "账单"],
|
||
}
|
||
SUPPORT_KEYWORDS = [
|
||
"frida", "adb", "device", "screen", "screenshot", "capture", "ocr", "unlock", "wake", "foreground", "launch", "start", "connect", "wireless", "uiautomator", "dump", "window", "前台", "截图", "识别", "亮屏", "解锁", "启动", "无线", "连接", "黑屏"
|
||
]
|
||
NON_CORE_KEYWORDS = [
|
||
"project", "registry", "experience", "gateway", "discovery", "server", "health", "metrics", "storage", "network", "cpu", "memory", "browser", "doubao", "soul", "项目", "注册中心", "经验库", "网关", "服务发现", "存储", "网络"
|
||
]
|
||
RISK_KEYWORDS = ["send", "post", "publish", "delete", "remove", "kick", "transfer", "pay", "red", "privacy", "cover", "voice_call", "video_call", "发送", "发布", "删除", "踢", "转账", "支付", "红包", "群发", "修改", "评论", "点赞"]
|
||
WECHAT_HINTS = ["wechat", "微信", "com.tencent.mm", "weixin", "frida"]
|
||
|
||
@dataclass
|
||
class InterfaceItem:
|
||
source_type: str
|
||
file: str
|
||
line: int
|
||
method: str
|
||
route_or_symbol: str
|
||
function: str
|
||
tag: str
|
||
scene: str
|
||
core_level: str
|
||
risk_level: str
|
||
reason: str
|
||
evidence_required: str
|
||
full_key: str
|
||
|
||
|
||
def rel(path: Path) -> str:
|
||
try:
|
||
return str(path.relative_to(ROOT))
|
||
except Exception:
|
||
return str(path)
|
||
|
||
|
||
def contains_any(text: str, words: Iterable[str]) -> bool:
|
||
t = text.lower()
|
||
return any(w.lower() in t for w in words)
|
||
|
||
|
||
def classify(text: str, file: str) -> tuple[str, str, str, str]:
|
||
combined = f"{text} {file}"
|
||
scene = "未分类"
|
||
for group, words in BUSINESS_KEYWORDS.items():
|
||
if contains_any(combined, words):
|
||
scene = group
|
||
break
|
||
has_wechat = contains_any(combined, WECHAT_HINTS)
|
||
has_business = scene != "未分类"
|
||
has_support = contains_any(combined, SUPPORT_KEYWORDS)
|
||
has_non_core = contains_any(combined, NON_CORE_KEYWORDS)
|
||
|
||
if has_business and (has_wechat or "unified" in file.lower() or "wechat_full" in file.lower()):
|
||
core_level = "核心微信业务接口"
|
||
reason = "接口语义指向微信业务控制，需验证服务器能控制手机端微信完成对应页面或数据动作。"
|
||
elif has_business:
|
||
core_level = "待人工归类的微信疑似接口"
|
||
reason = "接口语义像微信业务，但文件或上下文未明确绑定微信，需要人工确认是否落到手机端微信。"
|
||
elif has_support and (has_wechat or "adb" in file.lower() or "frida" in file.lower() or "capture" in file.lower() or "device" in file.lower()):
|
||
scene = "设备支撑"
|
||
core_level = "微信控制支撑接口"
|
||
reason = "接口用于连接、亮屏、解锁、截图、识别或恢复微信前台，作为微信控制链路支撑能力。"
|
||
elif has_non_core:
|
||
scene = "非核心"
|
||
core_level = "非核心接口"
|
||
reason = "接口偏项目、注册、网关、泛设备或非微信能力，不计入核心微信控制完成率。"
|
||
else:
|
||
core_level = "待确认非核心或遗漏接口"
|
||
reason = "未命中微信业务或支撑关键词，需要结合代码调用链复核。"
|
||
risk = "高风险需安全闸" if contains_any(combined, RISK_KEYWORDS) else "普通/读取/导航"
|
||
if "钱包" in scene or "交易" in scene:
|
||
risk = "高风险需安全闸"
|
||
evidence = "服务器路由/RPC、参数、控制通道、微信前台截图、返回字段摘要、黑屏检测、失败原因"
|
||
if risk == "高风险需安全闸":
|
||
evidence += "、确认前页面、人工确认状态"
|
||
return scene, core_level, risk, reason, evidence
|
||
|
||
|
||
def parse_py_routes(path: Path) -> list[InterfaceItem]:
|
||
items: list[InterfaceItem] = []
|
||
try:
|
||
text = path.read_text(encoding="utf-8")
|
||
except UnicodeDecodeError:
|
||
text = path.read_text(encoding="utf-8", errors="ignore")
|
||
lines = text.splitlines()
|
||
route_decorator_re = re.compile(r"@(?P<obj>router|app)\.(?P<method>get|post|put|delete|patch)\((?P<args>.*)")
|
||
def_re = re.compile(r"^(async\s+def|def)\s+(?P<name>[A-Za-z_][A-Za-z0-9_]*)\s*\(")
|
||
pending: list[tuple[int, str, str, str]] = []
|
||
for idx, line in enumerate(lines, 1):
|
||
m = route_decorator_re.search(line.strip())
|
||
if m:
|
||
args = m.group("args")
|
||
path_m = re.search(r"[\"']([^\"']+)[\"']", args)
|
||
tag_m = re.search(r"tags\s*=\s*\[\s*[\"']([^\"']+)[\"']", args)
|
||
route = path_m.group(1) if path_m else ""
|
||
tag = tag_m.group(1) if tag_m else ""
|
||
pending.append((idx, m.group("method").upper(), route, tag))
|
||
continue
|
||
dm = def_re.search(line.strip())
|
||
if dm and pending:
|
||
dec_line, method, route, tag = pending[-1]
|
||
function = dm.group("name")
|
||
context = "\n".join(lines[max(0, idx-4):min(len(lines), idx+12)])
|
||
scene, core_level, risk, reason, evidence = classify(f"{method} {route} {tag} {function} {context}", rel(path))
|
||
items.append(InterfaceItem("HTTP路由", rel(path), dec_line, method, route, function, tag, scene, core_level, risk, reason, evidence, f"{method} {route} -> {function}"))
|
||
pending.clear()
|
||
return items
|
||
|
||
|
||
def parse_py_symbols(path: Path) -> list[InterfaceItem]:
|
||
items: list[InterfaceItem] = []
|
||
try:
|
||
text = path.read_text(encoding="utf-8")
|
||
tree = ast.parse(text)
|
||
except Exception:
|
||
return items
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
|
||
name = node.name
|
||
if name.startswith("_"):
|
||
continue
|
||
src = ast.get_source_segment(text, node) or name
|
||
scene, core_level, risk, reason, evidence = classify(f"{name} {src[:1200]}", rel(path))
|
||
if core_level in {"核心微信业务接口", "微信控制支撑接口", "待人工归类的微信疑似接口"}:
|
||
items.append(InterfaceItem("Python符号", rel(path), getattr(node, "lineno", 0), "SYMBOL", name, name, "", scene, core_level, risk, reason, evidence, f"SYMBOL {name}"))
|
||
return items
|
||
|
||
|
||
def parse_js_exports(path: Path) -> list[InterfaceItem]:
|
||
items: list[InterfaceItem] = []
|
||
try:
|
||
text = path.read_text(encoding="utf-8")
|
||
except UnicodeDecodeError:
|
||
text = path.read_text(encoding="utf-8", errors="ignore")
|
||
patterns = [
|
||
re.compile(r"(?P<name>[A-Za-z_][A-Za-z0-9_]*)\s*:\s*function\s*\("),
|
||
re.compile(r"(?P<name>[A-Za-z_][A-Za-z0-9_]*)\s*\([^)]*\)\s*\{"),
|
||
re.compile(r"rpc\.exports\.(?P<name>[A-Za-z_][A-Za-z0-9_]*)\s*="),
|
||
]
|
||
lines = text.splitlines()
|
||
for i, line in enumerate(lines, 1):
|
||
for pat in patterns:
|
||
m = pat.search(line)
|
||
if not m:
|
||
continue
|
||
name = m.group("name")
|
||
if name in {"if", "for", "while", "switch", "function"}:
|
||
continue
|
||
context = "\n".join(lines[max(0, i-5):min(len(lines), i+15)])
|
||
scene, core_level, risk, reason, evidence = classify(f"{name} {context}", rel(path))
|
||
if core_level in {"核心微信业务接口", "微信控制支撑接口", "待人工归类的微信疑似接口"}:
|
||
items.append(InterfaceItem("Frida/JS符号", rel(path), i, "RPC/JS", name, name, "", scene, core_level, risk, reason, evidence, f"RPC/JS {name}"))
|
||
break
|
||
return items
|
||
|
||
|
||
def unique_items(items: list[InterfaceItem]) -> list[InterfaceItem]:
|
||
seen = set()
|
||
out = []
|
||
for it in items:
|
||
key = (it.source_type, it.file, it.method, it.route_or_symbol, it.function)
|
||
if key not in seen:
|
||
seen.add(key)
|
||
out.append(it)
|
||
return out
|
||
|
||
|
||
def scan() -> list[InterfaceItem]:
|
||
items: list[InterfaceItem] = []
|
||
for base in PY_SOURCES:
|
||
if not base.exists():
|
||
continue
|
||
for path in base.rglob("*.py"):
|
||
if "__pycache__" in path.parts:
|
||
continue
|
||
items.extend(parse_py_routes(path))
|
||
items.extend(parse_py_symbols(path))
|
||
for base in JS_SOURCES:
|
||
if not base.exists():
|
||
continue
|
||
for path in base.rglob("*.js"):
|
||
if "node_modules" in path.parts:
|
||
continue
|
||
items.extend(parse_js_exports(path))
|
||
return unique_items(items)
|
||
|
||
|
||
def write_outputs(items: list[InterfaceItem]) -> tuple[Path, Path, Path]:
|
||
csv_path = OUT_DIR / f"微信SDK服务器控制接口矩阵_追加版_{TS}.csv"
|
||
json_path = OUT_DIR / f"微信SDK服务器控制接口矩阵_追加版_{TS}.json"
|
||
md_path = OUT_DIR / f"微信SDK服务器控制接口扫描_追加版_{TS}.md"
|
||
fields = list(asdict(items[0]).keys()) if items else ["source_type", "file", "line", "method", "route_or_symbol", "function", "tag", "scene", "core_level", "risk_level", "reason", "evidence_required", "full_key"]
|
||
with csv_path.open("w", newline="", encoding="utf-8-sig") as f:
|
||
writer = csv.DictWriter(f, fieldnames=fields)
|
||
writer.writeheader()
|
||
for it in items:
|
||
writer.writerow(asdict(it))
|
||
json_path.write_text(json.dumps([asdict(it) for it in items], ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
counts: dict[str, int] = {}
|
||
scene_counts: dict[str, int] = {}
|
||
for it in items:
|
||
counts[it.core_level] = counts.get(it.core_level, 0) + 1
|
||
scene_counts[it.scene] = scene_counts.get(it.scene, 0) + 1
|
||
core_items = [it for it in items if it.core_level in {"核心微信业务接口", "微信控制支撑接口", "待人工归类的微信疑似接口"}]
|
||
business_items = [it for it in items if it.core_level == "核心微信业务接口"]
|
||
support_items = [it for it in items if it.core_level == "微信控制支撑接口"]
|
||
suspicious_items = [it for it in items if it.core_level == "待人工归类的微信疑似接口"]
|
||
non_core_items = [it for it in items if it.core_level in {"非核心接口", "待确认非核心或遗漏接口"}]
|
||
|
||
def table(rows: list[InterfaceItem], limit: int = 220) -> str:
|
||
header = "| 序号 | 来源 | 文件:行 | 方法/符号 | 接口/函数 | 场景 | 核心级别 | 风险 | 证据要求 |\n|---:|---|---|---|---|---|---|---|---|"
|
||
lines = [header]
|
||
for n, it in enumerate(rows[:limit], 1):
|
||
file_line = f"{it.file}:{it.line}"
|
||
iface = it.route_or_symbol.replace("|", "\\|")
|
||
func = it.function.replace("|", "\\|")
|
||
lines.append(f"| {n} | {it.source_type} | `{file_line}` | `{it.method}` | `{iface}` / `{func}` | {it.scene} | {it.core_level} | {it.risk_level} | {it.evidence_required} |")
|
||
if len(rows) > limit:
|
||
lines.append(f"| | | | | | | | | 另有 {len(rows)-limit} 条详见 CSV/JSON。 |")
|
||
return "\n".join(lines)
|
||
|
||
md = f"""# 微信 SDK 服务器控制接口扫描（追加版）
|
||
|
||
> 作者：Manus AI
|
||
> 生成时间：2026-05-18 {datetime.now().strftime('%H:%M:%S')}
|
||
> 原则：**只追加，不覆盖旧文档**。本报告用于纠偏：核心验收对象不是泛化 110 项，而是 SDK 服务器能否真实控制手机端微信。
|
||
|
||
## 一、扫描结论
|
||
|
||
本次从 `sdk/app/routers`、`sdk/app/services`、`sdk/agent/hook`、`sdk/tests`、`tools`、`工作手机SDK补全包_微信Frida`、`补全包/微信Frida_SDK_20260518` 中提取服务器路由、Python 符号和 Frida/JS 符号。扫描结果仅作为接口矩阵初稿，下一步还必须结合真实调用链和真机证据复核。
|
||
|
||
| 指标 | 数量 |
|
||
|---|---:|
|
||
| 扫描出的候选接口/符号总数 | {len(items)} |
|
||
| 核心微信业务接口 | {len(business_items)} |
|
||
| 微信控制支撑接口 | {len(support_items)} |
|
||
| 待人工归类的微信疑似接口 | {len(suspicious_items)} |
|
||
| 非核心或待排除接口 | {len(non_core_items)} |
|
||
|
||
| 核心级别 | 数量 |
|
||
|---|---:|
|
||
"""
|
||
for k, v in sorted(counts.items(), key=lambda kv: (-kv[1], kv[0])):
|
||
md += f"| {k} | {v} |\n"
|
||
md += "\n| 场景 | 数量 |\n|---|---:|\n"
|
||
for k, v in sorted(scene_counts.items(), key=lambda kv: (-kv[1], kv[0])):
|
||
md += f"| {k} | {v} |\n"
|
||
|
||
md += f"""
|
||
|
||
## 二、核心微信业务接口初稿
|
||
|
||
下面这些接口/符号将优先纳入核心验收。每项都必须补齐“服务器入口、控制通道、微信端页面/数据、可见截图、返回字段摘要、黑屏检测、失败原因”。高风险动作只能验收到确认前或必须等待人工确认。
|
||
|
||
{table(business_items, 260)}
|
||
|
||
## 三、微信控制支撑接口初稿
|
||
|
||
这些接口不等于微信业务本身，但它们决定 SDK 服务器是否能稳定控制手机微信，包括无线连接、Frida、ADB、亮屏、解锁、截图、识别、恢复前台等。支撑接口通过后，才允许进行微信业务接口真机验收。
|
||
|
||
{table(support_items, 180)}
|
||
|
||
## 四、待人工归类的微信疑似接口
|
||
|
||
这些接口命中了好友、消息、群、朋友圈等语义，但代码上下文未明确绑定微信端控制链路。下一步需要人工归类和调用链复核，确认是否能落到 `com.tencent.mm`。
|
||
|
||
{table(suspicious_items, 120)}
|
||
|
||
## 五、非核心或待排除接口说明
|
||
|
||
非微信、泛设备、项目管理、网关、注册中心、经验库、服务发现、Mock 或只有契约没有真机证据的接口，不能计入核心微信控制完成率。若其中某个泛设备接口是微信控制必需能力，应移入“微信控制支撑接口”，但仍不应伪装成微信业务功能。
|
||
|
||
## 六、下一步
|
||
|
||
下一步将基于本扫描结果重建“微信控制接口矩阵”，按好友、消息、群、朋友圈、搜索、资料、钱包安全闸、设备支撑八类重新归档。旧 110 项只作为历史记录，不再作为核心完成率依据。所有新结论将继续追加写入开发文档和项目落地执行表，不覆盖旧文件。
|
||
|
||
## 七、关联文件
|
||
|
||
| 类型 | 路径 |
|
||
|---|---|
|
||
| CSV 矩阵 | `{csv_path}` |
|
||
| JSON 明细 | `{json_path}` |
|
||
| 本报告 | `{md_path}` |
|
||
"""
|
||
md_path.write_text(md, encoding="utf-8")
|
||
return md_path, csv_path, json_path
|
||
|
||
|
||
def main() -> None:
|
||
items = scan()
|
||
items.sort(key=lambda x: (x.core_level, x.scene, x.file, x.line, x.function))
|
||
md, csvp, jsonp = write_outputs(items)
|
||
print(json.dumps({"items": len(items), "md": str(md), "csv": str(csvp), "json": str(jsonp)}, ensure_ascii=False, indent=2))
|
||
|
||
if __name__ == "__main__":
|
||
main()
|