""" Hook Executor v3.0 — 将 unified API 动作映射到 Frida rpc.exports 调用 职责： - 接收 unified 指令（send_message / get_contacts / add_friend / …） - 通过 FridaManager.call_rpc 调用微信内部函数 - Hook 失败时降级到 ADB/u2 通道 对接： - 服务端 unified.py → WebSocket 下发 → Agent → HookExecutor - 或 ADB 模式：unified.py → HookExecutor（本地 Frida） 支持 110 个操作 / 24 模块，与 wechat_hook_v3.0.js rpc.exports 一一对应 """ import logging from typing import Dict, Any from .frida_manager import FridaManager logger = logging.getLogger(__name__) ACTION_TO_RPC: Dict[str, str] = { # ── H17 消息发送 ── "send_message": "sendMessage", "send_group_message": "sendGroupMessage", # ── H15 消息获取 ── "get_messages": "getMessages", "get_recent_messages": "getRecentMessages", "search_messages": "searchMessages", # ── H16 联系人 ── "get_contacts": "getContacts", "get_contact_info": "getContactInfo", "search_contacts": "searchContacts", # ── H18/H19 好友管理 ── "add_friend": "addFriend", "accept_friend": "acceptFriend", "delete_friend": "deleteFriend", "set_friend_remark": "setFriendRemark", "get_friend_requests": "getFriendRequests", # ── H22 群管理 ── "get_groups": "getGroups", "get_group_info": "getGroupInfo", "get_group_members": "getGroupMembers", "create_group": "createGroup", "invite_to_group": "inviteToGroup", "remove_from_group": "removeFromGroup", "set_group_announcement": "setGroupAnnouncement", "set_group_name": "setGroupName", "quit_group": "quitGroup", # ── H20/H21 朋友圈 ── "post_moments": "postMoments", "get_moments": "getMoments", "like_moments": "likeMoments", "comment_moments": "commentMoments", "delete_moments": "deleteMoments", # ── H23 账号管理 ── "get_profile": "getProfile", "check_account_status": "checkAccountStatus", "set_nickname": "setNickname", "set_signature": "setSignature", "set_avatar": "setAvatar", "set_sex": "setSex", "set_region": "setRegion", "set_whats_up": "setWhatUp", # ── H24 账号安全 ── "unblock_self": "unblockSelf", "change_password": "changePassword", "bind_phone": "bindPhone", "unbind_phone": "unbindPhone", "get_login_devices": "getLoginDevices", "remove_login_device": "removeLoginDevice", "enable_fingerprint": "enableFingerprint", "set_account_protection": "setAccountProtection", # ── H25 支付 ── "send_red_packet": "sendRedPacket", "receive_red_packet": "receiveRedPacket", "send_transfer": "sendTransfer", "receive_transfer": "receiveTransfer", "get_wallet_balance": "getWalletBalance", "get_transaction_history": "getTransactionHistory", # ── H26 二维码 ── "scan_qr_code": "scanQrCode", "generate_my_qr_code": "generateMyQrCode", "generate_group_qr_code": "generateGroupQrCode", "add_friend_by_qr": "addFriendByQr", # ── H27 视频号 ── "browse_channels": "browseChannels", "like_channel_video": "likeChannelVideo", "comment_channel_video": "commentChannelVideo", "follow_channel": "followChannel", "unfollow_channel": "unfollowChannel", "share_channel_video": "shareChannelVideo", # ── H28 标签 ── "get_labels": "getLabels", "create_label": "createLabel", "delete_label": "deleteLabel", "set_contact_label": "setContactLabel", "get_contacts_by_label": "getContactsByLabel", # ── H29 收藏 ── "get_favorites": "getFavorites", "add_favorite": "addFavorite", "delete_favorite": "deleteFavorite", # ── H30 设置 ── "set_privacy": "setPrivacy", "set_notification": "setNotification", "clear_chat_history": "clearChatHistory", "set_chat_background": "setChatBackground", "set_do_not_disturb": "setDoNotDisturb", "pin_chat": "pinChat", # ── H31 搜索 ── "global_search": "globalSearch", # ── H32 小程序 ── "open_mini_program": "openMiniProgram", "get_recent_mini_programs": "getRecentMiniPrograms", "share_mini_program": "shareMiniProgram", # ── H33 文件传输 ── "send_image": "sendImage", "send_video": "sendVideo", "send_file": "sendFile", "send_voice": "sendVoice", "send_location": "sendLocation", "send_card": "sendCard", "send_link": "sendLink", # ── H34 消息转发 ── "forward_message": "forwardMessage", "forward_multiple": "forwardMultiple", "revoke_message": "revokeMessage", # ── H35 注册/登录 ── "register_account": "registerAccount", "login_by_password": "loginByPassword", "login_by_sms": "loginBySms", "logout": "logout", "switch_account": "switchAccount", "auto_register": "autoRegister", "check_login_state": "checkLoginState", "get_sim_phone": "getSimPhone", # ── H36 公众号 ── "get_official_accounts": "getOfficialAccounts", "follow_official_account": "followOfficialAccount", "unfollow_official_account": "unfollowOfficialAccount", "get_official_account_articles": "getOfficialAccountArticles", # ── H37 表情 ── "send_emoji": "sendEmoji", "add_custom_emoji": "addCustomEmoji", # ── H38 浮窗 ── "add_to_float": "addToFloat", "remove_from_float": "removeFromFloat", # ── H39 设备信息 ── "get_device_info": "getDeviceInfo", "get_storage_info": "getStorageInfo", "get_network_info": "getNetworkInfo", # ── 系统 ── "get_hook_status": "getHookStatus", "get_process_info": "getProcessInfo", "get_wechat_version": "getWechatVersion", "batch_execute": "batchExecute", # ── 矩阵 v8.0.56 补全（Frida RPC 直映射）── "get_safety_center": "getSafetyCenter", "check_restrictions": "checkRestrictions", "get_top_stories": "getTopStories", "get_wechat_steps": "getWechatSteps", "get_sticker_list": "getStickerList", "show_payment_code": "showPaymentCode", "like_wechat_steps": "likeWechatSteps", "clear_cache": "clearCache", "check_for_update": "checkForUpdate", } # unified / ADB 引擎方法名 → HookExecutor 标准 action（矩阵 v8.0.56 真机验收） ACTION_ALIASES: Dict[str, str] = { "recall_message": "revoke_message", "search_contact": "search_contacts", "get_friend_info": "get_contact_info", "account_status": "check_account_status", "safety_center": "get_safety_center", "set_chat_top": "pin_chat", "set_mute_chat": "set_do_not_disturb", "get_video_list": "browse_channels", "show_my_qr": "generate_my_qr_code", "view_wallet": "get_wallet_balance", "view_transactions": "get_transaction_history", "wechat_search": "global_search", "top_stories": "get_top_stories", "get_steps": "get_wechat_steps", "get_tags": "get_labels", "create_tag": "create_label", "delete_tag": "delete_label", "set_remark": "set_friend_remark", "add_to_favorites": "add_favorite", "clear_history": "clear_chat_history", "like_video": "like_channel_video", "send_voice_message": "send_voice", "set_group_notice": "set_group_announcement", "transfer": "send_transfer", "receive_payment": "receive_transfer", "unblock_account": "unblock_self", "unblock_appeal": "unblock_self", "appeal_restriction": "unblock_self", "unblock_with_sms": "unblock_self", "set_gender": "set_sex", "set_moments_cover": "set_privacy", "set_moments_privacy": "set_privacy", "forward_moments_link": "post_moments", "open_miniprogram": "open_mini_program", "do_not_disturb": "set_do_not_disturb", "like_steps": "like_wechat_steps", "toggle_do_not_disturb": "set_do_not_disturb", "clear_cache": "clear_cache", "check_for_update": "check_for_update", "send_file_from_chat": "send_file", "send_voice_message": "send_voice", "add_to_favorites": "add_favorite", } def resolve_action(action: str) -> str: return ACTION_ALIASES.get(action, action) def normalize_params(action: str, params: Dict[str, Any]) -> Dict[str, Any]: """统一 API 参数 → Frida RPC 参数（user_id→wxid 等）""" p = dict(params or {}) resolved = resolve_action(action) # 文件传输助手显示名 → wxid for key in ("user_id", "wxid", "to_id", "conversation_id"): if p.get(key) in ("文件传输助手", "File Transfer"): p[key] = "filehelper" if p.get("user_id") and not p.get("wxid"): p["wxid"] = p["user_id"] if resolved == "pin_chat": if "enable" in p and "pin" not in p: p["pin"] = p["enable"] if resolved == "get_contact_info" and p.get("user_id") and not p.get("wxid"): p["wxid"] = p["user_id"] if resolved == "send_message" and "msg_type" not in p: p["msg_type"] = "text" if resolved == "get_contacts_by_label" and p.get("tag_name") and not p.get("label_name"): p["label_name"] = p["tag_name"] if resolved == "create_label" and p.get("tag_name") and not p.get("name"): p["name"] = p["tag_name"] if p.get("to_id") and not p.get("user_id"): p["user_id"] = p["to_id"] if p.get("user_id") and not p.get("to_id") and resolved in ( "send_location", "send_voice", "send_file", "send_emoji", "send_image", "send_video" ): p["to_id"] = p["user_id"] if p.get("name") and not p.get("label"): p["label"] = p["name"] if p.get("emoji_name") and not p.get("emoji_md5"): p["emoji_md5"] = p["emoji_name"] if p.get("name") and not p.get("app_id") and resolved == "open_mini_program": p["app_id"] = p["name"] if p.get("index") is not None and not p.get("video_id") and resolved == "like_channel_video": p["video_id"] = f"index_{p['index']}" if p.get("privacy_type") and not p.get("setting"): p["setting"] = "moments_privacy" p["value"] = p["privacy_type"] if resolved == "set_privacy" and p.get("days") and not p.get("setting"): p["setting"] = "moments_days" p["value"] = p["days"] if p.get("content") and not p.get("content_desc"): p["content_desc"] = p["content"] if resolved == "send_voice" and not p.get("voice_path") and p.get("duration"): p["voice_path"] = f"/sdcard/workphone/voice_{int(p['duration'])}s.amr" return p MODULE_NAMES = { "H15": "消息接收", "H16": "联系人", "H17": "消息发送", "H18": "好友请求", "H19": "好友管理", "H20": "朋友圈发布", "H21": "朋友圈浏览", "H22": "群管理", "H23": "账号管理", "H24": "账号安全", "H25": "支付", "H26": "二维码", "H27": "视频号", "H28": "标签", "H29": "收藏", "H30": "设置", "H31": "搜索", "H32": "小程序", "H33": "文件传输", "H34": "消息转发", "H35": "注册/登录", "H36": "公众号", "H37": "表情", "H38": "浮窗", "H39": "设备信息", } class HookExecutor: """将 unified 动作转为 Frida RPC 调用 — 110 个操作全覆盖""" def __init__(self, frida_mgr: FridaManager): self.frida = frida_mgr @property def available(self) -> bool: return self.frida.connected def execute(self, action: str, params: Dict[str, Any] = None) -> Dict[str, Any]: if not self.available: return {"success": False, "error": "Hook 通道不可用（Frida 未连接）", "channel": "hook"} canonical = resolve_action(action) norm = normalize_params(action, params) rpc_method = ACTION_TO_RPC.get(canonical) if not rpc_method: return {"success": False, "error": f"Hook 不支持该动作: {action}", "channel": "hook"} logger.info(f"[HookExecutor] {action} → {canonical} → rpc.{rpc_method}") result = self.frida.call_rpc(rpc_method, norm) result["channel"] = "hook" result["action_resolved"] = canonical return result def supports(self, action: str) -> bool: return resolve_action(action) in ACTION_TO_RPC def get_supported_actions(self) -> list: return sorted(ACTION_TO_RPC.keys()) def get_status(self) -> Dict[str, Any]: return { "available": self.available, "total_actions": len(ACTION_TO_RPC), "supported_actions": self.get_supported_actions(), "modules": MODULE_NAMES, "frida": self.frida.get_status(), }