diff --git a/工作手机SDK补全包_微信Frida/README.md b/工作手机SDK补全包_微信Frida/README.md deleted file mode 100644 index 0da98b5504..0000000000 --- a/工作手机SDK补全包_微信Frida/README.md +++ /dev/null @@ -1,16 +0,0 @@ -# 工作手机 SDK · 微信 Frida 无线控制补全包 - -> **⚠️ 归档原型 · 文档已并入主线** -> **请读**:[开发文档/9、手册/工作手机·五图总览与使用手册.md](../开发文档/9、手册/工作手机·五图总览与使用手册.md)(v2.0) -> **请用**:主线 `sdk/`(非本目录独立 server) - -本包与 `补全包/微信Frida_SDK_20260518/` 为同批归档;正文、九模块路径、无线部署均已写入开发文档主手册。 - -## 快速启动(现行) - -```bash -cd sdk/app && python3 -m uvicorn main:app --host 0.0.0.0 --port 8899 -cd sdk/agent && python3 agent.py --server ws://127.0.0.1:8899/ws/device/ -``` - -Hook 替换与 RPC 四层兼容见主手册 **§七**。 diff --git a/工作手机SDK补全包_微信Frida/generate_impl.py b/工作手机SDK补全包_微信Frida/generate_impl.py deleted file mode 100644 index 63f3879cfb..0000000000 --- a/工作手机SDK补全包_微信Frida/generate_impl.py +++ /dev/null @@ -1,578 +0,0 @@ -from pathlib import Path - -ROOT = Path('/home/ubuntu/work_phone_sdk_completion') -files = {} - -files['sdk/frida/frida_manager.py'] = r'''""" -工作手机 SDK · FridaManager - -负责手机端或服务器端连接 Frida、attach 微信进程、加载 Hook 脚本,并提供兼容 RPC 调用。 -设计重点:历史归档显示 Frida Python 对 rpc.exports 方法名存在混淆,因此这里采用多候选名兼容策略。 -""" -from __future__ import annotations - -from dataclasses import dataclass, field -from pathlib import Path -import re -import time -from typing import Any, Callable, Dict, Iterable, Optional - - -class FridaUnavailable(RuntimeError): - """当前环境未安装或无法连接 Frida。""" - - -class RpcMethodMissing(RuntimeError): - """Hook 中缺少指定 RPC 方法。""" - - -@dataclass -class FridaConfig: - device_host: str = "127.0.0.1" - device_port: int = 27042 - package_name: str = "com.tencent.mm" - process_name: str = "WeChat" - attach_timeout: float = 15.0 - prefer_usb: bool = False - hook_path: Optional[str] = None - - -@dataclass -class RpcResult: - ok: bool - action: str - method: str - data: Any = None - error: Optional[str] = None - elapsed_ms: int = 0 - - -class FridaManager: - def __init__(self, config: Optional[FridaConfig] = None): - self.config = config or FridaConfig() - self.frida = None - self.device = None - self.session = None - self.script = None - self.exports = None - self.loaded_hook_path: Optional[str] = None - - def _import_frida(self): - if self.frida is not None: - return self.frida - try: - import frida # type: ignore - except Exception as exc: - raise FridaUnavailable(f"未安装 frida Python 包或加载失败:{exc}") from exc - self.frida = frida - return frida - - def connect(self): - frida = self._import_frida() - if self.config.prefer_usb: - self.device = frida.get_usb_device(timeout=int(self.config.attach_timeout)) - else: - self.device = frida.get_device_manager().add_remote_device( - f"{self.config.device_host}:{self.config.device_port}" - ) - return self.device - - def attach(self, target: Optional[str] = None): - if self.device is None: - self.connect() - assert self.device is not None - target = target or self.config.package_name - last_error: Optional[Exception] = None - candidates: Iterable[Any] = [target, self.config.process_name] - for candidate in candidates: - try: - self.session = self.device.attach(candidate) - return self.session - except Exception as exc: # pragma: no cover - depends on phone - last_error = exc - try: - processes = self.device.enumerate_processes() - for proc in processes: - name = getattr(proc, 'name', '') or '' - pid = getattr(proc, 'pid', None) - if pid and (self.config.package_name in name or self.config.process_name.lower() in name.lower() or '微信' in name): - self.session = self.device.attach(pid) - return self.session - except Exception as exc: # pragma: no cover - last_error = exc - raise FridaUnavailable(f"无法 attach 微信进程:{last_error}") - - def load_script(self, hook_path: Optional[str] = None): - path = Path(hook_path or self.config.hook_path or '') - if not path.exists(): - raise FileNotFoundError(f"Hook 脚本不存在:{path}") - if self.session is None: - self.attach() - assert self.session is not None - source = path.read_text(encoding='utf-8') - self.script = self.session.create_script(source) - self.script.on('message', self._on_message) - self.script.load() - self.exports = getattr(self.script, 'exports_sync', None) or getattr(self.script, 'exports', None) - self.loaded_hook_path = str(path) - return self.script - - def _on_message(self, message, data): # pragma: no cover - runtime callback - # 生产环境可转发到 EventReporter;这里保持最小日志。 - print({'frida_message': message, 'data_len': len(data) if data else 0}) - - @staticmethod - def _snake(name: str) -> str: - return re.sub(r'(? RpcResult: - started = time.time() - if self.exports is None: - raise FridaUnavailable('Hook 脚本尚未加载,无法调用 RPC') - last_error = None - for candidate in self.method_candidates(method): - try: - fn = getattr(self.exports, candidate) - data = fn(*args, **kwargs) - return RpcResult(True, method, candidate, data=data, elapsed_ms=int((time.time() - started) * 1000)) - except AttributeError as exc: - last_error = exc - except Exception as exc: - return RpcResult(False, method, candidate, error=str(exc), elapsed_ms=int((time.time() - started) * 1000)) - # 兼容新版 frida 的 exports_sync.invoke 或 script.exports_sync.invoke - try: - invoke = getattr(self.exports, 'invoke') - data = invoke(method, list(args), kwargs or None) - return RpcResult(True, method, 'invoke', data=data, elapsed_ms=int((time.time() - started) * 1000)) - except Exception as exc: - last_error = exc - return RpcResult(False, method, '', error=f'RPC 方法不存在或不可调用:{method}; last={last_error}', elapsed_ms=int((time.time() - started) * 1000)) - - def cleanup(self): - for obj, method in [(self.script, 'unload'), (self.session, 'detach')]: - if obj is not None: - try: - getattr(obj, method)() - except Exception: - pass - self.script = None - self.session = None - self.exports = None -''' - -files['sdk/wechat/wechat_actions.py'] = r'''"""微信动作目录:服务器、手机端 Agent、验证脚本共同使用。""" -from __future__ import annotations -from dataclasses import dataclass, field -from typing import Any, Dict, List - -@dataclass(frozen=True) -class WechatAction: - action: str - rpc: str - module: str - description: str - required: List[str] = field(default_factory=list) - safe: bool = True - verify: bool = True - -ACTIONS: List[WechatAction] = [ - WechatAction('ping', 'ping', 'system', 'Hook 连通性检测'), - WechatAction('get_connection_status', 'getConnectionStatus', 'system', '读取连接状态'), - WechatAction('screenshot', 'takeScreenshot', 'device', '手机当前屏幕截图'), - WechatAction('ui_dump', 'dumpUiTree', 'device', '导出当前 UI 树'), - WechatAction('go_home', 'goHome', 'navigation', '返回微信首页'), - WechatAction('open_contacts', 'openContacts', 'navigation', '打开通讯录'), - WechatAction('open_chats', 'openChats', 'navigation', '打开聊天列表'), - WechatAction('search_contact', 'searchContact', 'contact', '搜索联系人', ['keyword']), - WechatAction('get_contacts', 'getContacts', 'contact', '读取联系人列表'), - WechatAction('get_contact_profile', 'getContactProfile', 'contact', '读取联系人资料', ['wxid']), - WechatAction('add_friend', 'addFriend', 'contact', '添加好友', ['keyword'], safe=False), - WechatAction('accept_friend', 'acceptFriend', 'contact', '通过好友申请', ['wxid'], safe=False), - WechatAction('send_text', 'sendTextMessage', 'message', '发送文本消息', ['to', 'text'], safe=False), - WechatAction('send_image', 'sendImageMessage', 'message', '发送图片消息', ['to', 'path'], safe=False), - WechatAction('get_messages', 'getMessages', 'message', '读取消息列表', ['wxid']), - WechatAction('open_chat', 'openChat', 'message', '打开指定会话', ['wxid']), - WechatAction('create_group', 'createGroup', 'group', '创建群聊', ['members'], safe=False), - WechatAction('invite_group_member', 'inviteGroupMember', 'group', '邀请群成员', ['chatroom', 'members'], safe=False), - WechatAction('get_group_members', 'getGroupMembers', 'group', '读取群成员', ['chatroom']), - WechatAction('browse_channels', 'browseChannels', 'channels', '浏览视频号'), - WechatAction('add_favorite', 'addFavorite', 'favorite', '收藏当前内容'), - WechatAction('add_custom_emoji', 'addCustomEmoji', 'emoji', '添加自定义表情', safe=False), -] - -ACTION_MAP: Dict[str, WechatAction] = {item.action: item for item in ACTIONS} -RPC_MAP: Dict[str, WechatAction] = {item.rpc: item for item in ACTIONS} - -def list_actions() -> List[Dict[str, Any]]: - return [item.__dict__.copy() for item in ACTIONS] - -def validate_action(action: str, payload: Dict[str, Any]) -> WechatAction: - if action not in ACTION_MAP: - raise KeyError(f'未知微信动作:{action}') - spec = ACTION_MAP[action] - missing = [k for k in spec.required if k not in payload or payload[k] in (None, '')] - if missing: - raise ValueError(f'动作 {action} 缺少参数:{missing}') - return spec -''' - -files['sdk/frida/hook_executor.py'] = r'''"""将业务 action 映射到 Frida RPC。""" -from __future__ import annotations -from dataclasses import asdict -from typing import Any, Dict, Optional - -from sdk.frida.frida_manager import FridaManager, RpcResult -from sdk.wechat.wechat_actions import validate_action, list_actions - -class HookExecutor: - def __init__(self, manager: FridaManager): - self.manager = manager - - def execute(self, action: str, payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: - payload = payload or {} - try: - spec = validate_action(action, payload) - except Exception as exc: - return {'ok': False, 'action': action, 'error': str(exc), 'stage': 'validate'} - args = [payload[k] for k in spec.required] - # 允许额外参数透传给 Hook;Hook 若不支持会返回错误,便于报告定位。 - extra = {k: v for k, v in payload.items() if k not in spec.required} - result: RpcResult = self.manager.call(spec.rpc, *args, **extra) - return { - 'ok': result.ok, - 'action': action, - 'rpc': spec.rpc, - 'method_used': result.method, - 'module': spec.module, - 'description': spec.description, - 'safe': spec.safe, - 'data': result.data, - 'error': result.error, - 'elapsed_ms': result.elapsed_ms, - } - - def catalog(self): - return list_actions() -''' - -files['mobile_agent/wireless_agent.py'] = r'''"""手机端 Termux/Python Agent:通过 WebSocket 接收服务器动作,调用 Frida Hook。""" -from __future__ import annotations -import argparse -import asyncio -import json -import platform -import uuid -from pathlib import Path - -from sdk.frida.frida_manager import FridaConfig, FridaManager -from sdk.frida.hook_executor import HookExecutor - -try: - import websockets -except Exception: # pragma: no cover - websockets = None - -class WirelessAgent: - def __init__(self, server_ws: str, hook_path: str, device_id: str | None = None): - self.server_ws = server_ws - self.hook_path = hook_path - self.device_id = device_id or f"phone-{uuid.getnode():x}" - self.manager = FridaManager(FridaConfig(hook_path=hook_path)) - self.executor = HookExecutor(self.manager) - - async def boot(self): - self.manager.connect() - self.manager.attach() - self.manager.load_script(self.hook_path) - - def hello(self): - return { - 'type': 'hello', - 'device_id': self.device_id, - 'platform': platform.platform(), - 'actions': self.executor.catalog(), - } - - async def run(self): - if websockets is None: - raise RuntimeError('请先安装 websockets:pip install websockets') - await self.boot() - async with websockets.connect(self.server_ws, ping_interval=20, ping_timeout=20) as ws: - await ws.send(json.dumps(self.hello(), ensure_ascii=False)) - async for raw in ws: - try: - msg = json.loads(raw) - if msg.get('type') != 'command': - continue - result = self.executor.execute(msg['action'], msg.get('payload') or {}) - await ws.send(json.dumps({'type': 'result', 'request_id': msg.get('request_id'), 'device_id': self.device_id, 'result': result}, ensure_ascii=False)) - except Exception as exc: - await ws.send(json.dumps({'type': 'error', 'error': str(exc)}, ensure_ascii=False)) - - -def main(): - parser = argparse.ArgumentParser() - parser.add_argument('--server-ws', required=True, help='例如 ws://192.168.1.10:8000/ws/phone') - parser.add_argument('--hook', required=True, help='wechat_hook_bridge.js 或真实 wechat_hook_v3.js 路径') - parser.add_argument('--device-id') - args = parser.parse_args() - asyncio.run(WirelessAgent(args.server_ws, args.hook, args.device_id).run()) - -if __name__ == '__main__': - main() -''' - -files['server/routes/wechat_frida.py'] = r'''"""FastAPI 微信 Frida 无线控制路由。""" -from __future__ import annotations -import asyncio -import json -import uuid -from dataclasses import dataclass, field -from typing import Any, Dict, Optional - -from fastapi import APIRouter, WebSocket, WebSocketDisconnect, HTTPException -from pydantic import BaseModel - -from sdk.wechat.wechat_actions import list_actions - -router = APIRouter(prefix='/api/v3/wechat-frida', tags=['wechat-frida']) - -@dataclass -class PhoneConn: - device_id: str - ws: WebSocket - meta: Dict[str, Any] = field(default_factory=dict) - pending: Dict[str, asyncio.Future] = field(default_factory=dict) - -phones: Dict[str, PhoneConn] = {} - -class ActionRequest(BaseModel): - device_id: str - action: str - payload: Dict[str, Any] = {} - timeout: float = 30.0 - -@router.get('/actions') -def actions(): - return {'ok': True, 'actions': list_actions()} - -@router.get('/devices') -def devices(): - return {'ok': True, 'devices': [{'device_id': k, 'meta': v.meta} for k, v in phones.items()]} - -@router.post('/execute') -async def execute(req: ActionRequest): - if req.device_id not in phones: - raise HTTPException(404, f'设备未连接:{req.device_id}') - conn = phones[req.device_id] - request_id = uuid.uuid4().hex - fut = asyncio.get_event_loop().create_future() - conn.pending[request_id] = fut - await conn.ws.send_text(json.dumps({'type': 'command', 'request_id': request_id, 'action': req.action, 'payload': req.payload}, ensure_ascii=False)) - try: - return await asyncio.wait_for(fut, timeout=req.timeout) - finally: - conn.pending.pop(request_id, None) - -async def phone_socket(websocket: WebSocket): - await websocket.accept() - device_id: Optional[str] = None - try: - async for raw in websocket.iter_text(): - msg = json.loads(raw) - if msg.get('type') == 'hello': - device_id = msg.get('device_id') or uuid.uuid4().hex - phones[device_id] = PhoneConn(device_id=device_id, ws=websocket, meta=msg) - await websocket.send_text(json.dumps({'type': 'hello_ack', 'device_id': device_id}, ensure_ascii=False)) - elif msg.get('type') == 'result': - rid = msg.get('request_id') - did = msg.get('device_id') or device_id - conn = phones.get(did or '') - if conn and rid in conn.pending and not conn.pending[rid].done(): - conn.pending[rid].set_result({'ok': True, 'device_id': did, 'result': msg.get('result')}) - elif msg.get('type') == 'error': - # 保留连接,错误由客户端下一次请求再显式返回。 - pass - except WebSocketDisconnect: - pass - finally: - if device_id and device_id in phones: - phones.pop(device_id, None) -''' - -files['server/app.py'] = r'''from fastapi import FastAPI, WebSocket -from server.routes.wechat_frida import router as wechat_frida_router, phone_socket - -app = FastAPI(title='工作手机 SDK · 微信 Frida 控制服务', version='0.3.0') -app.include_router(wechat_frida_router) - -@app.get('/health') -def health(): - return {'ok': True, 'service': 'work-phone-sdk'} - -@app.websocket('/ws/phone') -async def ws_phone(websocket: WebSocket): - await phone_socket(websocket) -''' - -files['hooks/wechat_hook_bridge.js'] = r'''// 工作手机 SDK · 微信 Hook 桥接模板 -// 说明:这是安全桥接版,优先提供连通性、截图、UI 导航等通用方法。 -// 如果已有完整 wechat_hook_v3.js,可直接替换本文件;Python SDK 会兼容 camelCase/snake_case 调用。 -'use strict'; - -function ok(data) { return { ok: true, data: data || null, ts: Date.now() }; } -function fail(message) { return { ok: false, error: String(message), ts: Date.now() }; } - -function runJava(fn) { - let result; - Java.perform(function () { - try { result = fn(); } catch (e) { result = fail(e.stack || e.message || e); } - }); - return result; -} - -rpc.exports = { - ping: function () { return 'pong from wechat_hook_bridge'; }, - getConnectionStatus: function () { - return ok({ java_available: Java.available, process: Process.id, arch: Process.arch, platform: Process.platform }); - }, - takeScreenshot: function () { - // 截图建议在手机端通过 uiautomator/screencap 实现;Hook 层返回占位,Agent 可扩展真实文件路径。 - return ok({ mode: 'placeholder', message: '请在 mobile_agent 扩展 screencap -p 后回传文件路径' }); - }, - dumpUiTree: function () { return ok({ mode: 'placeholder', message: '建议通过 uiautomator dump 获取 UI XML' }); }, - goHome: function () { return ok({ action: 'goHome', message: '桥接模板未执行 UI 点击;请替换完整 Hook 后验证' }); }, - openContacts: function () { return ok({ action: 'openContacts', message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - openChats: function () { return ok({ action: 'openChats', message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - searchContact: function (keyword) { return ok({ keyword: keyword, message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - getContacts: function () { return ok({ contacts: [], message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - getContactProfile: function (wxid) { return ok({ wxid: wxid, message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - addFriend: function (keyword) { return fail('危险动作未在模板中实现:addFriend ' + keyword); }, - acceptFriend: function (wxid) { return fail('危险动作未在模板中实现:acceptFriend ' + wxid); }, - sendTextMessage: function (to, text) { return fail('危险动作未在模板中实现:sendTextMessage ' + to); }, - sendImageMessage: function (to, path) { return fail('危险动作未在模板中实现:sendImageMessage ' + to); }, - getMessages: function (wxid) { return ok({ wxid: wxid, messages: [], message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - openChat: function (wxid) { return ok({ wxid: wxid, message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - createGroup: function (members) { return fail('危险动作未在模板中实现:createGroup'); }, - inviteGroupMember: function (chatroom, members) { return fail('危险动作未在模板中实现:inviteGroupMember'); }, - getGroupMembers: function (chatroom) { return ok({ chatroom: chatroom, members: [], message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - browseChannels: function () { return ok({ message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - addFavorite: function () { return fail('危险动作未在模板中实现:addFavorite'); }, - addCustomEmoji: function () { return fail('危险动作未在模板中实现:addCustomEmoji'); } -}; -''' - -files['scripts/verify_wechat_frida.py'] = r'''"""一键验证服务器控制手机微信功能,输出 JSON 与 Markdown 报告。""" -from __future__ import annotations -import argparse -import json -import time -from pathlib import Path -from typing import Any, Dict - -import requests - -from sdk.wechat.wechat_actions import ACTIONS - -DEFAULT_PAYLOADS: Dict[str, Dict[str, Any]] = { - 'search_contact': {'keyword': '文件传输助手'}, - 'get_contact_profile': {'wxid': 'filehelper'}, - 'add_friend': {'keyword': '__dry_run__'}, - 'accept_friend': {'wxid': '__dry_run__'}, - 'send_text': {'to': 'filehelper', 'text': '工作手机SDK自动验证'}, - 'send_image': {'to': 'filehelper', 'path': '/sdcard/Pictures/test.png'}, - 'get_messages': {'wxid': 'filehelper'}, - 'open_chat': {'wxid': 'filehelper'}, - 'create_group': {'members': ['filehelper']}, - 'invite_group_member': {'chatroom': '__dry_run__', 'members': ['filehelper']}, - 'get_group_members': {'chatroom': '__dry_run__'}, -} - -def call(base_url: str, device_id: str, action: str, payload: Dict[str, Any], timeout: float): - r = requests.post(f'{base_url}/api/v3/wechat-frida/execute', json={'device_id': device_id, 'action': action, 'payload': payload, 'timeout': timeout}, timeout=timeout + 5) - try: - return r.status_code, r.json() - except Exception: - return r.status_code, {'ok': False, 'text': r.text} - -def main(): - parser = argparse.ArgumentParser() - parser.add_argument('--base-url', default='http://127.0.0.1:8000') - parser.add_argument('--device-id', required=True) - parser.add_argument('--out-dir', default='reports') - parser.add_argument('--include-dangerous', action='store_true', help='默认不执行加好友/发消息等危险动作,只记录 SKIPPED') - parser.add_argument('--timeout', type=float, default=30) - args = parser.parse_args() - - out_dir = Path(args.out_dir) - out_dir.mkdir(parents=True, exist_ok=True) - rows = [] - for spec in ACTIONS: - if not spec.safe and not args.include_dangerous: - rows.append({'action': spec.action, 'rpc': spec.rpc, 'ok': None, 'status': 'SKIPPED_DANGEROUS', 'description': spec.description}) - continue - status, data = call(args.base_url, args.device_id, spec.action, DEFAULT_PAYLOADS.get(spec.action, {}), args.timeout) - rows.append({'action': spec.action, 'rpc': spec.rpc, 'http_status': status, 'ok': bool(data.get('ok') and (data.get('result') or {}).get('ok', True)), 'response': data, 'description': spec.description}) - time.sleep(0.2) - - stamp = time.strftime('%Y%m%d_%H%M%S') - json_path = out_dir / f'wechat_frida_verify_{stamp}.json' - md_path = out_dir / f'wechat_frida_verify_{stamp}.md' - json_path.write_text(json.dumps(rows, ensure_ascii=False, indent=2), encoding='utf-8') - lines = ['# 微信 Frida 功能验证报告', '', f'- 时间:{time.strftime("%Y-%m-%d %H:%M:%S")}', f'- 设备:`{args.device_id}`', '', '| 动作 | RPC | 结果 | 说明 |', '|---|---|---|---|'] - for row in rows: - result = 'SKIP' if row.get('status') else ('PASS' if row.get('ok') else 'FAIL') - lines.append(f"| `{row['action']}` | `{row['rpc']}` | {result} | {row.get('description','')} |") - md_path.write_text('\n'.join(lines) + '\n', encoding='utf-8') - print(json_path) - print(md_path) - -if __name__ == '__main__': - main() -''' - -files['requirements.txt'] = 'fastapi\nuvicorn\nrequests\nwebsockets\nfrida\n\n' -files['README.md'] = r'''# 工作手机 SDK · 微信 Frida 无线控制补全包 - -本包用于承接上传 ZIP 中断的开发进度,补齐 **服务器 → WebSocket → 手机端 Agent → Frida → 微信 Hook** 的闭环。 - -## 快速启动 - -服务器端: - -```bash -pip install -r requirements.txt -uvicorn server.app:app --host 0.0.0.0 --port 8000 -``` - -手机 Termux 端: - -```bash -python mobile_agent/wireless_agent.py --server-ws ws://服务器IP:8000/ws/phone --hook hooks/wechat_hook_bridge.js -``` - -验证端: - -```bash -python scripts/verify_wechat_frida.py --base-url http://服务器IP:8000 --device-id phone-xxxx -``` - -如果已有完整 `wechat_hook_v3.js`,请替换 `hooks/wechat_hook_bridge.js`,SDK 会自动兼容 camelCase / snake_case / lowercase 方法名。 -''' - -for rel, content in files.items(): - path = ROOT / rel - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(content, encoding='utf-8') -print(f'generated {len(files)} files under {ROOT}') diff --git a/工作手机SDK补全包_微信Frida/hooks/wechat_hook_bridge.js b/工作手机SDK补全包_微信Frida/hooks/wechat_hook_bridge.js deleted file mode 100644 index d69423fa67..0000000000 --- a/工作手机SDK补全包_微信Frida/hooks/wechat_hook_bridge.js +++ /dev/null @@ -1,45 +0,0 @@ -// 工作手机 SDK · 微信 Hook 桥接模板 -// 说明:这是安全桥接版,优先提供连通性、截图、UI 导航等通用方法。 -// 如果已有完整 wechat_hook_v3.js,可直接替换本文件;Python SDK 会兼容 camelCase/snake_case 调用。 -'use strict'; - -function ok(data) { return { ok: true, data: data || null, ts: Date.now() }; } -function fail(message) { return { ok: false, error: String(message), ts: Date.now() }; } - -function runJava(fn) { - let result; - Java.perform(function () { - try { result = fn(); } catch (e) { result = fail(e.stack || e.message || e); } - }); - return result; -} - -rpc.exports = { - ping: function () { return 'pong from wechat_hook_bridge'; }, - getConnectionStatus: function () { - return ok({ java_available: Java.available, process: Process.id, arch: Process.arch, platform: Process.platform }); - }, - takeScreenshot: function () { - // 截图建议在手机端通过 uiautomator/screencap 实现;Hook 层返回占位,Agent 可扩展真实文件路径。 - return ok({ mode: 'placeholder', message: '请在 mobile_agent 扩展 screencap -p 后回传文件路径' }); - }, - dumpUiTree: function () { return ok({ mode: 'placeholder', message: '建议通过 uiautomator dump 获取 UI XML' }); }, - goHome: function () { return ok({ action: 'goHome', message: '桥接模板未执行 UI 点击;请替换完整 Hook 后验证' }); }, - openContacts: function () { return ok({ action: 'openContacts', message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - openChats: function () { return ok({ action: 'openChats', message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - searchContact: function (keyword) { return ok({ keyword: keyword, message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - getContacts: function () { return ok({ contacts: [], message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - getContactProfile: function (wxid) { return ok({ wxid: wxid, message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - addFriend: function (keyword) { return fail('危险动作未在模板中实现:addFriend ' + keyword); }, - acceptFriend: function (wxid) { return fail('危险动作未在模板中实现:acceptFriend ' + wxid); }, - sendTextMessage: function (to, text) { return fail('危险动作未在模板中实现:sendTextMessage ' + to); }, - sendImageMessage: function (to, path) { return fail('危险动作未在模板中实现:sendImageMessage ' + to); }, - getMessages: function (wxid) { return ok({ wxid: wxid, messages: [], message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - openChat: function (wxid) { return ok({ wxid: wxid, message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - createGroup: function (members) { return fail('危险动作未在模板中实现:createGroup'); }, - inviteGroupMember: function (chatroom, members) { return fail('危险动作未在模板中实现:inviteGroupMember'); }, - getGroupMembers: function (chatroom) { return ok({ chatroom: chatroom, members: [], message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - browseChannels: function () { return ok({ message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - addFavorite: function () { return fail('危险动作未在模板中实现:addFavorite'); }, - addCustomEmoji: function () { return fail('危险动作未在模板中实现:addCustomEmoji'); } -}; diff --git a/工作手机SDK补全包_微信Frida/mobile_agent/__init__.py b/工作手机SDK补全包_微信Frida/mobile_agent/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/工作手机SDK补全包_微信Frida/mobile_agent/wireless_agent.py b/工作手机SDK补全包_微信Frida/mobile_agent/wireless_agent.py deleted file mode 100644 index 594bd01f44..0000000000 --- a/工作手机SDK补全包_微信Frida/mobile_agent/wireless_agent.py +++ /dev/null @@ -1,65 +0,0 @@ -"""手机端 Termux/Python Agent:通过 WebSocket 接收服务器动作,调用 Frida Hook。""" -from __future__ import annotations -import argparse -import asyncio -import json -import platform -import uuid -from pathlib import Path - -from sdk.frida.frida_manager import FridaConfig, FridaManager -from sdk.frida.hook_executor import HookExecutor - -try: - import websockets -except Exception: # pragma: no cover - websockets = None - -class WirelessAgent: - def __init__(self, server_ws: str, hook_path: str, device_id: str | None = None): - self.server_ws = server_ws - self.hook_path = hook_path - self.device_id = device_id or f"phone-{uuid.getnode():x}" - self.manager = FridaManager(FridaConfig(hook_path=hook_path)) - self.executor = HookExecutor(self.manager) - - async def boot(self): - self.manager.connect() - self.manager.attach() - self.manager.load_script(self.hook_path) - - def hello(self): - return { - 'type': 'hello', - 'device_id': self.device_id, - 'platform': platform.platform(), - 'actions': self.executor.catalog(), - } - - async def run(self): - if websockets is None: - raise RuntimeError('请先安装 websockets:pip install websockets') - await self.boot() - async with websockets.connect(self.server_ws, ping_interval=20, ping_timeout=20) as ws: - await ws.send(json.dumps(self.hello(), ensure_ascii=False)) - async for raw in ws: - try: - msg = json.loads(raw) - if msg.get('type') != 'command': - continue - result = self.executor.execute(msg['action'], msg.get('payload') or {}) - await ws.send(json.dumps({'type': 'result', 'request_id': msg.get('request_id'), 'device_id': self.device_id, 'result': result}, ensure_ascii=False)) - except Exception as exc: - await ws.send(json.dumps({'type': 'error', 'error': str(exc)}, ensure_ascii=False)) - - -def main(): - parser = argparse.ArgumentParser() - parser.add_argument('--server-ws', required=True, help='例如 ws://192.168.1.10:8000/ws/phone') - parser.add_argument('--hook', required=True, help='wechat_hook_bridge.js 或真实 wechat_hook_v3.js 路径') - parser.add_argument('--device-id') - args = parser.parse_args() - asyncio.run(WirelessAgent(args.server_ws, args.hook, args.device_id).run()) - -if __name__ == '__main__': - main() diff --git a/工作手机SDK补全包_微信Frida/requirements.txt b/工作手机SDK补全包_微信Frida/requirements.txt deleted file mode 100644 index b52d28ca18..0000000000 --- a/工作手机SDK补全包_微信Frida/requirements.txt +++ /dev/null @@ -1,6 +0,0 @@ -fastapi -uvicorn -requests -websockets -frida - diff --git a/工作手机SDK补全包_微信Frida/sdk/__init__.py b/工作手机SDK补全包_微信Frida/sdk/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/工作手机SDK补全包_微信Frida/sdk/frida/__init__.py b/工作手机SDK补全包_微信Frida/sdk/frida/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/工作手机SDK补全包_微信Frida/sdk/frida/frida_manager.py b/工作手机SDK补全包_微信Frida/sdk/frida/frida_manager.py deleted file mode 100644 index 9ec11f589a..0000000000 --- a/工作手机SDK补全包_微信Frida/sdk/frida/frida_manager.py +++ /dev/null @@ -1,166 +0,0 @@ -""" -工作手机 SDK · FridaManager - -负责手机端或服务器端连接 Frida、attach 微信进程、加载 Hook 脚本,并提供兼容 RPC 调用。 -设计重点:历史归档显示 Frida Python 对 rpc.exports 方法名存在混淆,因此这里采用多候选名兼容策略。 -""" -from __future__ import annotations - -from dataclasses import dataclass, field -from pathlib import Path -import re -import time -from typing import Any, Callable, Dict, Iterable, Optional - - -class FridaUnavailable(RuntimeError): - """当前环境未安装或无法连接 Frida。""" - - -class RpcMethodMissing(RuntimeError): - """Hook 中缺少指定 RPC 方法。""" - - -@dataclass -class FridaConfig: - device_host: str = "127.0.0.1" - device_port: int = 27042 - package_name: str = "com.tencent.mm" - process_name: str = "WeChat" - attach_timeout: float = 15.0 - prefer_usb: bool = False - hook_path: Optional[str] = None - - -@dataclass -class RpcResult: - ok: bool - action: str - method: str - data: Any = None - error: Optional[str] = None - elapsed_ms: int = 0 - - -class FridaManager: - def __init__(self, config: Optional[FridaConfig] = None): - self.config = config or FridaConfig() - self.frida = None - self.device = None - self.session = None - self.script = None - self.exports = None - self.loaded_hook_path: Optional[str] = None - - def _import_frida(self): - if self.frida is not None: - return self.frida - try: - import frida # type: ignore - except Exception as exc: - raise FridaUnavailable(f"未安装 frida Python 包或加载失败:{exc}") from exc - self.frida = frida - return frida - - def connect(self): - frida = self._import_frida() - if self.config.prefer_usb: - self.device = frida.get_usb_device(timeout=int(self.config.attach_timeout)) - else: - self.device = frida.get_device_manager().add_remote_device( - f"{self.config.device_host}:{self.config.device_port}" - ) - return self.device - - def attach(self, target: Optional[str] = None): - if self.device is None: - self.connect() - assert self.device is not None - target = target or self.config.package_name - last_error: Optional[Exception] = None - candidates: Iterable[Any] = [target, self.config.process_name] - for candidate in candidates: - try: - self.session = self.device.attach(candidate) - return self.session - except Exception as exc: # pragma: no cover - depends on phone - last_error = exc - try: - processes = self.device.enumerate_processes() - for proc in processes: - name = getattr(proc, 'name', '') or '' - pid = getattr(proc, 'pid', None) - if pid and (self.config.package_name in name or self.config.process_name.lower() in name.lower() or '微信' in name): - self.session = self.device.attach(pid) - return self.session - except Exception as exc: # pragma: no cover - last_error = exc - raise FridaUnavailable(f"无法 attach 微信进程:{last_error}") - - def load_script(self, hook_path: Optional[str] = None): - path = Path(hook_path or self.config.hook_path or '') - if not path.exists(): - raise FileNotFoundError(f"Hook 脚本不存在:{path}") - if self.session is None: - self.attach() - assert self.session is not None - source = path.read_text(encoding='utf-8') - self.script = self.session.create_script(source) - self.script.on('message', self._on_message) - self.script.load() - self.exports = getattr(self.script, 'exports_sync', None) or getattr(self.script, 'exports', None) - self.loaded_hook_path = str(path) - return self.script - - def _on_message(self, message, data): # pragma: no cover - runtime callback - # 生产环境可转发到 EventReporter;这里保持最小日志。 - print({'frida_message': message, 'data_len': len(data) if data else 0}) - - @staticmethod - def _snake(name: str) -> str: - return re.sub(r'(? RpcResult: - started = time.time() - if self.exports is None: - raise FridaUnavailable('Hook 脚本尚未加载,无法调用 RPC') - last_error = None - for candidate in self.method_candidates(method): - try: - fn = getattr(self.exports, candidate) - data = fn(*args, **kwargs) - return RpcResult(True, method, candidate, data=data, elapsed_ms=int((time.time() - started) * 1000)) - except AttributeError as exc: - last_error = exc - except Exception as exc: - return RpcResult(False, method, candidate, error=str(exc), elapsed_ms=int((time.time() - started) * 1000)) - # 兼容新版 frida 的 exports_sync.invoke 或 script.exports_sync.invoke - try: - invoke = getattr(self.exports, 'invoke') - data = invoke(method, list(args), kwargs or None) - return RpcResult(True, method, 'invoke', data=data, elapsed_ms=int((time.time() - started) * 1000)) - except Exception as exc: - last_error = exc - return RpcResult(False, method, '', error=f'RPC 方法不存在或不可调用:{method}; last={last_error}', elapsed_ms=int((time.time() - started) * 1000)) - - def cleanup(self): - for obj, method in [(self.script, 'unload'), (self.session, 'detach')]: - if obj is not None: - try: - getattr(obj, method)() - except Exception: - pass - self.script = None - self.session = None - self.exports = None diff --git a/工作手机SDK补全包_微信Frida/sdk/frida/hook_executor.py b/工作手机SDK补全包_微信Frida/sdk/frida/hook_executor.py deleted file mode 100644 index da046d0d35..0000000000 --- a/工作手机SDK补全包_微信Frida/sdk/frida/hook_executor.py +++ /dev/null @@ -1,37 +0,0 @@ -"""将业务 action 映射到 Frida RPC。""" -from __future__ import annotations -from dataclasses import asdict -from typing import Any, Dict, Optional - -from sdk.frida.frida_manager import FridaManager, RpcResult -from sdk.wechat.wechat_actions import validate_action, list_actions - -class HookExecutor: - def __init__(self, manager: FridaManager): - self.manager = manager - - def execute(self, action: str, payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: - payload = payload or {} - try: - spec = validate_action(action, payload) - except Exception as exc: - return {'ok': False, 'action': action, 'error': str(exc), 'stage': 'validate'} - args = [payload[k] for k in spec.required] - # 允许额外参数透传给 Hook;Hook 若不支持会返回错误,便于报告定位。 - extra = {k: v for k, v in payload.items() if k not in spec.required} - result: RpcResult = self.manager.call(spec.rpc, *args, **extra) - return { - 'ok': result.ok, - 'action': action, - 'rpc': spec.rpc, - 'method_used': result.method, - 'module': spec.module, - 'description': spec.description, - 'safe': spec.safe, - 'data': result.data, - 'error': result.error, - 'elapsed_ms': result.elapsed_ms, - } - - def catalog(self): - return list_actions() diff --git a/工作手机SDK补全包_微信Frida/sdk/wechat/__init__.py b/工作手机SDK补全包_微信Frida/sdk/wechat/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/工作手机SDK补全包_微信Frida/sdk/wechat/wechat_actions.py b/工作手机SDK补全包_微信Frida/sdk/wechat/wechat_actions.py deleted file mode 100644 index c405fbb39a..0000000000 --- a/工作手机SDK补全包_微信Frida/sdk/wechat/wechat_actions.py +++ /dev/null @@ -1,54 +0,0 @@ -"""微信动作目录:服务器、手机端 Agent、验证脚本共同使用。""" -from __future__ import annotations -from dataclasses import dataclass, field -from typing import Any, Dict, List - -@dataclass(frozen=True) -class WechatAction: - action: str - rpc: str - module: str - description: str - required: List[str] = field(default_factory=list) - safe: bool = True - verify: bool = True - -ACTIONS: List[WechatAction] = [ - WechatAction('ping', 'ping', 'system', 'Hook 连通性检测'), - WechatAction('get_connection_status', 'getConnectionStatus', 'system', '读取连接状态'), - WechatAction('screenshot', 'takeScreenshot', 'device', '手机当前屏幕截图'), - WechatAction('ui_dump', 'dumpUiTree', 'device', '导出当前 UI 树'), - WechatAction('go_home', 'goHome', 'navigation', '返回微信首页'), - WechatAction('open_contacts', 'openContacts', 'navigation', '打开通讯录'), - WechatAction('open_chats', 'openChats', 'navigation', '打开聊天列表'), - WechatAction('search_contact', 'searchContact', 'contact', '搜索联系人', ['keyword']), - WechatAction('get_contacts', 'getContacts', 'contact', '读取联系人列表'), - WechatAction('get_contact_profile', 'getContactProfile', 'contact', '读取联系人资料', ['wxid']), - WechatAction('add_friend', 'addFriend', 'contact', '添加好友', ['keyword'], safe=False), - WechatAction('accept_friend', 'acceptFriend', 'contact', '通过好友申请', ['wxid'], safe=False), - WechatAction('send_text', 'sendTextMessage', 'message', '发送文本消息', ['to', 'text'], safe=False), - WechatAction('send_image', 'sendImageMessage', 'message', '发送图片消息', ['to', 'path'], safe=False), - WechatAction('get_messages', 'getMessages', 'message', '读取消息列表', ['wxid']), - WechatAction('open_chat', 'openChat', 'message', '打开指定会话', ['wxid']), - WechatAction('create_group', 'createGroup', 'group', '创建群聊', ['members'], safe=False), - WechatAction('invite_group_member', 'inviteGroupMember', 'group', '邀请群成员', ['chatroom', 'members'], safe=False), - WechatAction('get_group_members', 'getGroupMembers', 'group', '读取群成员', ['chatroom']), - WechatAction('browse_channels', 'browseChannels', 'channels', '浏览视频号'), - WechatAction('add_favorite', 'addFavorite', 'favorite', '收藏当前内容'), - WechatAction('add_custom_emoji', 'addCustomEmoji', 'emoji', '添加自定义表情', safe=False), -] - -ACTION_MAP: Dict[str, WechatAction] = {item.action: item for item in ACTIONS} -RPC_MAP: Dict[str, WechatAction] = {item.rpc: item for item in ACTIONS} - -def list_actions() -> List[Dict[str, Any]]: - return [item.__dict__.copy() for item in ACTIONS] - -def validate_action(action: str, payload: Dict[str, Any]) -> WechatAction: - if action not in ACTION_MAP: - raise KeyError(f'未知微信动作:{action}') - spec = ACTION_MAP[action] - missing = [k for k in spec.required if k not in payload or payload[k] in (None, '')] - if missing: - raise ValueError(f'动作 {action} 缺少参数:{missing}') - return spec diff --git a/工作手机SDK补全包_微信Frida/server/__init__.py b/工作手机SDK补全包_微信Frida/server/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/工作手机SDK补全包_微信Frida/server/app.py b/工作手机SDK补全包_微信Frida/server/app.py deleted file mode 100644 index 4c7de97ef9..0000000000 --- a/工作手机SDK补全包_微信Frida/server/app.py +++ /dev/null @@ -1,13 +0,0 @@ -from fastapi import FastAPI, WebSocket -from server.routes.wechat_frida import router as wechat_frida_router, phone_socket - -app = FastAPI(title='工作手机 SDK · 微信 Frida 控制服务', version='0.3.0') -app.include_router(wechat_frida_router) - -@app.get('/health') -def health(): - return {'ok': True, 'service': 'work-phone-sdk'} - -@app.websocket('/ws/phone') -async def ws_phone(websocket: WebSocket): - await phone_socket(websocket) diff --git a/工作手机SDK补全包_微信Frida/server/routes/__init__.py b/工作手机SDK补全包_微信Frida/server/routes/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/工作手机SDK补全包_微信Frida/server/routes/wechat_frida.py b/工作手机SDK补全包_微信Frida/server/routes/wechat_frida.py deleted file mode 100644 index 2f7b7d14f5..0000000000 --- a/工作手机SDK补全包_微信Frida/server/routes/wechat_frida.py +++ /dev/null @@ -1,76 +0,0 @@ -"""FastAPI 微信 Frida 无线控制路由。""" -from __future__ import annotations -import asyncio -import json -import uuid -from dataclasses import dataclass, field -from typing import Any, Dict, Optional - -from fastapi import APIRouter, WebSocket, WebSocketDisconnect, HTTPException -from pydantic import BaseModel - -from sdk.wechat.wechat_actions import list_actions - -router = APIRouter(prefix='/api/v3/wechat-frida', tags=['wechat-frida']) - -@dataclass -class PhoneConn: - device_id: str - ws: WebSocket - meta: Dict[str, Any] = field(default_factory=dict) - pending: Dict[str, asyncio.Future] = field(default_factory=dict) - -phones: Dict[str, PhoneConn] = {} - -class ActionRequest(BaseModel): - device_id: str - action: str - payload: Dict[str, Any] = {} - timeout: float = 30.0 - -@router.get('/actions') -def actions(): - return {'ok': True, 'actions': list_actions()} - -@router.get('/devices') -def devices(): - return {'ok': True, 'devices': [{'device_id': k, 'meta': v.meta} for k, v in phones.items()]} - -@router.post('/execute') -async def execute(req: ActionRequest): - if req.device_id not in phones: - raise HTTPException(404, f'设备未连接:{req.device_id}') - conn = phones[req.device_id] - request_id = uuid.uuid4().hex - fut = asyncio.get_event_loop().create_future() - conn.pending[request_id] = fut - await conn.ws.send_text(json.dumps({'type': 'command', 'request_id': request_id, 'action': req.action, 'payload': req.payload}, ensure_ascii=False)) - try: - return await asyncio.wait_for(fut, timeout=req.timeout) - finally: - conn.pending.pop(request_id, None) - -async def phone_socket(websocket: WebSocket): - await websocket.accept() - device_id: Optional[str] = None - try: - async for raw in websocket.iter_text(): - msg = json.loads(raw) - if msg.get('type') == 'hello': - device_id = msg.get('device_id') or uuid.uuid4().hex - phones[device_id] = PhoneConn(device_id=device_id, ws=websocket, meta=msg) - await websocket.send_text(json.dumps({'type': 'hello_ack', 'device_id': device_id}, ensure_ascii=False)) - elif msg.get('type') == 'result': - rid = msg.get('request_id') - did = msg.get('device_id') or device_id - conn = phones.get(did or '') - if conn and rid in conn.pending and not conn.pending[rid].done(): - conn.pending[rid].set_result({'ok': True, 'device_id': did, 'result': msg.get('result')}) - elif msg.get('type') == 'error': - # 保留连接,错误由客户端下一次请求再显式返回。 - pass - except WebSocketDisconnect: - pass - finally: - if device_id and device_id in phones: - phones.pop(device_id, None) diff --git a/工作手机SDK补全包_微信Frida/tests_api_routes.py b/工作手机SDK补全包_微信Frida/tests_api_routes.py deleted file mode 100644 index c0f589bd84..0000000000 --- a/工作手机SDK补全包_微信Frida/tests_api_routes.py +++ /dev/null @@ -1,23 +0,0 @@ -from fastapi.testclient import TestClient -from server.app import app - -client = TestClient(app) - -r = client.get('/health') -assert r.status_code == 200, r.text -assert r.json()['ok'] is True - -r = client.get('/api/v3/wechat-frida/actions') -assert r.status_code == 200, r.text -body = r.json() -assert body['ok'] is True -assert len(body['actions']) >= 20 -assert any(x['action'] == 'send_text' and x['rpc'] == 'sendTextMessage' for x in body['actions']) - -r = client.get('/api/v3/wechat-frida/devices') -assert r.status_code == 200, r.text -assert r.json()['devices'] == [] - -r = client.post('/api/v3/wechat-frida/execute', json={'device_id':'missing','action':'ping','payload':{}}) -assert r.status_code == 404, r.text -print('api_route_tests_passed') diff --git a/工作手机SDK补全包_微信Frida/tests_mock_rpc.py b/工作手机SDK补全包_微信Frida/tests_mock_rpc.py deleted file mode 100644 index c69d6c662b..0000000000 --- a/工作手机SDK补全包_微信Frida/tests_mock_rpc.py +++ /dev/null @@ -1,21 +0,0 @@ -from sdk.frida.frida_manager import FridaManager -from sdk.frida.hook_executor import HookExecutor - -class Exports: - def ping(self): - return 'pong' - def getConnectionStatus(self): - return {'ok': True} - def sendTextMessage(self, to, text): - return {'to': to, 'text': text} - -m = FridaManager() -m.exports = Exports() -assert m.call('ping').ok -assert m.call('getConnectionStatus').ok -assert m.call('sendTextMessage', 'filehelper', 'hi').ok -ex = HookExecutor(m) -assert ex.execute('ping')['ok'] is True -assert ex.execute('send_text', {'to':'filehelper','text':'hi'})['ok'] is True -assert ex.execute('send_text', {'to':'filehelper'})['ok'] is False -print('mock_rpc_tests_passed') diff --git a/工作手机SDK补全包_微信Frida/开发文档/10、项目管理/项目落地执行表.md b/工作手机SDK补全包_微信Frida/开发文档/10、项目管理/项目落地执行表.md deleted file mode 100644 index 5ecf47c0fb..0000000000 --- a/工作手机SDK补全包_微信Frida/开发文档/10、项目管理/项目落地执行表.md +++ /dev/null @@ -1,6 +0,0 @@ -# 项目落地执行表(补全包归档) - -> **已并入**:[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §1.2 -> **进度真源**:[开发文档/10、项目管理/开发进度总表.md](../../../开发文档/10、项目管理/开发进度总表.md)(99.5%) - -2026-05-18 补全包阶段约 55% 的记录已归档;主线进度以总表为准。 diff --git a/工作手机SDK补全包_微信Frida/开发文档/2、架构/系统架构.md b/工作手机SDK补全包_微信Frida/开发文档/2、架构/系统架构.md deleted file mode 100644 index 1ae0bb614c..0000000000 --- a/工作手机SDK补全包_微信Frida/开发文档/2、架构/系统架构.md +++ /dev/null @@ -1,6 +0,0 @@ -# 系统架构(补全包归档) - -> **已并入**:[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §一、§二 -> 完整架构见 [开发文档/2、架构/01-总览/系统架构.md](../../../开发文档/2、架构/01-总览/系统架构.md) - -本文件为补全包内副本,勿再扩写。 diff --git a/工作手机SDK补全包_微信Frida/开发文档/5、接口/微信Frida_API契约.md b/工作手机SDK补全包_微信Frida/开发文档/5、接口/微信Frida_API契约.md deleted file mode 100644 index 7a37326857..0000000000 --- a/工作手机SDK补全包_微信Frida/开发文档/5、接口/微信Frida_API契约.md +++ /dev/null @@ -1,71 +0,0 @@ -# 微信 Frida API 契约 - -本文由阿桥维护,描述服务器与手机端 Agent、外部控制端之间的接口约定。 - -## 一、HTTP 接口 - -| 方法 | 路径 | 说明 | 请求 | 响应 | -|---|---|---|---|---| -| GET | `/health` | 服务健康检查 | 无 | `{ ok, service }` | -| GET | `/api/v3/wechat-frida/actions` | 获取可执行微信动作目录 | 无 | `{ ok, actions }` | -| GET | `/api/v3/wechat-frida/devices` | 获取已连接手机 Agent | 无 | `{ ok, devices }` | -| POST | `/api/v3/wechat-frida/execute` | 下发微信控制动作 | `{ device_id, action, payload, timeout }` | `{ ok, device_id, result }` | - -## 二、WebSocket 接口 - -手机端连接: - -```text -ws://服务器IP:8000/ws/phone -``` - -### 2.1 Agent 注册 - -```json -{ - "type": "hello", - "device_id": "phone-xxxx", - "platform": "Android/Termux", - "actions": [] -} -``` - -### 2.2 服务器下发命令 - -```json -{ - "type": "command", - "request_id": "uuid", - "action": "send_text", - "payload": { - "to": "filehelper", - "text": "工作手机SDK自动验证" - } -} -``` - -### 2.3 Agent 返回结果 - -```json -{ - "type": "result", - "request_id": "uuid", - "device_id": "phone-xxxx", - "result": { - "ok": true, - "action": "send_text", - "rpc": "sendTextMessage", - "method_used": "sendTextMessage", - "data": {}, - "elapsed_ms": 123 - } -} -``` - -## 三、动作目录 - -动作目录集中维护在 `/home/ubuntu/work_phone_sdk_completion/sdk/wechat/wechat_actions.py`。当前首批覆盖系统检测、截图、UI、通讯录、好友、消息、群聊、视频号、收藏、表情等模块。危险动作默认标记为 `safe=false`,自动验证脚本默认跳过,避免误发消息或误加好友。 - -## 四、RPC 方法名兼容 - -SDK 调用顺序为:原始 `camelCase` → `snake_case` → 全小写 → `invoke()`。这用于解决归档中反复出现的 `getMessages/getmessages/get_messages` 调用差异问题。 diff --git a/工作手机SDK补全包_微信Frida/开发文档/5、接口/微信Frida控制接口契约.md b/工作手机SDK补全包_微信Frida/开发文档/5、接口/微信Frida控制接口契约.md deleted file mode 100644 index 96ce690bae..0000000000 --- a/工作手机SDK补全包_微信Frida/开发文档/5、接口/微信Frida控制接口契约.md +++ /dev/null @@ -1,5 +0,0 @@ -# 微信 Frida 控制接口契约(补全包归档) - -> **已并入**:[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §三、§七 - -统一 API:`POST /api/v3/hook/execute` · 代码:`sdk/app/routers/unified.py` diff --git a/工作手机SDK补全包_微信Frida/开发文档/8、部署/微信Frida无线部署与验证说明.md b/工作手机SDK补全包_微信Frida/开发文档/8、部署/微信Frida无线部署与验证说明.md deleted file mode 100644 index ac4600e866..0000000000 --- a/工作手机SDK补全包_微信Frida/开发文档/8、部署/微信Frida无线部署与验证说明.md +++ /dev/null @@ -1,5 +0,0 @@ -# 微信 Frida 无线部署与验证说明(补全包归档) - -> **已并入**:[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §五、§八 - -验证脚本主线版:`tools/wireless_frida_wechat_batch_verify.py` diff --git a/补全包/README.md b/补全包/README.md deleted file mode 100644 index f75933de78..0000000000 --- a/补全包/README.md +++ /dev/null @@ -1,12 +0,0 @@ -# 补全包目录 - -> **文档已并入开发文档** — 请勿在此维护正文。 -> **唯一主手册**:[开发文档/9、手册/工作手机·五图总览与使用手册.md](../开发文档/9、手册/工作手机·五图总览与使用手册.md)(v2.0 补全包并入版) - -## 子目录 - -| 目录 | 说明 | -|------|------| -| [微信Frida_SDK_20260518/](微信Frida_SDK_20260518/) | 2026-05-18 归档原型代码(只读对照) | - -日常开发请使用主线 **`sdk/`**,部署与 API 见主手册 §六~§八。 diff --git a/补全包/微信Frida_SDK_20260518/README.md b/补全包/微信Frida_SDK_20260518/README.md deleted file mode 100644 index 360644f6fc..0000000000 --- a/补全包/微信Frida_SDK_20260518/README.md +++ /dev/null @@ -1,29 +0,0 @@ -# 工作手机 SDK · 微信 Frida 无线控制补全包 - -> **⚠️ 归档原型 · 文档已并入主线** -> **请读**:[开发文档/9、手册/工作手机·五图总览与使用手册.md](../开发文档/9、手册/工作手机·五图总览与使用手册.md)(v2.0) -> **请用**:主线 `sdk/`(非本目录 `server/`、`mobile_agent/`) - -本目录保留 2026-05-18 中断 ZIP 的最小闭环源码,供 diff 对照。功能已吸收进 `sdk/agent/hook/`、`sdk/app/routers/unified.py` 等。 - -## 快速启动(历史命令,仅供对照) - -
-展开旧版启动命令 - -服务器端: - -```bash -pip install -r requirements.txt -uvicorn server.app:app --host 0.0.0.0 --port 8000 -``` - -手机 Termux 端: - -```bash -python mobile_agent/wireless_agent.py --server-ws ws://服务器IP:8000/ws/phone --hook hooks/wechat_hook_bridge.js -``` - -
- -**现行启动**见主手册 §六、§八(端口 **8899**,API **`/api/v3/hook/execute`**)。 diff --git a/补全包/微信Frida_SDK_20260518/generate_impl.py b/补全包/微信Frida_SDK_20260518/generate_impl.py deleted file mode 100644 index 63f3879cfb..0000000000 --- a/补全包/微信Frida_SDK_20260518/generate_impl.py +++ /dev/null @@ -1,578 +0,0 @@ -from pathlib import Path - -ROOT = Path('/home/ubuntu/work_phone_sdk_completion') -files = {} - -files['sdk/frida/frida_manager.py'] = r'''""" -工作手机 SDK · FridaManager - -负责手机端或服务器端连接 Frida、attach 微信进程、加载 Hook 脚本,并提供兼容 RPC 调用。 -设计重点:历史归档显示 Frida Python 对 rpc.exports 方法名存在混淆,因此这里采用多候选名兼容策略。 -""" -from __future__ import annotations - -from dataclasses import dataclass, field -from pathlib import Path -import re -import time -from typing import Any, Callable, Dict, Iterable, Optional - - -class FridaUnavailable(RuntimeError): - """当前环境未安装或无法连接 Frida。""" - - -class RpcMethodMissing(RuntimeError): - """Hook 中缺少指定 RPC 方法。""" - - -@dataclass -class FridaConfig: - device_host: str = "127.0.0.1" - device_port: int = 27042 - package_name: str = "com.tencent.mm" - process_name: str = "WeChat" - attach_timeout: float = 15.0 - prefer_usb: bool = False - hook_path: Optional[str] = None - - -@dataclass -class RpcResult: - ok: bool - action: str - method: str - data: Any = None - error: Optional[str] = None - elapsed_ms: int = 0 - - -class FridaManager: - def __init__(self, config: Optional[FridaConfig] = None): - self.config = config or FridaConfig() - self.frida = None - self.device = None - self.session = None - self.script = None - self.exports = None - self.loaded_hook_path: Optional[str] = None - - def _import_frida(self): - if self.frida is not None: - return self.frida - try: - import frida # type: ignore - except Exception as exc: - raise FridaUnavailable(f"未安装 frida Python 包或加载失败:{exc}") from exc - self.frida = frida - return frida - - def connect(self): - frida = self._import_frida() - if self.config.prefer_usb: - self.device = frida.get_usb_device(timeout=int(self.config.attach_timeout)) - else: - self.device = frida.get_device_manager().add_remote_device( - f"{self.config.device_host}:{self.config.device_port}" - ) - return self.device - - def attach(self, target: Optional[str] = None): - if self.device is None: - self.connect() - assert self.device is not None - target = target or self.config.package_name - last_error: Optional[Exception] = None - candidates: Iterable[Any] = [target, self.config.process_name] - for candidate in candidates: - try: - self.session = self.device.attach(candidate) - return self.session - except Exception as exc: # pragma: no cover - depends on phone - last_error = exc - try: - processes = self.device.enumerate_processes() - for proc in processes: - name = getattr(proc, 'name', '') or '' - pid = getattr(proc, 'pid', None) - if pid and (self.config.package_name in name or self.config.process_name.lower() in name.lower() or '微信' in name): - self.session = self.device.attach(pid) - return self.session - except Exception as exc: # pragma: no cover - last_error = exc - raise FridaUnavailable(f"无法 attach 微信进程:{last_error}") - - def load_script(self, hook_path: Optional[str] = None): - path = Path(hook_path or self.config.hook_path or '') - if not path.exists(): - raise FileNotFoundError(f"Hook 脚本不存在:{path}") - if self.session is None: - self.attach() - assert self.session is not None - source = path.read_text(encoding='utf-8') - self.script = self.session.create_script(source) - self.script.on('message', self._on_message) - self.script.load() - self.exports = getattr(self.script, 'exports_sync', None) or getattr(self.script, 'exports', None) - self.loaded_hook_path = str(path) - return self.script - - def _on_message(self, message, data): # pragma: no cover - runtime callback - # 生产环境可转发到 EventReporter;这里保持最小日志。 - print({'frida_message': message, 'data_len': len(data) if data else 0}) - - @staticmethod - def _snake(name: str) -> str: - return re.sub(r'(? RpcResult: - started = time.time() - if self.exports is None: - raise FridaUnavailable('Hook 脚本尚未加载,无法调用 RPC') - last_error = None - for candidate in self.method_candidates(method): - try: - fn = getattr(self.exports, candidate) - data = fn(*args, **kwargs) - return RpcResult(True, method, candidate, data=data, elapsed_ms=int((time.time() - started) * 1000)) - except AttributeError as exc: - last_error = exc - except Exception as exc: - return RpcResult(False, method, candidate, error=str(exc), elapsed_ms=int((time.time() - started) * 1000)) - # 兼容新版 frida 的 exports_sync.invoke 或 script.exports_sync.invoke - try: - invoke = getattr(self.exports, 'invoke') - data = invoke(method, list(args), kwargs or None) - return RpcResult(True, method, 'invoke', data=data, elapsed_ms=int((time.time() - started) * 1000)) - except Exception as exc: - last_error = exc - return RpcResult(False, method, '', error=f'RPC 方法不存在或不可调用:{method}; last={last_error}', elapsed_ms=int((time.time() - started) * 1000)) - - def cleanup(self): - for obj, method in [(self.script, 'unload'), (self.session, 'detach')]: - if obj is not None: - try: - getattr(obj, method)() - except Exception: - pass - self.script = None - self.session = None - self.exports = None -''' - -files['sdk/wechat/wechat_actions.py'] = r'''"""微信动作目录:服务器、手机端 Agent、验证脚本共同使用。""" -from __future__ import annotations -from dataclasses import dataclass, field -from typing import Any, Dict, List - -@dataclass(frozen=True) -class WechatAction: - action: str - rpc: str - module: str - description: str - required: List[str] = field(default_factory=list) - safe: bool = True - verify: bool = True - -ACTIONS: List[WechatAction] = [ - WechatAction('ping', 'ping', 'system', 'Hook 连通性检测'), - WechatAction('get_connection_status', 'getConnectionStatus', 'system', '读取连接状态'), - WechatAction('screenshot', 'takeScreenshot', 'device', '手机当前屏幕截图'), - WechatAction('ui_dump', 'dumpUiTree', 'device', '导出当前 UI 树'), - WechatAction('go_home', 'goHome', 'navigation', '返回微信首页'), - WechatAction('open_contacts', 'openContacts', 'navigation', '打开通讯录'), - WechatAction('open_chats', 'openChats', 'navigation', '打开聊天列表'), - WechatAction('search_contact', 'searchContact', 'contact', '搜索联系人', ['keyword']), - WechatAction('get_contacts', 'getContacts', 'contact', '读取联系人列表'), - WechatAction('get_contact_profile', 'getContactProfile', 'contact', '读取联系人资料', ['wxid']), - WechatAction('add_friend', 'addFriend', 'contact', '添加好友', ['keyword'], safe=False), - WechatAction('accept_friend', 'acceptFriend', 'contact', '通过好友申请', ['wxid'], safe=False), - WechatAction('send_text', 'sendTextMessage', 'message', '发送文本消息', ['to', 'text'], safe=False), - WechatAction('send_image', 'sendImageMessage', 'message', '发送图片消息', ['to', 'path'], safe=False), - WechatAction('get_messages', 'getMessages', 'message', '读取消息列表', ['wxid']), - WechatAction('open_chat', 'openChat', 'message', '打开指定会话', ['wxid']), - WechatAction('create_group', 'createGroup', 'group', '创建群聊', ['members'], safe=False), - WechatAction('invite_group_member', 'inviteGroupMember', 'group', '邀请群成员', ['chatroom', 'members'], safe=False), - WechatAction('get_group_members', 'getGroupMembers', 'group', '读取群成员', ['chatroom']), - WechatAction('browse_channels', 'browseChannels', 'channels', '浏览视频号'), - WechatAction('add_favorite', 'addFavorite', 'favorite', '收藏当前内容'), - WechatAction('add_custom_emoji', 'addCustomEmoji', 'emoji', '添加自定义表情', safe=False), -] - -ACTION_MAP: Dict[str, WechatAction] = {item.action: item for item in ACTIONS} -RPC_MAP: Dict[str, WechatAction] = {item.rpc: item for item in ACTIONS} - -def list_actions() -> List[Dict[str, Any]]: - return [item.__dict__.copy() for item in ACTIONS] - -def validate_action(action: str, payload: Dict[str, Any]) -> WechatAction: - if action not in ACTION_MAP: - raise KeyError(f'未知微信动作:{action}') - spec = ACTION_MAP[action] - missing = [k for k in spec.required if k not in payload or payload[k] in (None, '')] - if missing: - raise ValueError(f'动作 {action} 缺少参数:{missing}') - return spec -''' - -files['sdk/frida/hook_executor.py'] = r'''"""将业务 action 映射到 Frida RPC。""" -from __future__ import annotations -from dataclasses import asdict -from typing import Any, Dict, Optional - -from sdk.frida.frida_manager import FridaManager, RpcResult -from sdk.wechat.wechat_actions import validate_action, list_actions - -class HookExecutor: - def __init__(self, manager: FridaManager): - self.manager = manager - - def execute(self, action: str, payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: - payload = payload or {} - try: - spec = validate_action(action, payload) - except Exception as exc: - return {'ok': False, 'action': action, 'error': str(exc), 'stage': 'validate'} - args = [payload[k] for k in spec.required] - # 允许额外参数透传给 Hook;Hook 若不支持会返回错误,便于报告定位。 - extra = {k: v for k, v in payload.items() if k not in spec.required} - result: RpcResult = self.manager.call(spec.rpc, *args, **extra) - return { - 'ok': result.ok, - 'action': action, - 'rpc': spec.rpc, - 'method_used': result.method, - 'module': spec.module, - 'description': spec.description, - 'safe': spec.safe, - 'data': result.data, - 'error': result.error, - 'elapsed_ms': result.elapsed_ms, - } - - def catalog(self): - return list_actions() -''' - -files['mobile_agent/wireless_agent.py'] = r'''"""手机端 Termux/Python Agent:通过 WebSocket 接收服务器动作,调用 Frida Hook。""" -from __future__ import annotations -import argparse -import asyncio -import json -import platform -import uuid -from pathlib import Path - -from sdk.frida.frida_manager import FridaConfig, FridaManager -from sdk.frida.hook_executor import HookExecutor - -try: - import websockets -except Exception: # pragma: no cover - websockets = None - -class WirelessAgent: - def __init__(self, server_ws: str, hook_path: str, device_id: str | None = None): - self.server_ws = server_ws - self.hook_path = hook_path - self.device_id = device_id or f"phone-{uuid.getnode():x}" - self.manager = FridaManager(FridaConfig(hook_path=hook_path)) - self.executor = HookExecutor(self.manager) - - async def boot(self): - self.manager.connect() - self.manager.attach() - self.manager.load_script(self.hook_path) - - def hello(self): - return { - 'type': 'hello', - 'device_id': self.device_id, - 'platform': platform.platform(), - 'actions': self.executor.catalog(), - } - - async def run(self): - if websockets is None: - raise RuntimeError('请先安装 websockets:pip install websockets') - await self.boot() - async with websockets.connect(self.server_ws, ping_interval=20, ping_timeout=20) as ws: - await ws.send(json.dumps(self.hello(), ensure_ascii=False)) - async for raw in ws: - try: - msg = json.loads(raw) - if msg.get('type') != 'command': - continue - result = self.executor.execute(msg['action'], msg.get('payload') or {}) - await ws.send(json.dumps({'type': 'result', 'request_id': msg.get('request_id'), 'device_id': self.device_id, 'result': result}, ensure_ascii=False)) - except Exception as exc: - await ws.send(json.dumps({'type': 'error', 'error': str(exc)}, ensure_ascii=False)) - - -def main(): - parser = argparse.ArgumentParser() - parser.add_argument('--server-ws', required=True, help='例如 ws://192.168.1.10:8000/ws/phone') - parser.add_argument('--hook', required=True, help='wechat_hook_bridge.js 或真实 wechat_hook_v3.js 路径') - parser.add_argument('--device-id') - args = parser.parse_args() - asyncio.run(WirelessAgent(args.server_ws, args.hook, args.device_id).run()) - -if __name__ == '__main__': - main() -''' - -files['server/routes/wechat_frida.py'] = r'''"""FastAPI 微信 Frida 无线控制路由。""" -from __future__ import annotations -import asyncio -import json -import uuid -from dataclasses import dataclass, field -from typing import Any, Dict, Optional - -from fastapi import APIRouter, WebSocket, WebSocketDisconnect, HTTPException -from pydantic import BaseModel - -from sdk.wechat.wechat_actions import list_actions - -router = APIRouter(prefix='/api/v3/wechat-frida', tags=['wechat-frida']) - -@dataclass -class PhoneConn: - device_id: str - ws: WebSocket - meta: Dict[str, Any] = field(default_factory=dict) - pending: Dict[str, asyncio.Future] = field(default_factory=dict) - -phones: Dict[str, PhoneConn] = {} - -class ActionRequest(BaseModel): - device_id: str - action: str - payload: Dict[str, Any] = {} - timeout: float = 30.0 - -@router.get('/actions') -def actions(): - return {'ok': True, 'actions': list_actions()} - -@router.get('/devices') -def devices(): - return {'ok': True, 'devices': [{'device_id': k, 'meta': v.meta} for k, v in phones.items()]} - -@router.post('/execute') -async def execute(req: ActionRequest): - if req.device_id not in phones: - raise HTTPException(404, f'设备未连接:{req.device_id}') - conn = phones[req.device_id] - request_id = uuid.uuid4().hex - fut = asyncio.get_event_loop().create_future() - conn.pending[request_id] = fut - await conn.ws.send_text(json.dumps({'type': 'command', 'request_id': request_id, 'action': req.action, 'payload': req.payload}, ensure_ascii=False)) - try: - return await asyncio.wait_for(fut, timeout=req.timeout) - finally: - conn.pending.pop(request_id, None) - -async def phone_socket(websocket: WebSocket): - await websocket.accept() - device_id: Optional[str] = None - try: - async for raw in websocket.iter_text(): - msg = json.loads(raw) - if msg.get('type') == 'hello': - device_id = msg.get('device_id') or uuid.uuid4().hex - phones[device_id] = PhoneConn(device_id=device_id, ws=websocket, meta=msg) - await websocket.send_text(json.dumps({'type': 'hello_ack', 'device_id': device_id}, ensure_ascii=False)) - elif msg.get('type') == 'result': - rid = msg.get('request_id') - did = msg.get('device_id') or device_id - conn = phones.get(did or '') - if conn and rid in conn.pending and not conn.pending[rid].done(): - conn.pending[rid].set_result({'ok': True, 'device_id': did, 'result': msg.get('result')}) - elif msg.get('type') == 'error': - # 保留连接,错误由客户端下一次请求再显式返回。 - pass - except WebSocketDisconnect: - pass - finally: - if device_id and device_id in phones: - phones.pop(device_id, None) -''' - -files['server/app.py'] = r'''from fastapi import FastAPI, WebSocket -from server.routes.wechat_frida import router as wechat_frida_router, phone_socket - -app = FastAPI(title='工作手机 SDK · 微信 Frida 控制服务', version='0.3.0') -app.include_router(wechat_frida_router) - -@app.get('/health') -def health(): - return {'ok': True, 'service': 'work-phone-sdk'} - -@app.websocket('/ws/phone') -async def ws_phone(websocket: WebSocket): - await phone_socket(websocket) -''' - -files['hooks/wechat_hook_bridge.js'] = r'''// 工作手机 SDK · 微信 Hook 桥接模板 -// 说明:这是安全桥接版,优先提供连通性、截图、UI 导航等通用方法。 -// 如果已有完整 wechat_hook_v3.js,可直接替换本文件;Python SDK 会兼容 camelCase/snake_case 调用。 -'use strict'; - -function ok(data) { return { ok: true, data: data || null, ts: Date.now() }; } -function fail(message) { return { ok: false, error: String(message), ts: Date.now() }; } - -function runJava(fn) { - let result; - Java.perform(function () { - try { result = fn(); } catch (e) { result = fail(e.stack || e.message || e); } - }); - return result; -} - -rpc.exports = { - ping: function () { return 'pong from wechat_hook_bridge'; }, - getConnectionStatus: function () { - return ok({ java_available: Java.available, process: Process.id, arch: Process.arch, platform: Process.platform }); - }, - takeScreenshot: function () { - // 截图建议在手机端通过 uiautomator/screencap 实现;Hook 层返回占位,Agent 可扩展真实文件路径。 - return ok({ mode: 'placeholder', message: '请在 mobile_agent 扩展 screencap -p 后回传文件路径' }); - }, - dumpUiTree: function () { return ok({ mode: 'placeholder', message: '建议通过 uiautomator dump 获取 UI XML' }); }, - goHome: function () { return ok({ action: 'goHome', message: '桥接模板未执行 UI 点击;请替换完整 Hook 后验证' }); }, - openContacts: function () { return ok({ action: 'openContacts', message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - openChats: function () { return ok({ action: 'openChats', message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - searchContact: function (keyword) { return ok({ keyword: keyword, message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - getContacts: function () { return ok({ contacts: [], message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - getContactProfile: function (wxid) { return ok({ wxid: wxid, message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - addFriend: function (keyword) { return fail('危险动作未在模板中实现:addFriend ' + keyword); }, - acceptFriend: function (wxid) { return fail('危险动作未在模板中实现:acceptFriend ' + wxid); }, - sendTextMessage: function (to, text) { return fail('危险动作未在模板中实现:sendTextMessage ' + to); }, - sendImageMessage: function (to, path) { return fail('危险动作未在模板中实现:sendImageMessage ' + to); }, - getMessages: function (wxid) { return ok({ wxid: wxid, messages: [], message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - openChat: function (wxid) { return ok({ wxid: wxid, message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - createGroup: function (members) { return fail('危险动作未在模板中实现:createGroup'); }, - inviteGroupMember: function (chatroom, members) { return fail('危险动作未在模板中实现:inviteGroupMember'); }, - getGroupMembers: function (chatroom) { return ok({ chatroom: chatroom, members: [], message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - browseChannels: function () { return ok({ message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - addFavorite: function () { return fail('危险动作未在模板中实现:addFavorite'); }, - addCustomEmoji: function () { return fail('危险动作未在模板中实现:addCustomEmoji'); } -}; -''' - -files['scripts/verify_wechat_frida.py'] = r'''"""一键验证服务器控制手机微信功能,输出 JSON 与 Markdown 报告。""" -from __future__ import annotations -import argparse -import json -import time -from pathlib import Path -from typing import Any, Dict - -import requests - -from sdk.wechat.wechat_actions import ACTIONS - -DEFAULT_PAYLOADS: Dict[str, Dict[str, Any]] = { - 'search_contact': {'keyword': '文件传输助手'}, - 'get_contact_profile': {'wxid': 'filehelper'}, - 'add_friend': {'keyword': '__dry_run__'}, - 'accept_friend': {'wxid': '__dry_run__'}, - 'send_text': {'to': 'filehelper', 'text': '工作手机SDK自动验证'}, - 'send_image': {'to': 'filehelper', 'path': '/sdcard/Pictures/test.png'}, - 'get_messages': {'wxid': 'filehelper'}, - 'open_chat': {'wxid': 'filehelper'}, - 'create_group': {'members': ['filehelper']}, - 'invite_group_member': {'chatroom': '__dry_run__', 'members': ['filehelper']}, - 'get_group_members': {'chatroom': '__dry_run__'}, -} - -def call(base_url: str, device_id: str, action: str, payload: Dict[str, Any], timeout: float): - r = requests.post(f'{base_url}/api/v3/wechat-frida/execute', json={'device_id': device_id, 'action': action, 'payload': payload, 'timeout': timeout}, timeout=timeout + 5) - try: - return r.status_code, r.json() - except Exception: - return r.status_code, {'ok': False, 'text': r.text} - -def main(): - parser = argparse.ArgumentParser() - parser.add_argument('--base-url', default='http://127.0.0.1:8000') - parser.add_argument('--device-id', required=True) - parser.add_argument('--out-dir', default='reports') - parser.add_argument('--include-dangerous', action='store_true', help='默认不执行加好友/发消息等危险动作,只记录 SKIPPED') - parser.add_argument('--timeout', type=float, default=30) - args = parser.parse_args() - - out_dir = Path(args.out_dir) - out_dir.mkdir(parents=True, exist_ok=True) - rows = [] - for spec in ACTIONS: - if not spec.safe and not args.include_dangerous: - rows.append({'action': spec.action, 'rpc': spec.rpc, 'ok': None, 'status': 'SKIPPED_DANGEROUS', 'description': spec.description}) - continue - status, data = call(args.base_url, args.device_id, spec.action, DEFAULT_PAYLOADS.get(spec.action, {}), args.timeout) - rows.append({'action': spec.action, 'rpc': spec.rpc, 'http_status': status, 'ok': bool(data.get('ok') and (data.get('result') or {}).get('ok', True)), 'response': data, 'description': spec.description}) - time.sleep(0.2) - - stamp = time.strftime('%Y%m%d_%H%M%S') - json_path = out_dir / f'wechat_frida_verify_{stamp}.json' - md_path = out_dir / f'wechat_frida_verify_{stamp}.md' - json_path.write_text(json.dumps(rows, ensure_ascii=False, indent=2), encoding='utf-8') - lines = ['# 微信 Frida 功能验证报告', '', f'- 时间:{time.strftime("%Y-%m-%d %H:%M:%S")}', f'- 设备:`{args.device_id}`', '', '| 动作 | RPC | 结果 | 说明 |', '|---|---|---|---|'] - for row in rows: - result = 'SKIP' if row.get('status') else ('PASS' if row.get('ok') else 'FAIL') - lines.append(f"| `{row['action']}` | `{row['rpc']}` | {result} | {row.get('description','')} |") - md_path.write_text('\n'.join(lines) + '\n', encoding='utf-8') - print(json_path) - print(md_path) - -if __name__ == '__main__': - main() -''' - -files['requirements.txt'] = 'fastapi\nuvicorn\nrequests\nwebsockets\nfrida\n\n' -files['README.md'] = r'''# 工作手机 SDK · 微信 Frida 无线控制补全包 - -本包用于承接上传 ZIP 中断的开发进度,补齐 **服务器 → WebSocket → 手机端 Agent → Frida → 微信 Hook** 的闭环。 - -## 快速启动 - -服务器端: - -```bash -pip install -r requirements.txt -uvicorn server.app:app --host 0.0.0.0 --port 8000 -``` - -手机 Termux 端: - -```bash -python mobile_agent/wireless_agent.py --server-ws ws://服务器IP:8000/ws/phone --hook hooks/wechat_hook_bridge.js -``` - -验证端: - -```bash -python scripts/verify_wechat_frida.py --base-url http://服务器IP:8000 --device-id phone-xxxx -``` - -如果已有完整 `wechat_hook_v3.js`,请替换 `hooks/wechat_hook_bridge.js`,SDK 会自动兼容 camelCase / snake_case / lowercase 方法名。 -''' - -for rel, content in files.items(): - path = ROOT / rel - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(content, encoding='utf-8') -print(f'generated {len(files)} files under {ROOT}') diff --git a/补全包/微信Frida_SDK_20260518/hooks/wechat_hook_bridge.js b/补全包/微信Frida_SDK_20260518/hooks/wechat_hook_bridge.js deleted file mode 100644 index d69423fa67..0000000000 --- a/补全包/微信Frida_SDK_20260518/hooks/wechat_hook_bridge.js +++ /dev/null @@ -1,45 +0,0 @@ -// 工作手机 SDK · 微信 Hook 桥接模板 -// 说明:这是安全桥接版,优先提供连通性、截图、UI 导航等通用方法。 -// 如果已有完整 wechat_hook_v3.js,可直接替换本文件;Python SDK 会兼容 camelCase/snake_case 调用。 -'use strict'; - -function ok(data) { return { ok: true, data: data || null, ts: Date.now() }; } -function fail(message) { return { ok: false, error: String(message), ts: Date.now() }; } - -function runJava(fn) { - let result; - Java.perform(function () { - try { result = fn(); } catch (e) { result = fail(e.stack || e.message || e); } - }); - return result; -} - -rpc.exports = { - ping: function () { return 'pong from wechat_hook_bridge'; }, - getConnectionStatus: function () { - return ok({ java_available: Java.available, process: Process.id, arch: Process.arch, platform: Process.platform }); - }, - takeScreenshot: function () { - // 截图建议在手机端通过 uiautomator/screencap 实现;Hook 层返回占位,Agent 可扩展真实文件路径。 - return ok({ mode: 'placeholder', message: '请在 mobile_agent 扩展 screencap -p 后回传文件路径' }); - }, - dumpUiTree: function () { return ok({ mode: 'placeholder', message: '建议通过 uiautomator dump 获取 UI XML' }); }, - goHome: function () { return ok({ action: 'goHome', message: '桥接模板未执行 UI 点击;请替换完整 Hook 后验证' }); }, - openContacts: function () { return ok({ action: 'openContacts', message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - openChats: function () { return ok({ action: 'openChats', message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - searchContact: function (keyword) { return ok({ keyword: keyword, message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - getContacts: function () { return ok({ contacts: [], message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - getContactProfile: function (wxid) { return ok({ wxid: wxid, message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - addFriend: function (keyword) { return fail('危险动作未在模板中实现:addFriend ' + keyword); }, - acceptFriend: function (wxid) { return fail('危险动作未在模板中实现:acceptFriend ' + wxid); }, - sendTextMessage: function (to, text) { return fail('危险动作未在模板中实现:sendTextMessage ' + to); }, - sendImageMessage: function (to, path) { return fail('危险动作未在模板中实现:sendImageMessage ' + to); }, - getMessages: function (wxid) { return ok({ wxid: wxid, messages: [], message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - openChat: function (wxid) { return ok({ wxid: wxid, message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - createGroup: function (members) { return fail('危险动作未在模板中实现:createGroup'); }, - inviteGroupMember: function (chatroom, members) { return fail('危险动作未在模板中实现:inviteGroupMember'); }, - getGroupMembers: function (chatroom) { return ok({ chatroom: chatroom, members: [], message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - browseChannels: function () { return ok({ message: 'NOT_IMPLEMENTED_IN_TEMPLATE' }); }, - addFavorite: function () { return fail('危险动作未在模板中实现:addFavorite'); }, - addCustomEmoji: function () { return fail('危险动作未在模板中实现:addCustomEmoji'); } -}; diff --git a/补全包/微信Frida_SDK_20260518/mobile_agent/__init__.py b/补全包/微信Frida_SDK_20260518/mobile_agent/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/补全包/微信Frida_SDK_20260518/mobile_agent/wireless_agent.py b/补全包/微信Frida_SDK_20260518/mobile_agent/wireless_agent.py deleted file mode 100644 index 594bd01f44..0000000000 --- a/补全包/微信Frida_SDK_20260518/mobile_agent/wireless_agent.py +++ /dev/null @@ -1,65 +0,0 @@ -"""手机端 Termux/Python Agent:通过 WebSocket 接收服务器动作,调用 Frida Hook。""" -from __future__ import annotations -import argparse -import asyncio -import json -import platform -import uuid -from pathlib import Path - -from sdk.frida.frida_manager import FridaConfig, FridaManager -from sdk.frida.hook_executor import HookExecutor - -try: - import websockets -except Exception: # pragma: no cover - websockets = None - -class WirelessAgent: - def __init__(self, server_ws: str, hook_path: str, device_id: str | None = None): - self.server_ws = server_ws - self.hook_path = hook_path - self.device_id = device_id or f"phone-{uuid.getnode():x}" - self.manager = FridaManager(FridaConfig(hook_path=hook_path)) - self.executor = HookExecutor(self.manager) - - async def boot(self): - self.manager.connect() - self.manager.attach() - self.manager.load_script(self.hook_path) - - def hello(self): - return { - 'type': 'hello', - 'device_id': self.device_id, - 'platform': platform.platform(), - 'actions': self.executor.catalog(), - } - - async def run(self): - if websockets is None: - raise RuntimeError('请先安装 websockets:pip install websockets') - await self.boot() - async with websockets.connect(self.server_ws, ping_interval=20, ping_timeout=20) as ws: - await ws.send(json.dumps(self.hello(), ensure_ascii=False)) - async for raw in ws: - try: - msg = json.loads(raw) - if msg.get('type') != 'command': - continue - result = self.executor.execute(msg['action'], msg.get('payload') or {}) - await ws.send(json.dumps({'type': 'result', 'request_id': msg.get('request_id'), 'device_id': self.device_id, 'result': result}, ensure_ascii=False)) - except Exception as exc: - await ws.send(json.dumps({'type': 'error', 'error': str(exc)}, ensure_ascii=False)) - - -def main(): - parser = argparse.ArgumentParser() - parser.add_argument('--server-ws', required=True, help='例如 ws://192.168.1.10:8000/ws/phone') - parser.add_argument('--hook', required=True, help='wechat_hook_bridge.js 或真实 wechat_hook_v3.js 路径') - parser.add_argument('--device-id') - args = parser.parse_args() - asyncio.run(WirelessAgent(args.server_ws, args.hook, args.device_id).run()) - -if __name__ == '__main__': - main() diff --git a/补全包/微信Frida_SDK_20260518/requirements.txt b/补全包/微信Frida_SDK_20260518/requirements.txt deleted file mode 100644 index b52d28ca18..0000000000 --- a/补全包/微信Frida_SDK_20260518/requirements.txt +++ /dev/null @@ -1,6 +0,0 @@ -fastapi -uvicorn -requests -websockets -frida - diff --git a/补全包/微信Frida_SDK_20260518/sdk/__init__.py b/补全包/微信Frida_SDK_20260518/sdk/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/补全包/微信Frida_SDK_20260518/sdk/frida/__init__.py b/补全包/微信Frida_SDK_20260518/sdk/frida/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/补全包/微信Frida_SDK_20260518/sdk/frida/frida_manager.py b/补全包/微信Frida_SDK_20260518/sdk/frida/frida_manager.py deleted file mode 100644 index 9ec11f589a..0000000000 --- a/补全包/微信Frida_SDK_20260518/sdk/frida/frida_manager.py +++ /dev/null @@ -1,166 +0,0 @@ -""" -工作手机 SDK · FridaManager - -负责手机端或服务器端连接 Frida、attach 微信进程、加载 Hook 脚本,并提供兼容 RPC 调用。 -设计重点:历史归档显示 Frida Python 对 rpc.exports 方法名存在混淆,因此这里采用多候选名兼容策略。 -""" -from __future__ import annotations - -from dataclasses import dataclass, field -from pathlib import Path -import re -import time -from typing import Any, Callable, Dict, Iterable, Optional - - -class FridaUnavailable(RuntimeError): - """当前环境未安装或无法连接 Frida。""" - - -class RpcMethodMissing(RuntimeError): - """Hook 中缺少指定 RPC 方法。""" - - -@dataclass -class FridaConfig: - device_host: str = "127.0.0.1" - device_port: int = 27042 - package_name: str = "com.tencent.mm" - process_name: str = "WeChat" - attach_timeout: float = 15.0 - prefer_usb: bool = False - hook_path: Optional[str] = None - - -@dataclass -class RpcResult: - ok: bool - action: str - method: str - data: Any = None - error: Optional[str] = None - elapsed_ms: int = 0 - - -class FridaManager: - def __init__(self, config: Optional[FridaConfig] = None): - self.config = config or FridaConfig() - self.frida = None - self.device = None - self.session = None - self.script = None - self.exports = None - self.loaded_hook_path: Optional[str] = None - - def _import_frida(self): - if self.frida is not None: - return self.frida - try: - import frida # type: ignore - except Exception as exc: - raise FridaUnavailable(f"未安装 frida Python 包或加载失败:{exc}") from exc - self.frida = frida - return frida - - def connect(self): - frida = self._import_frida() - if self.config.prefer_usb: - self.device = frida.get_usb_device(timeout=int(self.config.attach_timeout)) - else: - self.device = frida.get_device_manager().add_remote_device( - f"{self.config.device_host}:{self.config.device_port}" - ) - return self.device - - def attach(self, target: Optional[str] = None): - if self.device is None: - self.connect() - assert self.device is not None - target = target or self.config.package_name - last_error: Optional[Exception] = None - candidates: Iterable[Any] = [target, self.config.process_name] - for candidate in candidates: - try: - self.session = self.device.attach(candidate) - return self.session - except Exception as exc: # pragma: no cover - depends on phone - last_error = exc - try: - processes = self.device.enumerate_processes() - for proc in processes: - name = getattr(proc, 'name', '') or '' - pid = getattr(proc, 'pid', None) - if pid and (self.config.package_name in name or self.config.process_name.lower() in name.lower() or '微信' in name): - self.session = self.device.attach(pid) - return self.session - except Exception as exc: # pragma: no cover - last_error = exc - raise FridaUnavailable(f"无法 attach 微信进程:{last_error}") - - def load_script(self, hook_path: Optional[str] = None): - path = Path(hook_path or self.config.hook_path or '') - if not path.exists(): - raise FileNotFoundError(f"Hook 脚本不存在:{path}") - if self.session is None: - self.attach() - assert self.session is not None - source = path.read_text(encoding='utf-8') - self.script = self.session.create_script(source) - self.script.on('message', self._on_message) - self.script.load() - self.exports = getattr(self.script, 'exports_sync', None) or getattr(self.script, 'exports', None) - self.loaded_hook_path = str(path) - return self.script - - def _on_message(self, message, data): # pragma: no cover - runtime callback - # 生产环境可转发到 EventReporter;这里保持最小日志。 - print({'frida_message': message, 'data_len': len(data) if data else 0}) - - @staticmethod - def _snake(name: str) -> str: - return re.sub(r'(? RpcResult: - started = time.time() - if self.exports is None: - raise FridaUnavailable('Hook 脚本尚未加载,无法调用 RPC') - last_error = None - for candidate in self.method_candidates(method): - try: - fn = getattr(self.exports, candidate) - data = fn(*args, **kwargs) - return RpcResult(True, method, candidate, data=data, elapsed_ms=int((time.time() - started) * 1000)) - except AttributeError as exc: - last_error = exc - except Exception as exc: - return RpcResult(False, method, candidate, error=str(exc), elapsed_ms=int((time.time() - started) * 1000)) - # 兼容新版 frida 的 exports_sync.invoke 或 script.exports_sync.invoke - try: - invoke = getattr(self.exports, 'invoke') - data = invoke(method, list(args), kwargs or None) - return RpcResult(True, method, 'invoke', data=data, elapsed_ms=int((time.time() - started) * 1000)) - except Exception as exc: - last_error = exc - return RpcResult(False, method, '', error=f'RPC 方法不存在或不可调用:{method}; last={last_error}', elapsed_ms=int((time.time() - started) * 1000)) - - def cleanup(self): - for obj, method in [(self.script, 'unload'), (self.session, 'detach')]: - if obj is not None: - try: - getattr(obj, method)() - except Exception: - pass - self.script = None - self.session = None - self.exports = None diff --git a/补全包/微信Frida_SDK_20260518/sdk/frida/hook_executor.py b/补全包/微信Frida_SDK_20260518/sdk/frida/hook_executor.py deleted file mode 100644 index da046d0d35..0000000000 --- a/补全包/微信Frida_SDK_20260518/sdk/frida/hook_executor.py +++ /dev/null @@ -1,37 +0,0 @@ -"""将业务 action 映射到 Frida RPC。""" -from __future__ import annotations -from dataclasses import asdict -from typing import Any, Dict, Optional - -from sdk.frida.frida_manager import FridaManager, RpcResult -from sdk.wechat.wechat_actions import validate_action, list_actions - -class HookExecutor: - def __init__(self, manager: FridaManager): - self.manager = manager - - def execute(self, action: str, payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: - payload = payload or {} - try: - spec = validate_action(action, payload) - except Exception as exc: - return {'ok': False, 'action': action, 'error': str(exc), 'stage': 'validate'} - args = [payload[k] for k in spec.required] - # 允许额外参数透传给 Hook;Hook 若不支持会返回错误,便于报告定位。 - extra = {k: v for k, v in payload.items() if k not in spec.required} - result: RpcResult = self.manager.call(spec.rpc, *args, **extra) - return { - 'ok': result.ok, - 'action': action, - 'rpc': spec.rpc, - 'method_used': result.method, - 'module': spec.module, - 'description': spec.description, - 'safe': spec.safe, - 'data': result.data, - 'error': result.error, - 'elapsed_ms': result.elapsed_ms, - } - - def catalog(self): - return list_actions() diff --git a/补全包/微信Frida_SDK_20260518/sdk/wechat/__init__.py b/补全包/微信Frida_SDK_20260518/sdk/wechat/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/补全包/微信Frida_SDK_20260518/sdk/wechat/wechat_actions.py b/补全包/微信Frida_SDK_20260518/sdk/wechat/wechat_actions.py deleted file mode 100644 index c405fbb39a..0000000000 --- a/补全包/微信Frida_SDK_20260518/sdk/wechat/wechat_actions.py +++ /dev/null @@ -1,54 +0,0 @@ -"""微信动作目录:服务器、手机端 Agent、验证脚本共同使用。""" -from __future__ import annotations -from dataclasses import dataclass, field -from typing import Any, Dict, List - -@dataclass(frozen=True) -class WechatAction: - action: str - rpc: str - module: str - description: str - required: List[str] = field(default_factory=list) - safe: bool = True - verify: bool = True - -ACTIONS: List[WechatAction] = [ - WechatAction('ping', 'ping', 'system', 'Hook 连通性检测'), - WechatAction('get_connection_status', 'getConnectionStatus', 'system', '读取连接状态'), - WechatAction('screenshot', 'takeScreenshot', 'device', '手机当前屏幕截图'), - WechatAction('ui_dump', 'dumpUiTree', 'device', '导出当前 UI 树'), - WechatAction('go_home', 'goHome', 'navigation', '返回微信首页'), - WechatAction('open_contacts', 'openContacts', 'navigation', '打开通讯录'), - WechatAction('open_chats', 'openChats', 'navigation', '打开聊天列表'), - WechatAction('search_contact', 'searchContact', 'contact', '搜索联系人', ['keyword']), - WechatAction('get_contacts', 'getContacts', 'contact', '读取联系人列表'), - WechatAction('get_contact_profile', 'getContactProfile', 'contact', '读取联系人资料', ['wxid']), - WechatAction('add_friend', 'addFriend', 'contact', '添加好友', ['keyword'], safe=False), - WechatAction('accept_friend', 'acceptFriend', 'contact', '通过好友申请', ['wxid'], safe=False), - WechatAction('send_text', 'sendTextMessage', 'message', '发送文本消息', ['to', 'text'], safe=False), - WechatAction('send_image', 'sendImageMessage', 'message', '发送图片消息', ['to', 'path'], safe=False), - WechatAction('get_messages', 'getMessages', 'message', '读取消息列表', ['wxid']), - WechatAction('open_chat', 'openChat', 'message', '打开指定会话', ['wxid']), - WechatAction('create_group', 'createGroup', 'group', '创建群聊', ['members'], safe=False), - WechatAction('invite_group_member', 'inviteGroupMember', 'group', '邀请群成员', ['chatroom', 'members'], safe=False), - WechatAction('get_group_members', 'getGroupMembers', 'group', '读取群成员', ['chatroom']), - WechatAction('browse_channels', 'browseChannels', 'channels', '浏览视频号'), - WechatAction('add_favorite', 'addFavorite', 'favorite', '收藏当前内容'), - WechatAction('add_custom_emoji', 'addCustomEmoji', 'emoji', '添加自定义表情', safe=False), -] - -ACTION_MAP: Dict[str, WechatAction] = {item.action: item for item in ACTIONS} -RPC_MAP: Dict[str, WechatAction] = {item.rpc: item for item in ACTIONS} - -def list_actions() -> List[Dict[str, Any]]: - return [item.__dict__.copy() for item in ACTIONS] - -def validate_action(action: str, payload: Dict[str, Any]) -> WechatAction: - if action not in ACTION_MAP: - raise KeyError(f'未知微信动作:{action}') - spec = ACTION_MAP[action] - missing = [k for k in spec.required if k not in payload or payload[k] in (None, '')] - if missing: - raise ValueError(f'动作 {action} 缺少参数:{missing}') - return spec diff --git a/补全包/微信Frida_SDK_20260518/server/__init__.py b/补全包/微信Frida_SDK_20260518/server/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/补全包/微信Frida_SDK_20260518/server/app.py b/补全包/微信Frida_SDK_20260518/server/app.py deleted file mode 100644 index 4c7de97ef9..0000000000 --- a/补全包/微信Frida_SDK_20260518/server/app.py +++ /dev/null @@ -1,13 +0,0 @@ -from fastapi import FastAPI, WebSocket -from server.routes.wechat_frida import router as wechat_frida_router, phone_socket - -app = FastAPI(title='工作手机 SDK · 微信 Frida 控制服务', version='0.3.0') -app.include_router(wechat_frida_router) - -@app.get('/health') -def health(): - return {'ok': True, 'service': 'work-phone-sdk'} - -@app.websocket('/ws/phone') -async def ws_phone(websocket: WebSocket): - await phone_socket(websocket) diff --git a/补全包/微信Frida_SDK_20260518/server/routes/__init__.py b/补全包/微信Frida_SDK_20260518/server/routes/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/补全包/微信Frida_SDK_20260518/server/routes/wechat_frida.py b/补全包/微信Frida_SDK_20260518/server/routes/wechat_frida.py deleted file mode 100644 index 2f7b7d14f5..0000000000 --- a/补全包/微信Frida_SDK_20260518/server/routes/wechat_frida.py +++ /dev/null @@ -1,76 +0,0 @@ -"""FastAPI 微信 Frida 无线控制路由。""" -from __future__ import annotations -import asyncio -import json -import uuid -from dataclasses import dataclass, field -from typing import Any, Dict, Optional - -from fastapi import APIRouter, WebSocket, WebSocketDisconnect, HTTPException -from pydantic import BaseModel - -from sdk.wechat.wechat_actions import list_actions - -router = APIRouter(prefix='/api/v3/wechat-frida', tags=['wechat-frida']) - -@dataclass -class PhoneConn: - device_id: str - ws: WebSocket - meta: Dict[str, Any] = field(default_factory=dict) - pending: Dict[str, asyncio.Future] = field(default_factory=dict) - -phones: Dict[str, PhoneConn] = {} - -class ActionRequest(BaseModel): - device_id: str - action: str - payload: Dict[str, Any] = {} - timeout: float = 30.0 - -@router.get('/actions') -def actions(): - return {'ok': True, 'actions': list_actions()} - -@router.get('/devices') -def devices(): - return {'ok': True, 'devices': [{'device_id': k, 'meta': v.meta} for k, v in phones.items()]} - -@router.post('/execute') -async def execute(req: ActionRequest): - if req.device_id not in phones: - raise HTTPException(404, f'设备未连接:{req.device_id}') - conn = phones[req.device_id] - request_id = uuid.uuid4().hex - fut = asyncio.get_event_loop().create_future() - conn.pending[request_id] = fut - await conn.ws.send_text(json.dumps({'type': 'command', 'request_id': request_id, 'action': req.action, 'payload': req.payload}, ensure_ascii=False)) - try: - return await asyncio.wait_for(fut, timeout=req.timeout) - finally: - conn.pending.pop(request_id, None) - -async def phone_socket(websocket: WebSocket): - await websocket.accept() - device_id: Optional[str] = None - try: - async for raw in websocket.iter_text(): - msg = json.loads(raw) - if msg.get('type') == 'hello': - device_id = msg.get('device_id') or uuid.uuid4().hex - phones[device_id] = PhoneConn(device_id=device_id, ws=websocket, meta=msg) - await websocket.send_text(json.dumps({'type': 'hello_ack', 'device_id': device_id}, ensure_ascii=False)) - elif msg.get('type') == 'result': - rid = msg.get('request_id') - did = msg.get('device_id') or device_id - conn = phones.get(did or '') - if conn and rid in conn.pending and not conn.pending[rid].done(): - conn.pending[rid].set_result({'ok': True, 'device_id': did, 'result': msg.get('result')}) - elif msg.get('type') == 'error': - # 保留连接,错误由客户端下一次请求再显式返回。 - pass - except WebSocketDisconnect: - pass - finally: - if device_id and device_id in phones: - phones.pop(device_id, None) diff --git a/补全包/微信Frida_SDK_20260518/tests_mock_rpc.py b/补全包/微信Frida_SDK_20260518/tests_mock_rpc.py deleted file mode 100644 index c69d6c662b..0000000000 --- a/补全包/微信Frida_SDK_20260518/tests_mock_rpc.py +++ /dev/null @@ -1,21 +0,0 @@ -from sdk.frida.frida_manager import FridaManager -from sdk.frida.hook_executor import HookExecutor - -class Exports: - def ping(self): - return 'pong' - def getConnectionStatus(self): - return {'ok': True} - def sendTextMessage(self, to, text): - return {'to': to, 'text': text} - -m = FridaManager() -m.exports = Exports() -assert m.call('ping').ok -assert m.call('getConnectionStatus').ok -assert m.call('sendTextMessage', 'filehelper', 'hi').ok -ex = HookExecutor(m) -assert ex.execute('ping')['ok'] is True -assert ex.execute('send_text', {'to':'filehelper','text':'hi'})['ok'] is True -assert ex.execute('send_text', {'to':'filehelper'})['ok'] is False -print('mock_rpc_tests_passed') diff --git a/补全包/微信Frida_SDK_20260518/开发文档/10、项目管理/项目落地执行表.md b/补全包/微信Frida_SDK_20260518/开发文档/10、项目管理/项目落地执行表.md deleted file mode 100644 index ce8a49cc99..0000000000 --- a/补全包/微信Frida_SDK_20260518/开发文档/10、项目管理/项目落地执行表.md +++ /dev/null @@ -1,3 +0,0 @@ -# 项目落地执行表(补全包归档) - -> **已并入**:[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §1.2 diff --git a/补全包/微信Frida_SDK_20260518/开发文档/5、接口/微信Frida_API契约.md b/补全包/微信Frida_SDK_20260518/开发文档/5、接口/微信Frida_API契约.md deleted file mode 100644 index 8dfd9f5dd9..0000000000 --- a/补全包/微信Frida_SDK_20260518/开发文档/5、接口/微信Frida_API契约.md +++ /dev/null @@ -1,3 +0,0 @@ -# 微信 Frida API 契约(补全包归档) - -> **已并入**:[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §三 diff --git a/补全包/微信Frida_SDK_20260518/开发文档/8、部署/微信Frida无线部署与验证说明.md b/补全包/微信Frida_SDK_20260518/开发文档/8、部署/微信Frida无线部署与验证说明.md deleted file mode 100644 index 6aa007efc5..0000000000 --- a/补全包/微信Frida_SDK_20260518/开发文档/8、部署/微信Frida无线部署与验证说明.md +++ /dev/null @@ -1,3 +0,0 @@ -# 微信 Frida 无线部署与验证说明(补全包归档) - -> **已并入**:[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §八 diff --git a/补全包/微信Frida_SDK_20260518_补全包.zip b/补全包/微信Frida_SDK_20260518_补全包.zip deleted file mode 100644 index 63b563b10a..0000000000 Binary files a/补全包/微信Frida_SDK_20260518_补全包.zip and /dev/null differ