diff --git a/README.md b/README.md index d729c3f..c3e5418 100644 --- a/README.md +++ b/README.md @@ -2,10 +2,12 @@ 一个本地运行的 Python + SQLite 工具。从群聊和私聊里发现反复提问、风险/事故、重复踩坑、后续复用线索，将其整理为带原文证据的知识需求卡，对应 CREATES 的 Choose → Retrieve → Extract / Validate。 -## 页面连接 Mac 微信（v0.2） +## 页面连接 Mac 微信（v0.3） 新增“连接 Mac 微信”向导：扫描账号 → 读取本机密钥 → 退出微信后备份、解密、解析并导入。密钥仅存运行内存；权限不足或版本不匹配会显示具体失败原因。首次仍需在 Mac 启动本地服务，之后通过页面操作。 +v0.3 将进程访问失败细分为签名阻塞、未运行和多进程等情况，管理员模式也保留诊断。新增需要用户预览并明确同意的临时调试副本路径：完整备份账号及登录配置后，只对独立副本添加调试权限，不改原版或 SIP。副本可能访问同一账号数据，且仍可能无法启动或取钥；操作影响和限制均在页面展示。页面标题同时移除重复的英文标签，统一中文。 + **接入属于实验性功能，尚未在目标 Intel Mac 微信 4.1.0.18 实机验证，不保证系统允许读取密钥。** 使用流程、权限与格式限制见 [Mac 接入说明](docs/mac-connection.md)。 ## 交付状态 diff --git a/docs/mac-connection.md b/docs/mac-connection.md index 002f44a..9ebd01a 100644 --- a/docs/mac-connection.md +++ b/docs/mac-connection.md @@ -1,23 +1,38 @@ -# 页面连接 Mac 微信（v0.2，实验性） +# 页面连接 Mac 微信（v0.3，实验性） 在安装微信的目标 Mac 上更新代码，重新启动 `python3 run.py`，浏览器打开本机地址，进入“连接 Mac 微信”。此后无需手动运行诊断或逐个导出数据库。 ## 页面操作 1. **扫描本机微信**：发现 `xwechat_files/<账号>/db_storage`，显示消息分片、联系人库与密钥匹配数量。自定义位置点“选择数据文件夹”。旧版 3.x 数据不会自动混入。 -2. **读取本机密钥**：先登录微信并打开目标聊天，再点击读取。普通权限失败时，可主动点击管理员按钮，使用 macOS 系统密码框授权；工具不保存密码。系统仍可能拒绝访问，授权并不保证成功。已有密钥时可选择本机 JSON 文件。 +2. **检查签名与权限 / 读取本机密钥**：先登录微信并打开目标聊天。权限诊断显示强化运行时、允许调试与系统拒绝码；管理员按钮的结果也会保留这些信息，不再只显示通用失败。已有密钥时可选择本机 JSON 文件。 3. **安装文本解压支持**：需要 Zstandard 压缩正文时点击。通过 pip 从 PyPI 安装固定版本 `zstandard==0.23.0` wheel 到 `data/mac_deps`，不全局安装、不发送聊天；需要网络和可用的 pip。 4. **备份并导入**：先正常退出微信，保持知微运行，再点击。工具将所需数据库、WAL/SHM 复制到 `data/mac_sources/<任务>/backup`，可读库保存在同任务 `readable`。请预留至少约四倍所选数据库大小的磁盘空间。 5. 成功后进入“发现知识需求”，选择时间段、全部会话或指定群聊/私聊。业务消息分片默认不导入，需要时单独勾选。 浏览器与本地服务须运行在安装微信的同一台 Mac；Windows 页面不能读取另一台 Mac 的微信。不需要向开发者上传数据库、聊天或密钥。 +## 管理员授权仍被拒绝时 + +签名的调试限制和管理员身份是不同的检查。Apple 文档说明，调试工具对第三方应用取得任务端口还取决于目标是否允许调试。管理员按钮并不自动改变目标签名；因此反复输入密码可能一直失败。 + +页面新增可选的**临时调试副本**流程： + +1. 选择账号，点击“预览临时副本方案”，查看原版位置、副本位置、完整备份位置、空间估计与影响。这一步不会修改微信。 +2. 阅读影响，正常退出所有微信，勾选同意后点击“同意并备份、创建临时副本”。预览有效 30 分钟，一次性使用，切换账号必须重新预览。API 也会检查同意字段和预览令牌。 +3. 工具先完整复制所选账号目录和 `all_users` 登录配置，再创建独立 `WeChat.app` 副本。账号包含图片/附件时备份可能较大。文件清单、大小或源文件状态变化会中止，不签名、不启动。 +4. 只对副本添加 `com.apple.security.get-task-allow`，使用本机 `codesign` 重新签名并验证。原版应用不重签，SIP、系统开发者策略和隔离属性不修改。应用符号链接越界、原版变化或签名检查失败会中止。 +5. 准备成功后，点击“启动已准备的临时副本”，按 macOS 提示打开并登录。副本保留同一应用标识，可能使用或更新同一账号本地数据，也可能要求重新登录或系统授权，因此必须先备份。不要在副本中发消息、升级或作为日常客户端使用。 +6. 在副本中打开目标聊天，再点击读取密钥；必要时由你主动选择管理员读取。取得匹配密钥后，退出副本，点击备份并导入。完成后退出副本、恢复原版使用。 + +**此路径仍未在你的 Intel Mac 4.1.0.18 上实测，不保证副本启动、登录、原账号数据访问或密钥读取成功。** 系统拦截时不会自动绕过；失败保留备份，不自动恢复、覆盖或删除用户数据。副本具有更宽的调试权限，使用前需在页面明确同意，不会因为普通读取失败就自动创建。 + ## 实际边界 **尚未通过 Intel Mac 微信 4.1.0.18 的实机验收。** 开发环境是 Windows；合成数据库与模拟测试无法证明真实微信版本可用。 - 密钥扫描通过 macOS Mach 只读接口，验证目标进程路径为 `WeChat.app/Contents/MacOS/WeChat`。最多扫描 45 秒/2 GiB 的可读可写非执行内存，仅筛选 SQLCipher 格式的 64/96 位十六进制候选，不保存内存转储。 -- 默认普通用户读取；只有用户点击管理员按钮才触发系统授权。微信签名与 macOS 保护仍可能拒绝附加。工具不重签微信、不注入代码、不关闭 SIP，也不保证取得密钥。失败在页面说明，可导入已有本地密钥继续。 +- 默认普通用户读取；只有用户点击管理员按钮才触发系统授权。微信签名与 macOS 保护仍可能拒绝附加。工具不重签原版微信、不注入代码、不关闭 SIP。临时副本仅在页面预览并明确同意后重新签名，不保证取得密钥。 - 密钥只在进程内存，状态 API 只返回匹配数量，不回传密钥；重启后需重新读取/导入。密钥不进入模型请求、日志、Git 或发布包。 - 加密读取仅支持 SQLCipher 4 常见 raw AES-256 key、4096 字节页、80 字节保留区、SHA512 页认证与 little-endian 页编号。其他参数明确失败，不跳过认证。 - 逐页验证 HMAC 后解密；WAL 校验头、salt 与滚动校验和，仅回放到最后一次提交，不回放已校验但未提交的事务。截断、损坏或格式不匹配会停止；可读输出还会通过 SQLite `quick_check`。 @@ -34,3 +49,4 @@ - [SQLCipher 页认证与参数实现](https://github.com/sqlcipher/sqlcipher/blob/master/src/sqlcipher.c) - [SQLite WAL 文件格式](https://www.sqlite.org/fileformat2.html#walformat) - [Apple Mach 内存区域结构](https://github.com/apple-oss-distributions/xnu/blob/main/osfmk/mach/vm_region.h) +- [Apple 调试权限说明](https://developer.apple.com/documentation/bundleresources/entitlements/com.apple.security.cs.debugger) diff --git a/scripts/package_release.py b/scripts/package_release.py index ef021fe..13f2c6f 100644 --- a/scripts/package_release.py +++ b/scripts/package_release.py @@ -9,7 +9,7 @@ DIRS = ['wechat_tool', 'scripts', 'examples', 'tests', 'docs'] def main(): - out = ROOT / 'dist' / 'zhiwei-wechat-analysis-v0.2.zip' + out = ROOT / 'dist' / 'zhiwei-wechat-analysis-v0.3.zip' out.parent.mkdir(exist_ok=True) paths = [ROOT / p for p in FILES] for directory in DIRS: diff --git a/tests/test_mac_recovery.py b/tests/test_mac_recovery.py new file mode 100644 index 0000000..7410638 --- /dev/null +++ b/tests/test_mac_recovery.py @@ -0,0 +1,127 @@ +import json +import plistlib +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from wechat_tool.mac_connect import MacConnection +from wechat_tool.mac_keys import permission_diagnostic +from wechat_tool.mac_recovery import DebugRecovery + + +class PermissionTests(unittest.TestCase): + def test_signed_target_is_distinct_from_administrator_permission(self): + proc = {'pid':123,'path':'/Applications/WeChat.app/Contents/MacOS/WeChat'} + with patch('wechat_tool.mac_keys.wechat_processes', return_value=[proc]), patch('wechat_tool.mac_keys.signing_flags', return_value={'hardened_runtime':True,'get_task_allow':False}): + result=permission_diagnostic(5,123) + self.assertEqual(result['code'],'target_restricted') + self.assertEqual(result['task_for_pid_code'],5) + self.assertIn('反复输入密码',result['summary']) + with patch('wechat_tool.mac_keys.wechat_processes',return_value=[]): + self.assertEqual(permission_diagnostic()['code'],'no_process') + with patch('wechat_tool.mac_keys.wechat_processes',return_value=[proc,dict(proc,pid=456)]): + self.assertEqual(permission_diagnostic()['code'],'multiple_processes') + + def test_administrator_helper_preserves_structured_failure(self): + diagnostic={'code':'target_restricted','summary':'目标签名拒绝调试','task_for_pid_code':5} + output=json.dumps({'error':'目标签名拒绝调试','diagnostic':diagnostic}) + with tempfile.TemporaryDirectory() as directory: + service=MacConnection(directory);service.job={'status':'running'} + with patch('wechat_tool.mac_connect.subprocess.run',return_value=subprocess.CompletedProcess([],0,output,'')): + service.work('id','capture',{'administrator':True},None,[]) + self.assertEqual(service.job['status'],'failed') + self.assertEqual(service.job['error_code'],'target_restricted') + self.assertEqual(service.status()['diagnostic'],diagnostic) + + +class RecoveryTests(unittest.TestCase): + def setUp(self): + self.temp=tempfile.TemporaryDirectory() + self.base=Path(self.temp.name) + self.app=self.base/'Applications/WeChat.app' + for directory in ('Contents/MacOS','Contents/_CodeSignature'): + (self.app/directory).mkdir(parents=True) + (self.app/'Contents/Info.plist').write_bytes(plistlib.dumps({'CFBundleIdentifier':'com.tencent.xinWeChat','CFBundleExecutable':'WeChat','CFBundleShortVersionString':'4.1.0'})) + (self.app/'Contents/MacOS/WeChat').write_bytes(b'fake-app-binary') + (self.app/'Contents/_CodeSignature/CodeResources').write_bytes(b'fake-resource-seal') + self.account=self.base/'xwechat_files/account/db_storage' + self.account.mkdir(parents=True) + (self.account/'test.db').write_bytes(b'account-original') + all_users=self.account.parent.parent/'all_users' + all_users.mkdir() + (all_users/'login-config').write_bytes(b'login-original') + self.recovery=DebugRecovery(self.base/'workspace') + self.command_patch=patch('wechat_tool.mac_recovery.run',return_value=subprocess.CompletedProcess([],0,b'',b'')) + self.runner=self.command_patch.start() + + def tearDown(self): + self.command_patch.stop() + self.temp.cleanup() + + def plan(self): + return self.recovery.plan(self.account,app=self.app) + + def test_preview_is_readonly_and_consent_is_required(self): + result=self.plan() + self.assertFalse((self.base/'workspace').exists()) + self.assertTrue(result['effects']) + with self.assertRaisesRegex(ValueError,'明确勾选'): + self.recovery.consume(result['plan_id'],False,self.account) + self.assertTrue(all('--force' not in call.args[0] for call in self.runner.call_args_list)) + self.recovery.consume(result['plan_id'],True,self.account) + with self.assertRaisesRegex(ValueError,'已使用'): + self.recovery.consume(result['plan_id'],True,self.account) + + def test_preparation_backs_up_before_signing_copy_and_preserves_original(self): + result=self.plan() + plan=self.recovery.consume(result['plan_id'],True,self.account) + original={str(p.relative_to(self.app)):p.read_bytes() for p in self.app.rglob('*') if p.is_file()} + def grants(app): + return {'com.apple.security.get-task-allow':app!=self.app} + def command(args,timeout=90): + if '--force' in args: + self.assertEqual(Path(args[-1]),plan['root']/'WeChat.app') + self.assertEqual((plan['root']/'backup/0-account/db_storage/test.db').read_bytes(),b'account-original') + self.assertEqual((plan['root']/'backup/1-all_users/login-config').read_bytes(),b'login-original') + return subprocess.CompletedProcess(args,0,b'',b'') + self.runner.side_effect=command + with patch('wechat_tool.mac_recovery.wechat_pid',return_value=None),patch('wechat_tool.mac_recovery.entitlements',side_effect=grants): + prepared=self.recovery.prepare(plan,lambda _:None) + self.assertTrue(prepared['original_unchanged']) + self.assertEqual(original,{str(p.relative_to(self.app)):p.read_bytes() for p in self.app.rglob('*') if p.is_file()}) + mutations=[c.args[0] for c in self.runner.call_args_list if '--force' in c.args[0]] + self.assertEqual(len(mutations),1) + self.assertNotIn(str(self.app),mutations[0]) + with patch('wechat_tool.mac_recovery.wechat_pid',return_value=None): + self.recovery.launch() + self.assertEqual(self.runner.call_args.args[0],['/usr/bin/open','-n',prepared['copy']]) + + def test_running_wechat_or_changed_original_prevents_preparation(self): + result=self.plan();plan=self.recovery.consume(result['plan_id'],True,self.account) + with patch('wechat_tool.mac_recovery.wechat_pid',return_value=123): + with self.assertRaisesRegex(ValueError,'退出所有微信'): + self.recovery.prepare(plan,lambda _:None) + (self.app/'Contents/MacOS/WeChat').write_bytes(b'changed') + with patch('wechat_tool.mac_recovery.wechat_pid',return_value=None): + with self.assertRaisesRegex(ValueError,'发生变化'): + self.recovery.prepare(plan,lambda _:None) + self.assertIsNone(self.recovery.prepared) + + def test_unknown_or_tampered_copy_cannot_be_launched(self): + with self.assertRaisesRegex(ValueError,'先预览'): + self.recovery.launch() + self.recovery.prepared={'app':self.app,'fingerprint':'wrong'} + with patch('wechat_tool.mac_recovery.wechat_pid',return_value=None): + with self.assertRaisesRegex(ValueError,'已变化'): + self.recovery.launch() + + def test_account_switch_and_expired_plan_are_rejected(self): + result=self.plan() + with self.assertRaisesRegex(ValueError,'账号发生变化'): + self.recovery.consume(result['plan_id'],True,self.account/'other') + result=self.plan() + self.recovery.plans[result['plan_id']]['created']-=1801 + with self.assertRaisesRegex(ValueError,'过期'): + self.recovery.consume(result['plan_id'],True,self.account) diff --git a/wechat_tool/mac_connect.py b/wechat_tool/mac_connect.py index a16b9c7..3b2710a 100644 --- a/wechat_tool/mac_connect.py +++ b/wechat_tool/mac_connect.py @@ -16,7 +16,8 @@ from pathlib import Path from . import store from .cipher import PAGE, decrypt_database, matches from .importer import import_rows -from .mac_keys import scan_keys, wechat_pid +from .mac_keys import AccessDenied, permission_diagnostic, scan_keys, wechat_pid +from .mac_recovery import DebugRecovery from .wechat_native import message_rows @@ -46,6 +47,8 @@ class MacConnection: self.data = Path(data) self.accounts, self.keys = {}, [] self.job = None + self.diagnostic = None + self.recovery = DebugRecovery(self.data) self.lock = threading.RLock() deps = self.data / 'mac_deps' if deps.is_dir(): @@ -59,7 +62,18 @@ class MacConnection: with self.lock: return {'available': platform.system() == 'Darwin', 'architecture': platform.machine(), 'accounts': [self.describe(key, path) for key,path in self.accounts.items()], - 'keys_ready': bool(self.keys), 'job': dict(self.job) if self.job else None} + 'keys_ready': bool(self.keys), 'job': dict(self.job) if self.job else None, + 'diagnostic': self.diagnostic, 'debug_copy_ready': bool(self.recovery.prepared)} + + def recovery_plan(self, account): + self.require_mac() + with self.lock: + if self.job and self.job['status'] == 'running': + raise ValueError('请等待当前接入操作完成') + path = self.accounts.get(account) + if path is None: + raise ValueError('请先扫描并选择要备份的账号') + return self.recovery.plan(path) def describe(self, key, path): files = self.files(path) @@ -125,8 +139,14 @@ class MacConnection: raise ValueError('已有接入任务正在运行') if action == 'import' and body.get('account') not in self.accounts: raise ValueError('请先扫描并选择账号') - if action not in ('capture', 'import', 'dependencies'): + if action not in ('capture', 'import', 'dependencies', 'permissions', 'debug_copy', 'launch_copy'): raise ValueError('未知接入操作') + body = dict(body) + if action == 'debug_copy': + account = self.accounts.get(body.get('account')) + if account is None: + raise ValueError('请选择已扫描的账号') + body['_plan'] = self.recovery.consume(body.get('plan_id'), body.get('consent'), account) job_id = secrets.token_hex(12) self.job = {'id': job_id, 'action': action, 'status': 'running', 'stage': '准备中', 'error': ''} path = self.accounts.get(body.get('account')) @@ -152,13 +172,25 @@ class MacConnection: raise ValueError('管理员授权已取消或系统拒绝访问；没有修改微信签名或系统设置') payload = json.loads(result.stdout) if payload.get('error'): + if isinstance(payload.get('diagnostic'), dict): + raise AccessDenied(payload['diagnostic']) raise ValueError(payload['error']) values = payload.get('keys', []) else: values = scan_keys() with self.lock: self.keys = parse_keys(values) + self.diagnostic = None result = {'message': '候选密钥已保存在运行内存。请查看账号的匹配数量；退出微信后点击备份并导入。'} + elif action == 'permissions': + self.stage('正在检查微信的签名与调试权限，不读取消息正文') + self.diagnostic = permission_diagnostic() + result = {'message': self.diagnostic['summary']} + elif action == 'debug_copy': + result = self.recovery.prepare(body['_plan'], self.stage) + elif action == 'launch_copy': + result = self.recovery.launch() + self.diagnostic = None elif action == 'dependencies': self.stage('从 PyPI 安装压缩文本支持；不会发送聊天或密钥') target = self.data/'mac_deps' @@ -178,6 +210,9 @@ class MacConnection: message = str(exc) if isinstance(exc, ValueError) else '接入失败，请检查文件访问权限、磁盘空间与数据格式。已有记录未被覆盖。' with self.lock: self.job.update(status='failed', error=message, stage='需要处理') + if isinstance(exc, AccessDenied): + self.diagnostic = exc.diagnostic + self.job['error_code'] = exc.diagnostic.get('code') def import_account(self, source, keys, biz): if wechat_pid(): diff --git a/wechat_tool/mac_keys.py b/wechat_tool/mac_keys.py index 19a06e7..7f54f98 100644 --- a/wechat_tool/mac_keys.py +++ b/wechat_tool/mac_keys.py @@ -1,16 +1,18 @@ """Bounded, read-only scan of the local WeChat process. No system changes.""" import ctypes as C import json +import os import platform import re import subprocess import time -def wechat_pid(): +def wechat_processes(): if platform.system() != 'Darwin': - return None + return [] result = subprocess.run(['/usr/bin/pgrep', '-x', 'WeChat'], capture_output=True, text=True, timeout=5) + processes = [] for value in result.stdout.split(): pid = int(value) lib = C.CDLL('/usr/lib/libproc.dylib') @@ -19,22 +21,66 @@ def wechat_pid(): if lib.proc_pidpath(pid, buf, len(buf)) > 0: path = buf.value.decode('utf-8', errors='replace') if path.endswith('/WeChat.app/Contents/MacOS/WeChat'): - return pid - return None + processes.append({'pid': pid, 'path': path}) + return processes + + +def wechat_pid(): + processes = wechat_processes() + return processes[0]['pid'] if processes else None + + +def signing_flags(pid): + lib = C.CDLL('/usr/lib/libSystem.B.dylib', use_errno=True) + lib.csops.argtypes = [C.c_int, C.c_uint, C.c_void_p, C.c_size_t] + value = C.c_uint32() + if lib.csops(pid, 0, C.byref(value), C.sizeof(value)): + return {'hardened_runtime': None, 'get_task_allow': None} + return {'hardened_runtime': bool(value.value & 0x10000), 'get_task_allow': bool(value.value & 4)} + + +def permission_diagnostic(code=None, pid=None): + processes = wechat_processes() + target = next((p for p in processes if p['pid'] == pid), processes[0] if len(processes) == 1 else None) + result = {'platform': platform.system(), 'architecture': platform.machine(), + 'administrator': hasattr(os, 'geteuid') and os.geteuid() == 0, + 'task_for_pid_code': code, 'processes': processes, 'code': 'no_process', + 'summary': '请先打开并登录微信，再检查读取权限。'} + if len(processes) > 1: + result.update(code='multiple_processes', summary='检测到多个微信进程，请仅保留要读取的一个微信。') + elif target: + result.update(signing_flags(target['pid'])) + if result['hardened_runtime'] and result['get_task_allow'] is False: + result.update(code='target_restricted', summary='微信启用了强化运行时，且没有允许调试。普通和管理员读取都可能被签名策略拒绝；反复输入密码不能改变此限制。') + elif code is not None: + result.update(code='access_denied', summary='macOS 拒绝读取进程，请检查系统“隐私与安全性 → 开发者工具”中的终端授权。权限设置不保证能突破目标签名限制。') + else: + result.update(code='can_attempt', summary='未检测到明确的目标签名阻塞，可以尝试读取；实际结果以系统返回为准。') + return result + + +class AccessDenied(ValueError): + def __init__(self, diagnostic): + self.diagnostic = diagnostic + super().__init__(diagnostic['summary']) def scan_keys(): if platform.system() != 'Darwin': raise ValueError('请在安装微信的 Mac 上运行本地服务') - pid = wechat_pid() + processes = wechat_processes() + if len(processes) > 1: + raise AccessDenied(permission_diagnostic()) + pid = processes[0]['pid'] if processes else None if not pid: raise ValueError('未找到运行中的微信，请先打开并登录微信') lib = C.CDLL('/usr/lib/libSystem.B.dylib') own = C.c_uint.in_dll(lib, "mach_task_self_").value task = C.c_uint() lib.task_for_pid.argtypes = [C.c_uint, C.c_int, C.POINTER(C.c_uint)] - if lib.task_for_pid(own, pid, C.byref(task)): - raise ValueError('系统拒绝读取微信进程。管理员授权也可能受微信签名限制；工具不会修改签名或关闭 SIP。可导入本地密钥文件继续。') + code = lib.task_for_pid(own, pid, C.byref(task)) + if code: + raise AccessDenied(permission_diagnostic(code, pid)) class Region(C.Structure): _pack_ = 4 _fields_ = [('protection', C.c_int), ('max_protection', C.c_int), ('inheritance', C.c_int), @@ -83,4 +129,7 @@ if __name__ == '__main__': try: print(json.dumps({'keys': scan_keys()})) except Exception as exc: - print(json.dumps({'error': str(exc) if isinstance(exc, ValueError) else '本机密钥扫描失败'})) + payload = {'error': str(exc) if isinstance(exc, ValueError) else '本机密钥扫描失败'} + if isinstance(exc, AccessDenied): + payload['diagnostic'] = exc.diagnostic + print(json.dumps(payload)) diff --git a/wechat_tool/mac_recovery.py b/wechat_tool/mac_recovery.py new file mode 100644 index 0000000..3b22f54 --- /dev/null +++ b/wechat_tool/mac_recovery.py @@ -0,0 +1,172 @@ +"""Opt-in debug copy. Never re-sign the installed app or change system policy.""" +import hashlib +import json +import plistlib +import secrets +import shutil +import subprocess +import time +from pathlib import Path + +from .mac_keys import wechat_pid + + +def run(args, timeout=90): + result = subprocess.run(args, capture_output=True, timeout=timeout) + if result.returncode: + raise ValueError('系统操作未成功：'+Path(args[0]).name+'。请检查应用完整性、文件访问权限或系统提示；原版应用没有被重签。') + return result + + +def entitlements(app): + result = run(['/usr/bin/codesign', '-d', '--entitlements', ':-', str(app)]) + for data in (result.stdout, result.stderr): + start = data.find(b'') + if start >= 0 and end >= start: + value = plistlib.loads(data[start:end+8]) + if not isinstance(value, dict): + raise ValueError('应用权限文件格式错误') + return value + if data.startswith(b'bplist'): + return plistlib.loads(data) + return {} + + +def fingerprint(app): + digest = hashlib.sha256() + for name in ('Contents/Info.plist', 'Contents/MacOS/WeChat', 'Contents/_CodeSignature/CodeResources'): + path = app/name + if path.is_file(): + with path.open('rb') as file: + while chunk := file.read(1024**2): + digest.update(chunk) + return digest.hexdigest() + + +def inventory(root): + result = {} + for path in root.rglob('*'): + if path.is_symlink(): + raise ValueError('账号数据包含符号链接，无法创建独立的完整备份，已停止') + if path.is_file(): + stat = path.stat() + result[str(path.relative_to(root))] = [stat.st_size, stat.st_mtime_ns] + return result + + +def safe_app_links(app): + for link in app.rglob('*'): + if link.is_symlink() and (link.readlink().is_absolute() or not link.resolve().is_relative_to(app)): + raise ValueError('应用包含绝对路径或越界的符号链接，无法确保仅修改副本，已停止') + + +class DebugRecovery: + def __init__(self, data): + self.data = Path(data).resolve() + self.plans = {} + self.prepared = None + + def plan(self, account, app=None): + candidates = [Path('/Applications/WeChat.app'), Path.home()/'Applications/WeChat.app'] + app = app or next((p for p in candidates if p.is_dir()), None) + if not app: + raise ValueError('未找到已安装的原版 WeChat.app') + app = app.resolve() + if self.data.is_relative_to(app) or app.is_relative_to(self.data): + raise ValueError('工具数据目录不能与原版微信应用目录重叠') + safe_app_links(app) + info = plistlib.loads((app/'Contents/Info.plist').read_bytes()) + if info.get('CFBundleIdentifier') != 'com.tencent.xinWeChat' or info.get('CFBundleExecutable') != 'WeChat': + raise ValueError('应用标识不符合预期，已停止') + run(['/usr/bin/codesign', '--verify', '--deep', '--strict', str(app)]) + token = secrets.token_hex(16) + root = self.data/'mac_debug'/token + # Copy the entire account plus login configuration before any clone launch. + sources = [Path(account).parent] + all_users = Path(account).parent.parent/'all_users' + if all_users.is_dir(): + sources.append(all_users) + size = sum(sum(v[0] for v in inventory(source).values()) for source in sources) + app_size = sum(p.stat().st_size for p in app.rglob('*') if p.is_file() and not p.is_symlink()) + plan = {'id': token, 'created': time.monotonic(), 'app': app, 'account': Path(account), + 'sources': sources, 'root': root, 'fingerprint': fingerprint(app), + 'estimated_bytes': size+app_size+512*1024**2, 'version': str(info.get('CFBundleShortVersionString', '未知'))} + self.plans = {token: plan} + return {'plan_id': token, 'source': str(app), 'copy': str(root/'WeChat.app'), + 'backup': str(root/'backup'), 'version': plan['version'], 'estimated_bytes': plan['estimated_bytes'], + 'effects': ['先完整备份所选账号目录与本机登录配置，可能包含图片、附件，所需空间比只备份数据库更多。', + '仅对工具目录里的临时应用副本重新签名并允许调试；不改动已安装的原版微信，不关闭 SIP。', + '副本保留原应用标识，登录时可能使用或更新同一账号的本地数据，可能要求重新登录或系统授权。', + '有调试权限的副本更容易被本机调试工具读取。仅用于本人账号取钥，用完退出副本并恢复原版。', + '副本能否启动、登录和取钥仍需本机验证；失败不会自动重试、删除备份或修改原版。']} + + def consume(self, token, consent, account): + if consent is not True: + raise ValueError('请先阅读操作影响并明确勾选同意，才能创建调试副本') + plan = self.plans.pop(token, None) + if not plan or time.monotonic()-plan['created'] > 1800: + raise ValueError('副本操作预览已过期或已使用，请重新预览') + if Path(account) != plan['account']: + raise ValueError('账号发生变化，请重新预览') + return plan + + def prepare(self, plan, stage): + if wechat_pid(): + raise ValueError('请先正常退出所有微信，再创建临时副本；知微保持运行') + if fingerprint(plan['app']) != plan['fingerprint']: + raise ValueError('原版微信在预览后发生变化，请重新预览') + safe_app_links(plan['app']) + run(['/usr/bin/codesign', '--verify', '--deep', '--strict', str(plan['app'])]) + self.data.mkdir(parents=True, exist_ok=True, mode=0o700) + if shutil.disk_usage(self.data).free < plan['estimated_bytes']: + raise ValueError('磁盘空间不足以完整备份账号并创建应用副本') + root = plan['root'] + if not root.resolve().is_relative_to(self.data) or root.exists(): + raise ValueError('副本目录无效或已存在，已停止') + root.mkdir(parents=True, mode=0o700) + stage('正在完整备份账号数据和登录配置；请保持微信退出') + manifest = [] + for index, source in enumerate(plan['sources']): + before = inventory(source) + destination = root/'backup'/f'{index}-{source.name}' + shutil.copytree(source, destination) + copied = inventory(destination) + if {name:value[0] for name,value in copied.items()} != {name:value[0] for name,value in before.items()}: + raise ValueError('备份文件清单或大小不完整，已停止，未创建调试副本') + if inventory(source) != before or wechat_pid(): + raise ValueError('备份期间账号数据发生变化，已停止；请退出微信后重新预览') + manifest.append({'source': str(source), 'backup': str(destination), 'files': len(before)}) + (root/'backup-manifest.json').write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding='utf-8') + stage('正在创建临时应用副本；原版安装文件不修改') + copy = root/'WeChat.app' + shutil.copytree(plan['app'], copy, symlinks=True) + safe_app_links(copy) + if fingerprint(copy) != plan['fingerprint'] or fingerprint(plan['app']) != plan['fingerprint']: + raise ValueError('应用复制前后不一致，请重新预览') + grants = entitlements(plan['app']) + grants['com.apple.security.get-task-allow'] = True + permissions = root/'debug-entitlements.plist' + permissions.write_bytes(plistlib.dumps(grants)) + stage('仅为临时副本添加调试权限并校验签名') + run(['/usr/bin/codesign', '--force', '--deep', '--options', 'runtime', '--timestamp=none', + '--sign', '-', '--entitlements', str(permissions), str(copy)], timeout=180) + run(['/usr/bin/codesign', '--verify', '--deep', '--strict', str(copy)]) + if entitlements(copy).get('com.apple.security.get-task-allow') is not True: + raise ValueError('临时副本未取得预期调试权限，已停止') + if fingerprint(plan['app']) != plan['fingerprint']: + raise ValueError('原版应用在操作期间发生变化，请先检查原版状态') + self.prepared = {'app': copy, 'fingerprint': fingerprint(copy), 'root': root, 'original': plan['app']} + return {'message': '备份与临时副本已准备。下一步点击“启动临时副本”，在副本中登录并打开聊天，再读取密钥。', + 'backup': str(root/'backup'), 'copy': str(copy), 'original_unchanged': True} + + def launch(self): + if not self.prepared: + raise ValueError('请先预览并创建临时副本') + if wechat_pid(): + raise ValueError('请先退出正在运行的微信，再启动临时副本') + copy = self.prepared['app'] + if not copy.resolve().is_relative_to(self.data) or fingerprint(copy) != self.prepared['fingerprint']: + raise ValueError('临时副本已变化，拒绝启动，请重新准备') + run(['/usr/bin/open', '-n', str(copy)]) + return {'message': '已向 macOS 请求启动临时副本。请按系统提示完成打开/登录，进入聊天后再点击读取密钥；未自动绕过系统拦截。'} diff --git a/wechat_tool/server.py b/wechat_tool/server.py index 13745f2..cf085f4 100644 --- a/wechat_tool/server.py +++ b/wechat_tool/server.py @@ -301,6 +301,8 @@ class Handler(BaseHTTPRequestHandler): return self.respond(self.server.mac.load_keys(body.get('keys'))) if path == '/api/mac/start': return self.respond(self.server.mac.start(body.get('action'), body)) + if path == '/api/mac/recovery-plan': + return self.respond(self.server.mac.recovery_plan(body.get('account'))) if path == '/api/import': data = base64.b64decode(body['data'], validate=True) options = body.get('options') or {} diff --git a/wechat_tool/static/app.js b/wechat_tool/static/app.js index 25e076b..d92c13f 100644 --- a/wechat_tool/static/app.js +++ b/wechat_tool/static/app.js @@ -13,7 +13,7 @@ async function api(path, body) { const opts=body===undefined?{}:{method:'POST',headers:{'Content-Type':'application/json','X-Workbench-Token':token},body:JSON.stringify(body)}; const response=await fetch(path,opts); const data=await response.json(); if(!response.ok)throw Error(data.error||'请求失败'); return data; } -async function busy(button, work) {button.disabled=true;try{await work();}catch(e){notify(e.message,true);}finally{button.disabled=button.id.startsWith("mac-")&&button.id!=="mac-refresh"&&macRunning;}} +async function busy(button, work) {button.disabled=true;try{await work();}catch(e){notify(e.message,true);}finally{button.disabled=button.id.startsWith('mac-')&&button.id!=='mac-refresh'&&(macRunning||!macAvailable);if(button.id==='mac-create-copy')button.disabled=button.disabled||!macRecoveryPlan||!$('#mac-copy-consent').checked;if(button.id==='mac-launch-copy')button.disabled=button.disabled||!macCopyReady;}} function page(name) {if(!titles[name])name='overview';$$('.page').forEach(n=>n.hidden=n.id!=='page-'+name);$$('nav button').forEach(n=>n.classList.toggle('active',n.dataset.page===name));$('#page-title').textContent=titles[name];location.hash=name;} function displayTime(time) { if(!time)return '—'; const offset=settings.utc_offset||'+08:00'; const sign=offset[0]==='-'?-1:1; @@ -119,14 +119,17 @@ $('#close-detail').addEventListener('click',()=>$('#detail-dialog').close()); $('#merge-cards').addEventListener('click',()=>{const ids=$$('.card-select:checked').map(n=>n.value);if(ids.length<2)return notify('请至少勾选两张主题卡片。',true);$('#merge-title').value=currentRun.cards.find(c=>c.id===ids[0]).title;$('#merge-dialog').showModal();}); $('#cancel-merge').addEventListener('click',()=>$('#merge-dialog').close()); $('#confirm-merge').addEventListener('click',e=>busy(e.target,async()=>{await api('/api/card',{run_id:currentRun.id,ids:$$('.card-select:checked').map(n=>n.value),action:'merge',title:$('#merge-title').value.trim()});await loadRun(currentRun.id);$('#merge-dialog').close();notify('已合并主题，原摘要保留在备注中，请重新审核。');})); -let macTimer = null, lastMacJob = null, macRunning = false; +let macTimer = null, lastMacJob = null, macRunning = false, macAvailable = false, macRecoveryPlan = null, macCopyReady = false; async function macStatus(){ clearTimeout(macTimer); const state=await api('/api/mac/status'); $('#mac-platform').textContent=state.available?`本地服务运行于 Mac · ${state.architecture}。扫描和读取只操作这台 Mac。`:'当前服务运行于非 Mac 环境。请把工具复制到安装微信的 Mac 上启动，再在那台 Mac 的浏览器打开此页面。'; const running=state.job?.status==='running'; macRunning=running; - for(const id of ['mac-scan','mac-folder','mac-capture','mac-capture-admin','mac-dependencies','mac-import'])$('#'+id).disabled=!state.available||running; + macAvailable=state.available;macCopyReady=state.debug_copy_ready; + for(const id of ['mac-scan','mac-folder','mac-capture','mac-capture-admin','mac-dependencies','mac-import','mac-permissions','mac-preview-copy'])$('#'+id).disabled=!state.available||running; + $('#mac-launch-copy').disabled=!state.available||running||!state.debug_copy_ready; + $('#mac-create-copy').disabled=!state.available||running||!macRecoveryPlan||!$('#mac-copy-consent').checked; $('#mac-key-file').disabled=running; const select=$('#mac-account'),previous=select.value;select.replaceChildren();$('#mac-accounts').replaceChildren(); for(const account of state.accounts){ @@ -136,6 +139,8 @@ async function macStatus(){ if(!state.accounts.length){const option=el('option','请先扫描或选择数据文件夹');option.value='';select.append(option);} $('#mac-key-state').textContent=state.keys_ready?'候选密钥已在运行内存。账号卡上的匹配数量决定实际可读范围。重启知微后需重新读取。':'尚未读取或导入密钥；未加密的可读数据库可直接进入第 3 步。'; const job=state.job;$('#mac-job-panel').hidden=!job; + const diagnostic=state.diagnostic;$('#mac-diagnostic-panel').hidden=!diagnostic; + if(diagnostic){$('#mac-diagnostic-summary').textContent=diagnostic.summary;const yes=value=>value===true?'是':value===false?'否':'未知';$('#mac-diagnostic-details').textContent=[`诊断类型：${diagnostic.code}`,`强化运行时：${yes(diagnostic.hardened_runtime)}`,`允许调试：${yes(diagnostic.get_task_allow)}`,`管理员进程：${yes(diagnostic.administrator)}`,`进程读取返回码：${diagnostic.task_for_pid_code??'尚未尝试'}`].join('\n');} if(job){$('#mac-job-stage').textContent=job.stage;$('#mac-job-error').hidden=job.status!=='failed';$('#mac-job-error').textContent=job.error||'';$('#mac-job-result').hidden=!job.result;$('#mac-job-result').textContent=job.result?JSON.stringify(job.result,null,2):''; if(job.status==='complete'&&lastMacJob!==job.id){lastMacJob=job.id;await refresh();notify(job.result?.message||'接入操作已完成');} if(running)macTimer=setTimeout(()=>macStatus().catch(e=>notify(e.message,true)),1800); @@ -148,6 +153,12 @@ $('#mac-folder').addEventListener('click',e=>busy(e.target,async()=>{await api(' $('#mac-refresh').addEventListener('click',e=>busy(e.target,macStatus)); $('#mac-capture').addEventListener('click',e=>busy(e.target,()=>macStart('capture'))); $('#mac-capture-admin').addEventListener('click',e=>busy(e.target,()=>macStart('capture',{administrator:true}))); +$('#mac-permissions').addEventListener('click',e=>busy(e.target,()=>macStart('permissions'))); +$('#mac-preview-copy').addEventListener('click',e=>busy(e.target,async()=>{const plan=await api('/api/mac/recovery-plan',{account:$('#mac-account').value});macRecoveryPlan=plan.plan_id;$('#mac-copy-consent').checked=false;$('#mac-create-copy').disabled=true;$('#mac-copy-plan').hidden=false;$('#mac-copy-paths').textContent=`原版：${plan.source}\n临时副本：${plan.copy}\n完整备份：${plan.backup}\n预计至少需要 ${(plan.estimated_bytes/1024**3).toFixed(2)} GiB 可用空间`;$('#mac-copy-effects').replaceChildren(...plan.effects.map(effect=>el('li',effect)));})); +$('#mac-copy-consent').addEventListener('change',()=>{$('#mac-create-copy').disabled=!macAvailable||macRunning||!macRecoveryPlan||!$('#mac-copy-consent').checked;}); +$('#mac-account').addEventListener('change',()=>{macRecoveryPlan=null;$('#mac-copy-plan').hidden=true;$('#mac-copy-consent').checked=false;$('#mac-create-copy').disabled=true;}); +$('#mac-create-copy').addEventListener('click',e=>busy(e.target,async()=>{if(!macRecoveryPlan||!$('#mac-copy-consent').checked)throw Error('请先预览并明确同意操作影响');const plan=macRecoveryPlan;macRecoveryPlan=null;$('#mac-copy-plan').hidden=true;await macStart('debug_copy',{account:$('#mac-account').value,plan_id:plan,consent:true});})); +$('#mac-launch-copy').addEventListener('click',e=>busy(e.target,()=>macStart('launch_copy'))); $('#mac-dependencies').addEventListener('click',e=>busy(e.target,()=>macStart('dependencies'))); $('#mac-import').addEventListener('click',e=>busy(e.target,()=>macStart('import',{account:$('#mac-account').value,include_biz:$('#mac-include-biz').checked}))); $('#mac-key-file').addEventListener('change',async e=>{try{const file=e.target.files[0];if(!file)return;if(file.size>1024**2)throw Error('密钥文件最多 1 MB');await api('/api/mac/keys',{keys:JSON.parse(await file.text())});await macStatus();notify('密钥文件已在本机内存处理，不会发送给模型。');}catch(error){notify(error.message,true);}finally{e.target.value='';}}); diff --git a/wechat_tool/static/index.html b/wechat_tool/static/index.html index 6194285..e8cde6f 100644 --- a/wechat_tool/static/index.html +++ b/wechat_tool/static/index.html @@ -2,9 +2,9 @@ 知微 · 微信知识需求分析 - +
-
WECHAT KNOWLEDGE WORKBENCH

