chore: 移除根目录补全包目录（已收编进 sdk/）
删除 补全包/ 与 工作手机SDK补全包_微信Frida/ 冗余副本； 真源见 sdk/agent/hook、资料/archive/ 历史 ZIP。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Manus AI
@@ -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/<device_id>
|
||||
```
|
||||
|
||||
Hook 替换与 RPC 四层兼容见主手册 **§七**。
|
||||
@@ -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'(?<!^)(?=[A-Z])', '_', name).lower()
|
||||
|
||||
@classmethod
|
||||
def method_candidates(cls, name: str):
|
||||
snake = cls._snake(name)
|
||||
lower = name.lower()
|
||||
candidates = [name, snake, lower]
|
||||
seen = set()
|
||||
for item in candidates:
|
||||
if item and item not in seen:
|
||||
seen.add(item)
|
||||
yield item
|
||||
|
||||
def call(self, method: str, *args, **kwargs) -> 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}')
|
||||
@@ -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'); }
|
||||
};
|
||||
@@ -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()
|
||||
@@ -1,6 +0,0 @@
|
||||
fastapi
|
||||
uvicorn
|
||||
requests
|
||||
websockets
|
||||
frida
|
||||
|
||||
@@ -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'(?<!^)(?=[A-Z])', '_', name).lower()
|
||||
|
||||
@classmethod
|
||||
def method_candidates(cls, name: str):
|
||||
snake = cls._snake(name)
|
||||
lower = name.lower()
|
||||
candidates = [name, snake, lower]
|
||||
seen = set()
|
||||
for item in candidates:
|
||||
if item and item not in seen:
|
||||
seen.add(item)
|
||||
yield item
|
||||
|
||||
def call(self, method: str, *args, **kwargs) -> 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
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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')
|
||||
@@ -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')
|
||||
@@ -1,6 +0,0 @@
|
||||
# 项目落地执行表（补全包归档）
|
||||
|
||||
> **已并入**：[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §1.2
|
||||
> **进度真源**：[开发文档/10、项目管理/开发进度总表.md](../../../开发文档/10、项目管理/开发进度总表.md)（99.5%）
|
||||
|
||||
2026-05-18 补全包阶段约 55% 的记录已归档；主线进度以总表为准。
|
||||
@@ -1,6 +0,0 @@
|
||||
# 系统架构（补全包归档）
|
||||
|
||||
> **已并入**：[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §一、§二
|
||||
> 完整架构见 [开发文档/2、架构/01-总览/系统架构.md](../../../开发文档/2、架构/01-总览/系统架构.md)
|
||||
|
||||
本文件为补全包内副本，勿再扩写。
|
||||
@@ -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` 调用差异问题。
|
||||
@@ -1,5 +0,0 @@
|
||||
# 微信 Frida 控制接口契约（补全包归档）
|
||||
|
||||
> **已并入**：[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §三、§七
|
||||
|
||||
统一 API：`POST /api/v3/hook/execute` · 代码：`sdk/app/routers/unified.py`
|
||||
@@ -1,5 +0,0 @@
|
||||
# 微信 Frida 无线部署与验证说明（补全包归档）
|
||||
|
||||
> **已并入**：[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §五、§八
|
||||
|
||||
验证脚本主线版：`tools/wireless_frida_wechat_batch_verify.py`
|
||||
@@ -1,12 +0,0 @@
|
||||
# 补全包目录
|
||||
|
||||
> **文档已并入开发文档** — 请勿在此维护正文。
|
||||
> **唯一主手册**：[开发文档/9、手册/工作手机·五图总览与使用手册.md](../开发文档/9、手册/工作手机·五图总览与使用手册.md)（v2.0 补全包并入版）
|
||||
|
||||
## 子目录
|
||||
|
||||
| 目录 | 说明 |
|
||||
|------|------|
|
||||
| [微信Frida_SDK_20260518/](微信Frida_SDK_20260518/) | 2026-05-18 归档原型代码（只读对照） |
|
||||
|
||||
日常开发请使用主线 **`sdk/`**，部署与 API 见主手册 §六～§八。
|
||||
@@ -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` 等。
|
||||
|
||||
## 快速启动（历史命令，仅供对照）
|
||||
|
||||
<details>
|
||||
<summary>展开旧版启动命令</summary>
|
||||
|
||||
服务器端：
|
||||
|
||||
```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
|
||||
```
|
||||
|
||||
</details>
|
||||
|
||||
**现行启动**见主手册 §六、§八（端口 **8899**，API **`/api/v3/hook/execute`**）。
|
||||
@@ -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'(?<!^)(?=[A-Z])', '_', name).lower()
|
||||
|
||||
@classmethod
|
||||
def method_candidates(cls, name: str):
|
||||
snake = cls._snake(name)
|
||||
lower = name.lower()
|
||||
candidates = [name, snake, lower]
|
||||
seen = set()
|
||||
for item in candidates:
|
||||
if item and item not in seen:
|
||||
seen.add(item)
|
||||
yield item
|
||||
|
||||
def call(self, method: str, *args, **kwargs) -> 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}')
|
||||
@@ -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'); }
|
||||
};
|
||||
@@ -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()
|
||||
@@ -1,6 +0,0 @@
|
||||
fastapi
|
||||
uvicorn
|
||||
requests
|
||||
websockets
|
||||
frida
|
||||
|
||||
@@ -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'(?<!^)(?=[A-Z])', '_', name).lower()
|
||||
|
||||
@classmethod
|
||||
def method_candidates(cls, name: str):
|
||||
snake = cls._snake(name)
|
||||
lower = name.lower()
|
||||
candidates = [name, snake, lower]
|
||||
seen = set()
|
||||
for item in candidates:
|
||||
if item and item not in seen:
|
||||
seen.add(item)
|
||||
yield item
|
||||
|
||||
def call(self, method: str, *args, **kwargs) -> 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
|
||||
@@ -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()
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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')
|
||||
@@ -1,3 +0,0 @@
|
||||
# 项目落地执行表（补全包归档）
|
||||
|
||||
> **已并入**：[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §1.2
|
||||
@@ -1,3 +0,0 @@
|
||||
# 微信 Frida API 契约（补全包归档）
|
||||
|
||||
> **已并入**：[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §三
|
||||
@@ -1,3 +0,0 @@
|
||||
# 微信 Frida 无线部署与验证说明（补全包归档）
|
||||
|
||||
> **已并入**：[开发文档/9、手册/工作手机·五图总览与使用手册.md](../../../开发文档/9、手册/工作手机·五图总览与使用手册.md) §八
|
||||
Binary file not shown.
Reference in New Issue
Block a user