看见对话中的知识需求

LOCAL FIRST · 本地优先
+

看见对话中的知识需求

本地优先
从真实问题出发

哪些问题，
值得被认真整理？

发现重复提问、风险线索与可复用经验。
每一个判断，都能回到原始对话。

@@ -17,7 +17,9 @@
正在检查本机环境…
01 找到账号02 本机读取密钥03 备份并导入

01 · 找到你的微信账号

自动查找微信 4.x 数据目录。自定义位置可通过系统文件夹选择器指定；旧版目录不会自动混入。

-

02 · 让本机读取消息

先打开并登录微信，在微信中打开你想分析的聊天，然后点击读取。候选密钥仅保存在本工具的运行内存，不会发送给模型、写入日志或仓库。

管理员按钮会弹出系统密码框，用于尝试读取微信进程；不会修改微信签名、关闭 SIP 或保存密码。系统保护仍可能拒绝访问，授权不保证成功。文本解压支持从 PyPI 安装到本工具的数据目录，不发送聊天。

已有本地密钥文件？选择文件继续

支持包含 key / key_hex 字段的 JSON；64 位 raw key 或 96 位 key+salt。微信登录密码不能用来代替。

+

02 · 让本机读取消息

先打开并登录微信，在微信中打开你想分析的聊天，然后点击读取。候选密钥仅保存在本工具的运行内存，不会发送给模型、写入日志或仓库。

管理员按钮会弹出系统密码框，用于尝试读取微信进程；不会修改原版微信签名、关闭 SIP 或保存密码。系统保护仍可能拒绝访问，授权不保证成功。文本解压支持从 PyPI 安装到本工具的数据目录，不发送聊天。

已有本地密钥文件？选择文件继续

支持包含 key / key_hex 字段的 JSON；64 位 raw key 或 96 位 key+salt。微信登录密码不能用来代替。

+ +

读取被签名限制时：使用临时调试副本

先预览备份位置与操作影响。只有你明确同意后，工具才会备份账号数据，并对独立副本添加调试权限；原版安装文件保持不变。

副本启动后，按系统提示登录微信、打开目标聊天，再点击“读取本机密钥”；如仍被拒绝，可在页面主动尝试管理员读取。取钥后退出副本，再备份并导入。仅用于本人账号，取钥完成后恢复原版使用。

03 · 备份、解密、校验并导入

密钥匹配后，请正常退出微信，保持知微运行，再点击下面的按钮。原始数据库不会被修改，完整副本与可读库留在本机。

请预留至少约四倍所选数据库大小的磁盘空间。未知格式、缺失身份映射或认证失败时会停止，避免静默漏消息。

实验性接入：支持 SQLCipher 4 raw key / 4096 字节页 / SHA512 认证，以及部分微信 4.x 表结构。尚未在你的 Intel Mac 4.1.0.18 实机验收；页面会显示实际结果，不保证任何版本都能一键读取。

@@ -39,11 +41,11 @@

从问题到知识需求清单

选择一个时间段，覆盖全部会话，也可以聚焦指定群聊或私聊。

多选用 ⌘ / Ctrl；关键词筛选会剔除不匹配的上下文，完整讨论分析建议留空。规则按最多 30 分钟间隔、30 条消息收集讨论候选，延迟回复与交错话题需要人工检查。

本地模式无需 API，采用规则与文本相似度。模型模式按批提炼；跨批相似主题可在结果中人工合并。

- +
-
EVIDENCE & REVIEW

 +

合并选中主题

原有摘要会保留在备注，合并后的主题需要重新审核。