diff --git a/.gitignore b/.gitignore index 24eecf2..9df15d3 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # Python __pycache__/ +.ruff_cache/ *.py[cod] *$py.class *.egg-info/ diff --git a/README.md b/README.md index 1d877d2..2732186 100644 --- a/README.md +++ b/README.md @@ -83,21 +83,28 @@ Note: Make sure you have Node.js installed first. You can ask your agent to set ### Step 1 — Initialize -Make sure WeChat is running, then: +Start with a read-only check. It does not read WeChat process memory: ```bash -# macOS/Linux: may need sudo for memory scanning -sudo wechat-cli init +wechat-cli doctor +wechat-cli init --dry-run +``` + +`init` now shows a plan by default and does not automatically scan process memory. Prefer an audited helper whose key document is piped only over stdin: -# Windows: run in a terminal with sufficient privileges -wechat-cli init +```bash +trusted-wechat-key-helper | wechat-cli keys import --stdin ``` -This auto-detects your WeChat data directory, extracts encryption keys, and saves config to `~/.wechat-cli/`. +Every imported key must pass HMAC verification against its real database first page before state is saved under `~/.wechat-cli/`. A new config/key generation is fully written and read back before `active.json` is switched. Windows uses CurrentUser DPAPI, macOS uses the login Keychain, and Linux uses Secret Service; initialization fails closed when no system key store is available. + +Process-memory scanning is a high-risk compatibility path. It requires both `--scan-memory --confirm-memory-scan`, plus an exact accepted capability profile binding platform, client version, valid publisher signature, and scanner algorithm. Unknown clients stop before process-memory read access is requested. No scanner profile is accepted yet, so current builds will not perform a memory scan. + +When upgrading from a release with `~/.wechat-cli/all_keys.json`, run `wechat-cli init --force` to migrate. The CLI no longer silently accepts a legacy plaintext key file by default. ![init-claude-code-1](image/init-claude-code-1.png) -On macOS, you'll need to run the `sudo` command and enter your password: +On macOS/Linux, the scanner helper may ask for an elevation password: ![init-claude-code-2](image/init-claude-code-2.png) @@ -123,30 +130,7 @@ Without this permission, the tool cannot access WeChat's data directory and key On some macOS systems, `init` may fail with `task_for_pid failed` even when running with `sudo`. This is due to macOS security restrictions on process memory access. -**WeChat CLI will automatically attempt to fix this** by re-signing WeChat with the required entitlement (original entitlements are preserved). Just follow the on-screen instructions: - -1. The tool will re-sign WeChat automatically -2. Quit WeChat completely (not just minimize) -3. Reopen WeChat and log in -4. Run `sudo wechat-cli init` again - -If auto re-signing fails, you can do it manually: - -```bash -# Quit WeChat first, then: -sudo codesign --force --sign - --entitlements /dev/stdin /Applications/WeChat.app <<'EOF' - - - - - com.apple.security.get-task-allow - - - -EOF -``` - -> **Heads up:** Re-signing WeChat is safe and will **not** cause account issues or bans. However, it may affect WeChat's auto-update mechanism. If you notice any feature not working properly, or want to update WeChat to the latest version, simply re-download and reinstall WeChat from the [official website](https://mac.weixin.qq.com/) — no need to re-run `init`, your existing config and keys will continue to work. +WeChat CLI fails closed and leaves `WeChat.app` untouched. It never re-signs, injects into, or modifies the third-party app. Any debugging entitlement must be assessed and applied explicitly by the device administrator. The bundled arm64 scanner is a historical artifact and is rejected until rebuilt from the current C source with the redaction artifact gate; no x86_64 artifact is currently shipped. ### Step 2 — Use It @@ -337,8 +321,8 @@ The `--type` option (on `history` and `search`): | Platform | Status | Notes | |----------|--------|-------| -| macOS (Apple Silicon) | ✅ Supported | Bundled arm64 binary | -| macOS (Intel) | ✅ Supported | x86_64 binary needed | +| macOS (Apple Silicon) | ⚠️ Rebuild required | Historical arm64 artifact is rejected by the safety gate | +| macOS (Intel) | ❌ Not shipped | No x86_64 artifact is present | | Windows | ✅ Supported | Reads Weixin.exe process memory | | Linux | ✅ Supported | Reads /proc/pid/mem, requires root | @@ -348,10 +332,18 @@ The `--type` option (on `history` and `search`): WeChat stores chat data in SQLCipher-encrypted SQLite databases locally. WeChat CLI: -1. **Extracts keys** — scans WeChat process memory for encryption keys (`init`) -2. **Decrypts on-the-fly** — transparent page-level AES-256-CBC decryption with caching +1. **Obtains keys under policy** — imports audited-helper output through stdin and validates it against real database pages; scanning is limited to accepted clients +2. **Decrypts on-the-fly** — transparent page-level AES-256-CBC decryption in a private per-process cache removed on normal exit 3. **Queries locally** — all data stays on your machine, no network access +### Key and cache security boundary + +- Initialization is stored under `~/.wechat-cli/generations//`; `active.json` changes only after config and key readback succeeds. +- `doctor`, `init --dry-run`, and default `init` never read WeChat process memory; unknown clients fail closed before process-memory read access. +- Initialization and scanner logs report matching state, database paths, and counts without echoing complete keys. +- Windows uses CurrentUser DPAPI and explicit NTFS ACLs; macOS uses the login Keychain; Linux uses Secret Service. The default is fail-closed when no system key store exists. `WECHAT_CLI_ALLOW_PLAINTEXT_KEYS=1` explicitly enables the compatibility plaintext file backend. +- Decryption uses a stable DB/WAL snapshot and verifies SQLCipher page HMACs, SQLite WAL checksums, and the last commit frame. Plaintext cache directories hold an exclusive lease; startup removes only well-formed stale runs that are confirmed unlocked. + --- ## 📄 License diff --git a/README_CN.md b/README_CN.md index 6483703..a2174fc 100644 --- a/README_CN.md +++ b/README_CN.md @@ -81,21 +81,28 @@ pip install -e . ### 第一步 — 初始化 -确保微信正在运行，然后： +先执行只读检查；它不会读取微信进程内存： ```bash -# macOS/Linux: 可能需要 sudo 权限 -sudo wechat-cli init +wechat-cli doctor +wechat-cli init --dry-run +``` + +`init` 默认也只显示计划，不再自动扫描进程内存。推荐由经过审计的独立 helper 生成密钥文档，并仅通过标准输入导入： -# Windows: 在有足够权限的终端中运行 -wechat-cli init +```bash +trusted-wechat-key-helper | wechat-cli keys import --stdin ``` -这一步会自动检测微信数据目录、提取加密密钥，并保存到 `~/.wechat-cli/`。 +导入时每个密钥都必须通过对应真实数据库首页的 HMAC 校验，之后才会保存到 `~/.wechat-cli/`。新配置与密钥先写入独立 generation，完整读回后才原子切换。Windows 使用 CurrentUser DPAPI，macOS 使用登录钥匙串，Linux 使用 Secret Service；系统密钥库不可用时默认拒绝写入明文密钥。 + +内存扫描属于高风险兼容路径，必须同时使用 `--scan-memory --confirm-memory-scan`，并且客户端平台、精确版本、有效发布者签名和扫描算法必须存在于已验收能力清单；未知版本会在申请进程内存读取权限前停止。目前能力清单没有已验收版本，因此不会执行内存扫描。 + +从旧版本升级且已有 `~/.wechat-cli/all_keys.json` 时，执行 `wechat-cli init --force` 迁移；默认不会继续静默使用旧版明文密钥文件。 ![init-claude-code-1](image/init-claude-code-1.png) -如果是 mac，需要执行 sudo 命令，然后需要输入密码： +macOS/Linux 扫描 helper 请求提权时，系统可能要求输入密码： ![init-claude-code-code-2](image/init-claude-code-2.png) @@ -121,30 +128,7 @@ wechat-cli init 在某些 macOS 系统上，即使使用了 `sudo`，`init` 也可能报 `task_for_pid failed`。这是 macOS 的安全策略限制了进程内存访问。 -**WeChat CLI 会自动尝试修复此问题**——对微信重新签名以获取必要权限（会保留微信原有权限）。按提示操作即可： - -1. 工具会自动对微信重新签名 -2. 完全退出微信（不是最小化） -3. 重新打开微信并登录 -4. 再次执行 `sudo wechat-cli init` - -如果自动签名失败，可以手动执行： - -```bash -# 先退出微信，然后： -sudo codesign --force --sign - --entitlements /dev/stdin /Applications/WeChat.app <<'EOF' - - - - - com.apple.security.get-task-allow - - - -EOF -``` - -> **温馨提示：** 重新签名是安全的，**不会**导致封号或账号异常。但可能影响微信的部分功能或自动更新。如果发现任何功能异常（如搜一搜无法使用），或想更新到微信最新版，直接从[微信官网](https://mac.weixin.qq.com/)重新下载安装即可，**无需重新执行 init**，已有的配置和密钥不受影响。 +WeChat CLI 会失败关闭并保留原始 `WeChat.app`，不会自动重签、注入或修改第三方应用。调试权限如确有必要，必须由设备管理员独立评估并显式操作。当前仓库内置的 arm64 扫描器是历史产物；在 macOS 上从当前 C 源码重建并通过脱敏产物门禁前，CLI 会拒绝执行它。仓库目前没有 x86_64 产物。 ### 第二步 — 开始使用 @@ -335,10 +319,10 @@ wechat-cli new-messages # 后续: 仅返回上次以来的新 | 平台 | 状态 | 说明 | |------|------|------| -| macOS (Apple Silicon) | ✅ 支持 | 内置 arm64 二进制 | -| macOS (Intel) | ✅ 支持 | 需要 x86_64 二进制 | -| Windows | ✅ 支持 | 读取 Weixin.exe 进程内存 | -| Linux | ✅ 支持 | 读取 /proc/pid/mem，需要 root | +| macOS (Apple Silicon) | ⚠️ 需重建 | 历史 arm64 产物被安全门禁拒绝，须从当前源码重建 | +| macOS (Intel) | ❌ 未提供 | 仓库没有 x86_64 二进制 | +| Windows | ⚠️ 查询/导入可用 | 内存扫描须命中已验收的精确版本能力清单 | +| Linux | ⚠️ 查询/导入可用 | 内存扫描须命中已验收的精确版本能力清单 | --- @@ -346,10 +330,18 @@ wechat-cli new-messages # 后续: 仅返回上次以来的新 微信将聊天数据存储在本地的 SQLCipher 加密 SQLite 数据库中。WeChat CLI： -1. **提取密钥** — 扫描微信进程内存获取加密密钥（`init`） -2. **即时解密** — 透明页级 AES-256-CBC 解密，带缓存 +1. **受控取得密钥** — 默认使用受信 helper 标准输入导入并以真实数据库页校验；内存扫描仅限已验收版本 +2. **即时解密** — 透明页级 AES-256-CBC 解密；解密缓存只存于本次进程的私有目录，正常退出时删除 3. **本地查询** — 所有数据留在本机，无需网络访问 +### 密钥与缓存安全边界 + +- 初始化结果保存在 `~/.wechat-cli/generations//`，`active.json` 只在配置与密钥全部读回成功后切换。 +- `doctor`、`init --dry-run` 和默认 `init` 不会读取微信进程内存；未知客户端版本在申请进程内存读取权限前失败关闭。 +- 初始化和扫描日志只显示匹配状态、数据库路径和密钥数量，不回显完整密钥。 +- Windows 使用 CurrentUser DPAPI 并显式收紧 NTFS ACL；macOS 使用登录钥匙串；Linux 使用 Secret Service。若系统密钥库不可用，默认失败关闭。仅可通过 `WECHAT_CLI_ALLOW_PLAINTEXT_KEYS=1` 显式启用兼容明文文件后端。 +- 解密前先复制稳定的 DB/WAL 快照；SQLCipher 页 HMAC、SQLite WAL checksum 与最后提交帧均通过后才应用。明文缓存位于带独占租约的 `cache/run-*`；启动时只清理确认无人持锁的规范遗留目录。 + --- ## 📄 开源协议 diff --git a/npm/scripts/build.py b/npm/scripts/build.py index 0e82932..f724648 100644 --- a/npm/scripts/build.py +++ b/npm/scripts/build.py @@ -2,7 +2,6 @@ """Build wechat-cli standalone binaries with PyInstaller.""" import os -import shutil import subprocess import sys from pathlib import Path @@ -20,6 +19,37 @@ } +def current_platform(): + import platform as host + + system = host.system().lower() + machine = host.machine().lower() + if system == "darwin": + return "darwin-arm64" if machine == "arm64" else "darwin-x64" + if system == "windows" and machine in {"amd64", "x86_64"}: + return "win32-x64" + if system == "linux": + return "linux-arm64" if machine in {"arm64", "aarch64"} else "linux-x64" + raise RuntimeError(f"Unsupported build host: {system}-{machine}") + + +def build_macos_scanner(platform): + source = ROOT / "wechat_cli" / "bin" / "find_all_keys_macos.c" + arch = "arm64" if platform == "darwin-arm64" else "x86_64" + output_dir = ROOT / "build" / "native" / platform + output_dir.mkdir(parents=True, exist_ok=True) + output = output_dir / f"find_all_keys_macos.{arch}" + subprocess.check_call([ + "/usr/bin/clang", "-O2", "-arch", arch, str(source), + "-framework", "Foundation", "-o", str(output), + ]) + payload = output.read_bytes() + if b"[REDACTED]" not in payload or b"%-25s %-66s %s" in payload: + output.unlink(missing_ok=True) + raise RuntimeError("macOS scanner failed the redaction artifact gate") + return output + + def ensure_pyinstaller(): try: import PyInstaller # noqa: F401 @@ -31,7 +61,10 @@ def ensure_pyinstaller(): def build_platform(platform: str): - info = PLATFORM_MAP[platform] + host_platform = current_platform() + if platform != host_platform: + print(f"[-] Refusing cross-labelled build: host={host_platform}, target={platform}") + return False os_name, arch = platform.split("-") ext = ".exe" if os_name == "win32" else "" binary_name = f"wechat-cli{ext}" @@ -54,12 +87,14 @@ def build_platform(platform: str): "--clean", ] - # Bundle C binaries for key extraction - bin_dir = ROOT / "wechat_cli" / "bin" - if bin_dir.exists(): - for f in bin_dir.iterdir(): - if not f.name.startswith(".") and f.is_file(): - cmd.extend(["--add-binary", f"{f}:wechat_cli/bin"]) + # macOS scanner must be rebuilt from the current source on the matching host. + if os_name == "darwin": + try: + scanner = build_macos_scanner(platform) + except (OSError, subprocess.CalledProcessError, RuntimeError) as error: + print(f"[-] macOS scanner build/gate failed: {error}") + return False + cmd.extend(["--add-binary", f"{scanner}:wechat_cli/bin"]) # Hidden imports hidden = ["pysqlcipher3", "sqlcipher3", "Cryptodome", "zstandard"] @@ -91,29 +126,23 @@ def main(): platforms = sys.argv[1:] else: # Default: build for current platform only - import platform as _pf - current = f"{_pf.system().lower()}-{_pf.machine()}" - # Normalize - if current == "darwin-arm64": - platforms = ["darwin-arm64"] - elif current == "darwin-x86_64" or current == "darwin-amd64": - platforms = ["darwin-x64"] - else: - # Try to match - platforms = [] - for p in PLATFORM_MAP: - os_name, arch = p.split("-") - if os_name in current and (arch in current or - (arch == "x64" and ("x86_64" in current or "amd64" in current))): - platforms = [p] - break - if not platforms: - print(f"Cannot determine platform from '{current}'") - print(f"Usage: {sys.argv[0]} [platform...]") - print(f" Platforms: {', '.join(PLATFORM_MAP.keys())}") - sys.exit(1) + try: + platforms = [current_platform()] + except RuntimeError as error: + print(error) + print(f"Usage: {sys.argv[0]} [platform...]") + print(f" Platforms: {', '.join(PLATFORM_MAP.keys())}") + sys.exit(1) print(f"[+] Building for: {', '.join(platforms)}") + unknown = [name for name in platforms if name not in PLATFORM_MAP] + if unknown: + print(f"[-] Unknown platform(s): {', '.join(unknown)}") + sys.exit(1) + host_platform = current_platform() + if any(name != host_platform for name in platforms): + print(f"[-] Refusing cross-labelled build: host={host_platform}, targets={platforms}") + sys.exit(1) ensure_pyinstaller() results = {} diff --git a/tests/test_security.py b/tests/test_security.py new file mode 100644 index 0000000..970185f --- /dev/null +++ b/tests/test_security.py @@ -0,0 +1,608 @@ +import hashlib +import hmac +import json +import os +import struct +import sys +import tempfile +import unittest +from contextlib import redirect_stdout +from io import StringIO +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +from click.testing import CliRunner +from Crypto.Cipher import AES + +from wechat_cli.commands.doctor import build_doctor_report +from wechat_cli.commands.favorites import _parse_fav_content +from wechat_cli.commands.init import init as init_command +from wechat_cli.commands.keys import keys as keys_command +from wechat_cli.core.config import load_config +from wechat_cli.core.crypto import ( + PAGE_SZ, + RESERVE_SZ, + _derive_hmac_key, + _wal_checksum, + decrypt_page, + decrypt_wal, +) +from wechat_cli.core.db_cache import DBCache +from wechat_cli.core.key_document import ( + DpapiDatabaseKeyProvider, + JsonDatabaseKeyProvider, + MacKeychainDatabaseKeyProvider, + SecretServiceDatabaseKeyProvider, + create_key_provider, + validate_key_document, +) +from wechat_cli.core.key_utils import _is_safe_rel_path +from wechat_cli.core.secure_files import ( + assert_private_file, + atomic_write_bytes, + atomic_write_json, + ensure_private_directory, + read_private_json, + redact_sensitive_text, +) +from wechat_cli.core.state import commit_generation, resolve_active_generation +from wechat_cli.keys.capabilities import ScannerProfile, evaluate_scan_capability +from wechat_cli.keys.common import save_results +from wechat_cli.keys.scanner_macos import extract_keys as extract_macos_keys +from wechat_cli.keys.trusted_import import verify_imported_key_document + +KEY = "ab" * 32 +SALT = "cd" * 16 +_MEMORY_READ_GUARD = None + + +def setUpModule(): + """安全回归禁止触达真实进程内存，错误门禁应在这里立即失败。""" + global _MEMORY_READ_GUARD + if os.name == "nt": + _MEMORY_READ_GUARD = patch( + "wechat_cli.keys.scanner_windows._read_mem", + side_effect=AssertionError("tests must never read real process memory"), + ) + _MEMORY_READ_GUARD.start() + + +def tearDownModule(): + if _MEMORY_READ_GUARD is not None: + _MEMORY_READ_GUARD.stop() + + +def _encrypted_page(enc_key, salt, pgno, plaintext_byte): + iv = bytes([pgno]) * 16 + start = 16 if pgno == 1 else 0 + plaintext = bytes([plaintext_byte]) * (PAGE_SZ - RESERVE_SZ - start) + encrypted = AES.new(enc_key, AES.MODE_CBC, iv).encrypt(plaintext) + prefix = salt + encrypted if pgno == 1 else encrypted + partial = prefix + iv + mac_key = _derive_hmac_key(enc_key, salt) + digest = hmac.new(mac_key, partial[start:] + struct.pack("" + "x" * 20_001 + "", 1), "") + self.assertEqual( + _parse_fav_content(']>&y;', 1), + "", + ) + + +class CryptoIntegrityTests(unittest.TestCase): + def test_page_hmac_rejects_corruption(self): + key = bytes.fromhex(KEY) + salt = bytes.fromhex(SALT) + page = bytearray(_encrypted_page(key, salt, 1, 0x41)) + page[100] ^= 1 + with self.assertRaisesRegex(ValueError, "HMAC"): + decrypt_page(key, salt, bytes(page), 1) + + def test_wal_applies_only_checksum_valid_committed_frames(self): + key = bytes.fromhex(KEY) + salt = bytes.fromhex(SALT) + wal_salt = b"12345678" + header_prefix = struct.pack(">IIII", 0x377F0682, 3_007_000, PAGE_SZ, 0) + wal_salt + checksum = _wal_checksum(header_prefix, "<") + wal = bytearray(header_prefix + struct.pack(">II", *checksum)) + for pgno, database_size, marker in ((1, 0, 0x42), (2, 2, 0x43)): + page = _encrypted_page(key, salt, pgno, marker) + frame_prefix = struct.pack(">II", pgno, database_size) + wal_salt + checksum = _wal_checksum(frame_prefix[:8], "<", checksum) + checksum = _wal_checksum(page, "<", checksum) + wal.extend(frame_prefix + struct.pack(">II", *checksum) + page) + wal.extend(b"corrupt trailing frame") + + with tempfile.TemporaryDirectory() as root: + wal_path = os.path.join(root, "db-wal") + db_path = os.path.join(root, "db") + Path(wal_path).write_bytes(wal) + Path(db_path).write_bytes(b"\0" * (PAGE_SZ * 3)) + self.assertEqual(decrypt_wal(wal_path, db_path, key, salt), 2) + data = Path(db_path).read_bytes() + self.assertEqual(len(data), PAGE_SZ * 2) + self.assertEqual(data[:16], b"SQLite format 3\0") + self.assertEqual(data[PAGE_SZ], 0x43) + + +class ScannerPersistenceTests(unittest.TestCase): + def test_save_results_never_logs_key_and_writes_private_document(self): + with tempfile.TemporaryDirectory() as root: + output = os.path.join(root, "state", "all_keys.json") + logs = [] + db_files = [("message/a.db", "unused", 4096, SALT, b"")] + result = save_results(db_files, {SALT: ["message/a.db"]}, {SALT: KEY}, output, logs.append) + + self.assertEqual(result, {SALT: KEY}) + self.assertFalse(any(KEY in line for line in logs)) + self.assertEqual(read_private_json(output)["message/a.db"]["enc_key"], KEY) + + def test_macos_wrapper_redacts_binary_output_and_removes_intermediate(self): + with tempfile.TemporaryDirectory() as root: + db_dir = os.path.join(root, "db_storage") + os.mkdir(db_dir) + output = os.path.join(root, "state", "all_keys.json") + historical_output = os.path.join(root, "all_keys.json") + binary = os.path.join(root, "scanner") + Path(binary).write_bytes(b"safe scanner [REDACTED]") + observed_command = [] + + def fake_run(command, **_kwargs): + observed_command.extend(command) + native_output = command[command.index("--output") + 1] + atomic_write_json(native_output, { + "message/a.db": {"enc_key": KEY, "salt": SALT} + }) + return SimpleNamespace( + returncode=0, + stdout=f"message/a.db {KEY} {SALT}", + stderr="", + ) + + rendered = StringIO() + with patch("wechat_cli.keys.capabilities.require_supported_memory_scan"), \ + patch("wechat_cli.keys.scanner_macos._find_binary", return_value=binary), \ + patch("wechat_cli.keys.scanner_macos.subprocess.run", side_effect=fake_run), \ + redirect_stdout(rendered): + result = extract_macos_keys(db_dir, output) + + self.assertEqual(result, {SALT: KEY}) + self.assertNotIn(KEY, rendered.getvalue()) + self.assertFalse(os.path.exists(historical_output)) + self.assertEqual( + observed_command[observed_command.index("--db-dir") + 1], + os.path.realpath(db_dir), + ) + self.assertNotEqual( + observed_command[observed_command.index("--output") + 1], + historical_output, + ) + self.assertEqual(read_private_json(output)["message/a.db"]["enc_key"], KEY) + + def test_frozen_posix_helper_reenters_executable_without_python_module(self): + from wechat_cli.keys import extract_keys_for_init + + response = json.dumps({"message/a.db": {"enc_key": KEY, "salt": SALT}}) + with patch("wechat_cli.keys.platform.system", return_value="Linux"), \ + patch("wechat_cli.keys.require_supported_memory_scan"), \ + patch("wechat_cli.keys.os.path.isfile", return_value=True), \ + patch("wechat_cli.keys.subprocess.run") as run, \ + patch.object(sys, "frozen", True, create=True), \ + patch.object(sys, "executable", "/opt/wechat-cli"): + run.return_value = SimpleNamespace(returncode=0, stdout=response, stderr="") + document = extract_keys_for_init("/safe/db_storage") + + self.assertEqual(document["message/a.db"]["enc_key"], KEY) + command = run.call_args.args[0] + self.assertEqual( + command, + [ + "/usr/bin/sudo", "--", "/opt/wechat-cli", + "_elevated-scan-helper", "/safe/db_storage", + ], + ) + self.assertNotIn("-m", command) + self.assertNotIn(KEY, " ".join(command)) + + +class SafeBootstrapTests(unittest.TestCase): + def test_shared_extract_entrypoint_enforces_capability_gate(self): + from wechat_cli.keys import extract_keys + + with tempfile.TemporaryDirectory() as root, patch( + "wechat_cli.keys.require_supported_memory_scan", + side_effect=RuntimeError("WXCLI_SCAN_UNSUPPORTED_CLIENT"), + ): + with self.assertRaisesRegex(RuntimeError, "WXCLI_SCAN_UNSUPPORTED_CLIENT"): + extract_keys(root, os.path.join(root, "keys.json")) + self.assertFalse(os.path.exists(os.path.join(root, "keys.json"))) + + def test_unknown_client_is_rejected_by_exact_capability_manifest(self): + identity = { + "platform": "windows", + "executable": "Weixin.exe", + "client_version": "4.1.13.12", + "publisher": "Tencent Technology (Shenzhen) Company Limited", + "signature_valid": True, + "process_running": True, + } + self.assertFalse(evaluate_scan_capability(identity)["supported"]) + + verified_profile = ScannerProfile( + profile_id="test-only", + platform="windows", + executable="Weixin.exe", + client_version="4.1.13.12", + publisher=identity["publisher"], + scanner_algorithm="legacy-hex-v1", + acceptance="verified", + ) + result = evaluate_scan_capability(identity, (verified_profile,)) + self.assertTrue(result["supported"]) + self.assertEqual(result["profile"]["profile_id"], "test-only") + + def test_init_defaults_to_plan_and_never_calls_scanner(self): + with tempfile.TemporaryDirectory() as db_dir, patch( + "wechat_cli.commands.init.evaluate_scan_capability", + return_value={ + "supported": False, + "reason_code": "WXCLI_SCAN_UNSUPPORTED_CLIENT", + "profile": None, + "identity": { + "platform": "windows", + "executable": "Weixin.exe", + "client_version": "4.1.13.12", + "publisher": "unknown", + "process_running": True, + }, + }, + ), patch("wechat_cli.keys.extract_keys_for_init") as scanner: + result = CliRunner().invoke(init_command, ["--db-dir", db_dir]) + self.assertEqual(result.exit_code, 2, result.output) + self.assertIn("未执行扫描", result.output) + scanner.assert_not_called() + + def test_unknown_scan_stops_before_scanner_even_with_confirmation(self): + capability = { + "supported": False, + "reason_code": "WXCLI_SCAN_UNSUPPORTED_CLIENT", + "profile": None, + "identity": { + "platform": "windows", + "executable": "Weixin.exe", + "client_version": "4.1.13.12", + "publisher": "unknown", + "process_running": True, + }, + } + with tempfile.TemporaryDirectory() as db_dir, patch( + "wechat_cli.commands.init.evaluate_scan_capability", return_value=capability + ), patch("wechat_cli.keys.extract_keys_for_init") as scanner: + result = CliRunner().invoke( + init_command, + ["--db-dir", db_dir, "--scan-memory", "--confirm-memory-scan"], + ) + self.assertNotEqual(result.exit_code, 0) + self.assertIn("WXCLI_SCAN_UNSUPPORTED_CLIENT", result.output) + scanner.assert_not_called() + + def test_doctor_declares_and_observes_no_memory_read(self): + with patch("wechat_cli.keys.scanner_windows._read_mem") as read_mem: + report = build_doctor_report() + self.assertFalse(report["read_process_memory"]) + read_mem.assert_not_called() + + def test_trusted_import_requires_real_database_page_hmac(self): + key = bytes.fromhex(KEY) + salt = bytes.fromhex(SALT) + with tempfile.TemporaryDirectory() as db_dir: + db_path = os.path.join(db_dir, "message", "a.db") + os.makedirs(os.path.dirname(db_path)) + Path(db_path).write_bytes(_encrypted_page(key, salt, 1, 0x41)) + verified = verify_imported_key_document( + db_dir, {"message/a.db": {"enc_key": KEY, "salt": SALT}} + ) + self.assertEqual(verified[os.path.join("message", "a.db")]["enc_key"], KEY) + with self.assertRaisesRegex(ValueError, "WXCLI_IMPORT_HMAC_FAILED"): + verify_imported_key_document( + db_dir, {"message/a.db": {"enc_key": "ef" * 32, "salt": SALT}} + ) + + def test_keys_import_rejects_invalid_json_without_committing(self): + with patch("wechat_cli.commands.keys.commit_generation") as commit: + result = CliRunner().invoke(keys_command, ["import", "--stdin"], input="not-json") + self.assertNotEqual(result.exit_code, 0) + self.assertIn("WXCLI_IMPORT_INVALID_JSON", result.output) + commit.assert_not_called() + + def test_keys_import_rejects_duplicate_json_properties(self): + payload = '{"message/a.db":{"enc_key":"' + KEY + '","enc_key":"' + KEY + '"}}' + with patch("wechat_cli.commands.keys.commit_generation") as commit: + result = CliRunner().invoke(keys_command, ["import", "--stdin"], input=payload) + self.assertNotEqual(result.exit_code, 0) + self.assertIn("WXCLI_IMPORT_INVALID_JSON", result.output) + commit.assert_not_called() + + def test_macos_wrapper_rejects_historical_unredacted_artifact(self): + with tempfile.TemporaryDirectory() as root: + binary = os.path.join(root, "scanner") + Path(binary).write_bytes(b"%-25s %-66s %s") + with patch("wechat_cli.keys.capabilities.require_supported_memory_scan"), \ + patch("wechat_cli.keys.scanner_macos._find_binary", return_value=binary), \ + patch("wechat_cli.keys.scanner_macos.subprocess.run") as run: + with self.assertRaisesRegex(RuntimeError, "版本不安全"): + extract_macos_keys(root, os.path.join(root, "keys.json")) + run.assert_not_called() + + +class CacheLifecycleTests(unittest.TestCase): + def test_cache_redecrypts_after_provider_key_rotation(self): + with tempfile.TemporaryDirectory() as root: + state = os.path.join(root, "state") + db_path = os.path.join(root, "message", "a.db") + os.makedirs(os.path.dirname(db_path), exist_ok=True) + Path(db_path).write_bytes(bytes.fromhex(SALT) + b"encrypted") + + provider = JsonDatabaseKeyProvider(os.path.join(state, "all_keys.json")) + provider.replace({"message/a.db": {"enc_key": KEY, "salt": SALT}}) + cache = DBCache(provider, root, state) + observed_keys = [] + + def fake_decrypt(_source, destination, enc_key): + observed_keys.append(enc_key.hex()) + Path(destination).write_bytes(b"SQLite format 3\0") + + rotated = "ef" * 32 + try: + with patch("wechat_cli.core.db_cache.full_decrypt", side_effect=fake_decrypt): + first = cache.get("message/a.db") + self.assertEqual(cache.get("message/a.db"), first) + provider.replace({"message/a.db": {"enc_key": rotated, "salt": SALT}}) + self.assertEqual(cache.get("message/a.db"), first) + finally: + cache.cleanup() + + self.assertEqual(observed_keys, [KEY, rotated]) + + def test_cache_uses_run_private_directory_and_cleanup_is_scoped(self): + with tempfile.TemporaryDirectory() as root: + state = os.path.join(root, "state") + sibling = os.path.join(state, "cache", "keep.txt") + os.makedirs(os.path.dirname(sibling), exist_ok=True) + Path(sibling).write_text("keep", encoding="utf-8") + + key_path = os.path.join(state, "all_keys.json") + provider = JsonDatabaseKeyProvider(key_path) + provider.replace({"message/a.db": {"enc_key": KEY, "salt": SALT}}) + cache = DBCache(provider, root, state) + run_dir = cache._cache_dir + Path(os.path.join(run_dir, "decrypted.db")).write_bytes(b"SQLite format 3\0") + cache.cleanup() + + self.assertFalse(os.path.exists(run_dir)) + self.assertTrue(os.path.exists(sibling)) + + def test_cache_removes_only_unlocked_well_formed_stale_runs(self): + with tempfile.TemporaryDirectory() as root: + state = os.path.join(root, "state") + cache_root = os.path.join(state, "cache") + stale = os.path.join(cache_root, "run-999-" + "a" * 32) + unrelated = os.path.join(cache_root, "keep-directory") + os.makedirs(stale) + os.makedirs(unrelated) + Path(os.path.join(stale, ".lease")).write_bytes(b"0") + provider = JsonDatabaseKeyProvider(os.path.join(state, "all_keys.json")) + provider.replace({"message/a.db": {"enc_key": KEY, "salt": SALT}}) + + cache = DBCache(provider, root, state) + try: + self.assertFalse(os.path.exists(stale)) + self.assertTrue(os.path.exists(unrelated)) + finally: + cache.cleanup() + + def test_cache_rejects_drive_relative_database_reference(self): + with tempfile.TemporaryDirectory() as root: + state = os.path.join(root, "state") + provider = JsonDatabaseKeyProvider(os.path.join(state, "all_keys.json")) + provider.replace({"message/a.db": {"enc_key": KEY, "salt": SALT}}) + cache = DBCache(provider, root, state) + try: + with self.assertRaises(ValueError): + cache._source_paths(r"C:outside.db") + finally: + cache.cleanup() + + +if __name__ == "__main__": + unittest.main() diff --git a/wechat_cli/bin/find_all_keys_macos.c b/wechat_cli/bin/find_all_keys_macos.c index eb6a9e5..dd0532d 100644 --- a/wechat_cli/bin/find_all_keys_macos.c +++ b/wechat_cli/bin/find_all_keys_macos.c @@ -12,19 +12,17 @@ * cc -O2 -o find_all_keys_macos find_all_keys_macos.c -framework Foundation * * Usage: - * sudo ./find_all_keys_macos [pid] - * If pid is omitted, automatically finds WeChat PID. + * sudo ./find_all_keys_macos --db-dir --output [--pid ] * - * Output: JSON file at ./all_keys.json (compatible with decrypt_db.py) + * Output: JSON file at the explicit --output path. */ #include #include #include #include -#include #include -#include +#include #include #include #include @@ -49,6 +47,8 @@ static int read_db_salt(const char *path, char *salt_hex_out); static char g_db_salts[MAX_DBS][33]; static char g_db_names[MAX_DBS][256]; static int g_db_count = 0; +static char g_db_root[PATH_MAX]; +static size_t g_db_root_len = 0; static int nftw_collect_db(const char *fpath, const struct stat *sb, int typeflag, struct FTW *ftwbuf) { (void)sb; (void)ftwbuf; @@ -61,13 +61,9 @@ static int nftw_collect_db(const char *fpath, const struct stat *sb, if (read_db_salt(fpath, salt) != 0) return 0; strcpy(g_db_salts[g_db_count], salt); - /* Extract relative path from db_storage/ */ - const char *rel = strstr(fpath, "db_storage/"); - if (rel) rel += strlen("db_storage/"); - else { - rel = strrchr(fpath, '/'); - rel = rel ? rel + 1 : fpath; - } + if (strncmp(fpath, g_db_root, g_db_root_len) != 0 || + fpath[g_db_root_len] != '/') return 0; + const char *rel = fpath + g_db_root_len + 1; strncpy(g_db_names[g_db_count], rel, 255); g_db_names[g_db_count][255] = '\0'; printf(" %s: salt=%s\n", g_db_names[g_db_count], salt); @@ -80,7 +76,7 @@ static int is_hex_char(unsigned char c) { } static pid_t find_wechat_pid(void) { - FILE *fp = popen("pgrep -x WeChat", "r"); + FILE *fp = popen("/usr/bin/pgrep -x WeChat", "r"); if (!fp) return -1; char buf[64]; pid_t pid = -1; @@ -106,11 +102,27 @@ static int read_db_salt(const char *path, char *salt_hex_out) { } int main(int argc, char *argv[]) { - pid_t pid; - if (argc >= 2) - pid = atoi(argv[1]); - else - pid = find_wechat_pid(); + umask(0077); + pid_t pid = 0; + const char *db_storage_path = NULL; + const char *out_path = NULL; + for (int argi = 1; argi < argc; argi++) { + if (strcmp(argv[argi], "--db-dir") == 0 && argi + 1 < argc) { + db_storage_path = argv[++argi]; + } else if (strcmp(argv[argi], "--output") == 0 && argi + 1 < argc) { + out_path = argv[++argi]; + } else if (strcmp(argv[argi], "--pid") == 0 && argi + 1 < argc) { + pid = atoi(argv[++argi]); + } else { + fprintf(stderr, "invalid arguments\n"); + return 2; + } + } + if (!db_storage_path || !out_path) { + fprintf(stderr, "--db-dir and --output are required\n"); + return 2; + } + if (pid <= 0) pid = find_wechat_pid(); if (pid <= 0) { fprintf(stderr, "WeChat not running or invalid PID\n"); @@ -132,41 +144,18 @@ int main(int argc, char *argv[]) { } printf("Got task port: %u\n", task); - /* Resolve real user's HOME (sudo may change HOME to /var/root) */ - const char *home = getenv("HOME"); - const char *sudo_user = getenv("SUDO_USER"); - if (sudo_user) { - struct passwd *pw = getpwnam(sudo_user); - if (pw && pw->pw_dir) - home = pw->pw_dir; - } - if (!home) home = "/root"; - printf("User home: %s\n", home); - - /* Collect DB salts by recursively walking db_storage directories. - * Note: POSIX glob() does not support ** recursive matching on macOS, - * so we use nftw() to walk the directory tree instead. */ + /* Collect DB salts from the exact selected db_storage tree. */ printf("\nScanning for DB files...\n"); - char db_base_dir[512]; - snprintf(db_base_dir, sizeof(db_base_dir), - "%s/Library/Containers/com.tencent.xinWeChat/Data/Documents/xwechat_files", - home); - - /* Walk each account's db_storage directory */ - DIR *xdir = opendir(db_base_dir); - if (xdir) { - struct dirent *ent; - while ((ent = readdir(xdir)) != NULL) { - if (ent->d_name[0] == '.') continue; - char storage_path[768]; - snprintf(storage_path, sizeof(storage_path), - "%s/%s/db_storage", db_base_dir, ent->d_name); - struct stat st; - if (stat(storage_path, &st) == 0 && S_ISDIR(st.st_mode)) { - nftw(storage_path, nftw_collect_db, 20, FTW_PHYS); - } - } - closedir(xdir); + struct stat db_stat; + if (lstat(db_storage_path, &db_stat) != 0 || !S_ISDIR(db_stat.st_mode) || + realpath(db_storage_path, g_db_root) == NULL) { + fprintf(stderr, "invalid db_storage directory\n"); + return 2; + } + g_db_root_len = strlen(g_db_root); + if (nftw(g_db_root, nftw_collect_db, 20, FTW_PHYS) != 0) { + fprintf(stderr, "failed to scan db_storage directory\n"); + return 1; } printf("Found %d encrypted DBs\n", g_db_count); @@ -266,10 +255,10 @@ int main(int argc, char *argv[]) { total_scanned / 1024 / 1024, region_count, key_count); /* Match keys to DBs */ - printf("\n%-25s %-66s %s\n", "Database", "Key", "Salt"); - printf("%-25s %-66s %s\n", + printf("\n%-25s %-12s %s\n", "Database", "Key", "Salt"); + printf("%-25s %-12s %s\n", "-------------------------", - "------------------------------------------------------------------", + "------------", "--------------------------------"); int matched = 0; @@ -282,9 +271,9 @@ int main(int argc, char *argv[]) { break; } } - printf("%-25s %-66s %s\n", + printf("%-25s %-12s %s\n", db ? db : "(unknown)", - keys[i].key_hex, + "[REDACTED]", keys[i].salt_hex); } printf("\nMatched %d/%d keys to known DBs\n", matched, key_count); @@ -292,28 +281,32 @@ int main(int argc, char *argv[]) { /* Save JSON: { "rel/path.db": { "enc_key": "hex" }, ... } * Uses forward slashes (native macOS paths, valid JSON without escaping). */ - const char *out_path = "all_keys.json"; FILE *fp = fopen(out_path, "w"); - if (fp) { - fprintf(fp, "{\n"); - int first = 1; - for (int i = 0; i < key_count; i++) { - const char *db = NULL; - for (int j = 0; j < g_db_count; j++) { - if (strcmp(keys[i].salt_hex, g_db_salts[j]) == 0) { - db = g_db_names[j]; - break; - } + if (!fp) { + fprintf(stderr, "failed to open output file\n"); + return 1; + } + fprintf(fp, "{\n"); + int first = 1; + for (int i = 0; i < key_count; i++) { + const char *db = NULL; + for (int j = 0; j < g_db_count; j++) { + if (strcmp(keys[i].salt_hex, g_db_salts[j]) == 0) { + db = g_db_names[j]; + break; } - if (!db) continue; - fprintf(fp, "%s \"%s\": {\"enc_key\": \"%s\"}", - first ? "" : ",\n", db, keys[i].key_hex); - first = 0; } - fprintf(fp, "\n}\n"); - fclose(fp); - printf("Saved to %s\n", out_path); + if (!db) continue; + fprintf(fp, "%s \"%s\": {\"enc_key\": \"%s\", \"salt\": \"%s\"}", + first ? "" : ",\n", db, keys[i].key_hex, keys[i].salt_hex); + first = 0; + } + fprintf(fp, "\n}\n"); + if (fclose(fp) != 0) { + fprintf(stderr, "failed to finalize output file\n"); + return 1; } + printf("Saved to %s\n", out_path); return 0; } diff --git a/wechat_cli/commands/doctor.py b/wechat_cli/commands/doctor.py new file mode 100644 index 0000000..d9dab0e --- /dev/null +++ b/wechat_cli/commands/doctor.py @@ -0,0 +1,58 @@ +"""只读安全预检。""" + +import json +import os + +import click + +from ..core.config import STATE_DIR, auto_detect_db_dir +from ..core.key_document import select_default_key_backend +from ..core.state import resolve_active_generation +from ..keys.capabilities import evaluate_scan_capability + + +def build_doctor_report(db_dir=None): + detected = os.path.abspath(db_dir) if db_dir else auto_detect_db_dir() + try: + active = resolve_active_generation(STATE_DIR) + state_error = None + except Exception as error: + active = None + state_error = type(error).__name__ + try: + backend = select_default_key_backend() + backend_available = True + except RuntimeError: + backend = None + backend_available = False + return { + "schema_version": 1, + "read_process_memory": False, + "db_dir_present": bool(detected and os.path.isdir(detected)), + "state": { + "initialized": active is not None, + "valid": state_error is None, + "error_type": state_error, + }, + "key_store": {"available": backend_available, "backend": backend}, + "memory_scan": evaluate_scan_capability(), + } + + +@click.command("doctor") +@click.option("--db-dir", default=None, help="待检查的微信数据目录") +@click.option("--json", "as_json", is_flag=True, help="输出机器可读 JSON") +def doctor(db_dir, as_json): + """检查数据库、状态和扫描能力；不会读取微信进程内存。""" + report = build_doctor_report(db_dir) + if as_json: + click.echo(json.dumps(report, ensure_ascii=False, sort_keys=True)) + return + click.echo("WeChat CLI 安全检查") + click.echo(f" 数据目录: {'可用' if report['db_dir_present'] else '未找到'}") + click.echo(f" 初始化状态: {'可用' if report['state']['initialized'] else '未初始化'}") + click.echo(f" 系统密钥库: {report['key_store']['backend'] or '不可用'}") + scan = report["memory_scan"] + click.echo(f" 内存扫描: {'已验收' if scan['supported'] else '未验收'}") + click.echo(f" 结果代码: {scan['reason_code']}") + click.echo(" 本次检查读取进程内存: 否") diff --git a/wechat_cli/commands/favorites.py b/wechat_cli/commands/favorites.py index 2b91d19..665612a 100644 --- a/wechat_cli/commands/favorites.py +++ b/wechat_cli/commands/favorites.py @@ -2,13 +2,13 @@ import os import sqlite3 -import xml.etree.ElementTree as ET from contextlib import closing from datetime import datetime import click from ..core.contacts import get_contact_names +from ..core.xml_utils import parse_untrusted_xml from ..output.formatter import output _FAV_TYPE_MAP = { @@ -24,9 +24,8 @@ def _parse_fav_content(content, fav_type): """从 XML content 提取摘要信息。""" if not content: return '' - try: - root = ET.fromstring(content) - except ET.ParseError: + root = parse_untrusted_xml(content) + if root is None: return '' item = root if root.tag == 'favitem' else root.find('.//favitem') if item is None: @@ -52,7 +51,7 @@ def _parse_fav_content(content, fav_type): @click.command("favorites") -@click.option("--limit", default=20, help="返回数量") +@click.option("--limit", default=20, type=click.IntRange(1, 500), help="返回数量（1-500）") @click.option("--type", "fav_type", default=None, type=click.Choice(list(_FAV_TYPE_FILTERS.keys())), help="按类型过滤: text/image/article/card/video") diff --git a/wechat_cli/commands/init.py b/wechat_cli/commands/init.py index 5396a4f..08c6653 100644 --- a/wechat_cli/commands/init.py +++ b/wechat_cli/commands/init.py @@ -1,72 +1,100 @@ -"""init 命令 — 交互式初始化，提取密钥并生成配置""" +"""init 命令 — 显式授权后扫描并以 generation 原子初始化。""" -import json import os -import sys import click -from ..core.config import STATE_DIR, CONFIG_FILE, KEYS_FILE, auto_detect_db_dir +from ..core.config import CONFIG_FILE, KEYS_FILE, STATE_DIR, auto_detect_db_dir +from ..core.secure_files import ensure_private_directory, redact_sensitive_text +from ..core.state import commit_generation, resolve_active_generation, state_lock +from ..keys.capabilities import evaluate_scan_capability + + +def _resolve_db_dir(db_dir): + if db_dir is None: + db_dir = auto_detect_db_dir() + if db_dir is None: + raise click.ClickException( + "[WXCLI_DB_NOT_FOUND] 未能自动检测微信数据目录，请使用 --db-dir 指定" + ) + db_dir = os.path.abspath(db_dir) + if not os.path.isdir(db_dir): + raise click.ClickException(f"[WXCLI_DB_NOT_FOUND] 目录不存在: {db_dir}") + return db_dir + + +def _render_scan_plan(db_dir, capability): + identity = capability["identity"] + click.echo("初始化计划（只读，尚未读取微信进程内存）") + click.echo(f" 数据目录: {db_dir}") + click.echo(f" 客户端: {identity['executable']} {identity['client_version']}") + click.echo(f" 发布者: {identity['publisher']}") + click.echo(f" 扫描能力: {'已验收' if capability['supported'] else '未验收'}") + click.echo(f" 结果代码: {capability['reason_code']}") @click.command() @click.option("--db-dir", default=None, help="微信数据目录路径（默认自动检测）") -@click.option("--force", is_flag=True, help="强制重新提取密钥") -def init(db_dir, force): - """初始化 wechat-cli：提取密钥并生成配置""" - click.echo("WeChat CLI 初始化") - click.echo("=" * 40) +@click.option("--force", is_flag=True, help="替换已有初始化状态") +@click.option("--dry-run", is_flag=True, help="只显示计划，不读进程内存且不写状态") +@click.option("--scan-memory", is_flag=True, help="明确请求读取微信进程内存") +@click.option( + "--confirm-memory-scan", + is_flag=True, + help="确认已知晓进程内存扫描可能触发客户端安全告警", +) +def init(db_dir, force, dry_run, scan_memory, confirm_memory_scan): + """初始化 wechat-cli；默认只做安全预检，不会扫描进程内存。""" + if hasattr(os, "geteuid") and os.geteuid() == 0 and os.environ.get("SUDO_USER"): + raise click.ClickException( + "[WXCLI_ROOT_SESSION_DENIED] 请以普通用户运行；仅扫描 helper 可请求 sudo" + ) - # 1. 检查是否已初始化 - if os.path.exists(CONFIG_FILE) and os.path.exists(KEYS_FILE) and not force: - click.echo(f"已初始化（配置: {CONFIG_FILE}）") - click.echo("使用 --force 重新提取密钥") - return + db_dir = _resolve_db_dir(db_dir) + capability = evaluate_scan_capability() + _render_scan_plan(db_dir, capability) - # 2. 创建状态目录 - os.makedirs(STATE_DIR, exist_ok=True) + if dry_run: + return + if not scan_memory: + click.echo( + "\n未执行扫描。推荐将受信 helper 的 JSON 输出通过 " + "wechat-cli keys import --stdin 导入。" + ) + click.echo("如确需内存扫描，必须同时使用 --scan-memory --confirm-memory-scan。") + click.get_current_context().exit(2) + if not confirm_memory_scan: + raise click.ClickException( + "[WXCLI_SCAN_CONFIRMATION_REQUIRED] 缺少 --confirm-memory-scan" + ) + if not capability["supported"]: + raise click.ClickException( + "[WXCLI_SCAN_UNSUPPORTED_CLIENT] 当前客户端版本/发布者没有通过真实环境验收；" + "已在申请进程内存读取权限前停止" + ) - # 3. 确定 db_dir - if db_dir is None: - db_dir = auto_detect_db_dir() - if db_dir is None: - click.echo("[!] 未能自动检测到微信数据目录", err=True) - click.echo("请通过 --db-dir 参数指定，例如:", err=True) - click.echo(" wechat-cli init --db-dir ~/path/to/db_storage", err=True) - sys.exit(1) - click.echo(f"[+] 检测到微信数据目录: {db_dir}") - else: - db_dir = os.path.abspath(db_dir) - if not os.path.isdir(db_dir): - click.echo(f"[!] 目录不存在: {db_dir}", err=True) - sys.exit(1) - click.echo(f"[+] 使用指定数据目录: {db_dir}") + ensure_private_directory(STATE_DIR, tighten_existing=True) + with state_lock(STATE_DIR): + active = resolve_active_generation(STATE_DIR) + legacy_ready = os.path.exists(CONFIG_FILE) and os.path.exists(KEYS_FILE) + if active and not force: + click.echo(f"已初始化（配置: {active}）") + click.echo("使用 --force 替换已有状态") + return + if legacy_ready and not force: + raise click.ClickException( + "[WXCLI_LEGACY_KEYS_PRESENT] 检测到旧版明文密钥状态；使用 --force 迁移" + ) - # 4. 提取密钥 - click.echo("\n开始提取密钥...") - try: - from ..keys import extract_keys - key_map = extract_keys(db_dir, KEYS_FILE) - except RuntimeError as e: - click.echo(f"\n[!] 密钥提取失败: {e}", err=True) - if "sudo" not in str(e).lower(): - click.echo("提示: macOS/Linux 可能需要 sudo 权限", err=True) - sys.exit(1) - except Exception as e: - click.echo(f"\n[!] 密钥提取出错: {e}", err=True) - sys.exit(1) + click.echo("\n开始已授权的内存扫描...") + try: + from ..keys import extract_keys_for_init - # 5. 写入配置 - cfg = { - "db_dir": db_dir, - } - with open(CONFIG_FILE, "w", encoding="utf-8") as f: - json.dump(cfg, f, indent=2, ensure_ascii=False) + key_document = extract_keys_for_init(db_dir) + generation = commit_generation(STATE_DIR, {"db_dir": db_dir}, key_document) + except Exception as error: + raise click.ClickException(redact_sensitive_text(str(error))) from None - click.echo(f"\n[+] 初始化完成!") - click.echo(f" 配置: {CONFIG_FILE}") - click.echo(f" 密钥: {KEYS_FILE}") - click.echo(f" 提取到 {len(key_map)} 个数据库密钥") - click.echo("\n现在可以使用:") - click.echo(" wechat-cli sessions") - click.echo(" wechat-cli history \"联系人\"") + click.echo("\n[+] 初始化完成") + click.echo(f" 配置: {os.path.join(generation, 'config.json')}") + click.echo(f" 提取到 {len(key_document)} 个数据库密钥") diff --git a/wechat_cli/commands/keys.py b/wechat_cli/commands/keys.py new file mode 100644 index 0000000..b206aaf --- /dev/null +++ b/wechat_cli/commands/keys.py @@ -0,0 +1,78 @@ +"""密钥导入命令。""" + +import json +import os +import sys + +import click + +from ..core.config import CONFIG_FILE, KEYS_FILE, STATE_DIR, auto_detect_db_dir +from ..core.secure_files import ensure_private_directory, redact_sensitive_text +from ..core.state import commit_generation, resolve_active_generation, state_lock +from ..keys.trusted_import import verify_imported_key_document + +MAX_STDIN_BYTES = 2 * 1024 * 1024 + + +@click.group("keys") +def keys(): + """管理数据库密钥；不会读取微信进程内存。""" + + +def _reject_duplicate_pairs(pairs): + document = {} + for name, value in pairs: + if name in document: + raise ValueError("duplicate JSON property") + document[name] = value + return document + + +def _read_stdin_document(): + if sys.stdin.isatty(): + raise click.ClickException( + "[WXCLI_IMPORT_PIPE_REQUIRED] --stdin 只接受管道输入，不接受交互粘贴" + ) + payload = click.get_text_stream("stdin").read(MAX_STDIN_BYTES + 1) + if len(payload.encode("utf-8")) > MAX_STDIN_BYTES: + raise click.ClickException("[WXCLI_IMPORT_TOO_LARGE] stdin 密钥文档超过 2 MiB") + try: + return json.loads(payload, object_pairs_hook=_reject_duplicate_pairs) + except (json.JSONDecodeError, UnicodeError, ValueError): + raise click.ClickException("[WXCLI_IMPORT_INVALID_JSON] stdin 不是有效 JSON") from None + + +@keys.command("import") +@click.option("--stdin", "from_stdin", is_flag=True, help="从标准输入读取 JSON 密钥文档") +@click.option("--db-dir", default=None, help="微信数据目录路径（默认自动检测）") +@click.option("--force", is_flag=True, help="替换已有初始化状态") +def import_keys(from_stdin, db_dir, force): + """导入受信 helper 输出；每个密钥必须通过真实数据库页 HMAC。""" + if not from_stdin: + raise click.ClickException("[WXCLI_IMPORT_STDIN_REQUIRED] 必须显式使用 --stdin") + document = _read_stdin_document() + detected = db_dir or auto_detect_db_dir() + if not detected: + raise click.ClickException("[WXCLI_DB_NOT_FOUND] 未找到微信数据目录") + db_dir = os.path.abspath(detected) + if not os.path.isdir(db_dir): + raise click.ClickException("[WXCLI_DB_NOT_FOUND] 未找到微信数据目录") + try: + verified = verify_imported_key_document(db_dir, document) + except Exception as error: + raise click.ClickException(redact_sensitive_text(str(error))) from None + + ensure_private_directory(STATE_DIR, tighten_existing=True) + with state_lock(STATE_DIR): + active = resolve_active_generation(STATE_DIR) + legacy_ready = os.path.exists(CONFIG_FILE) and os.path.exists(KEYS_FILE) + if (active or legacy_ready) and not force: + raise click.ClickException( + "[WXCLI_STATE_EXISTS] 已存在初始化状态；如需替换请使用 --force" + ) + try: + generation = commit_generation(STATE_DIR, {"db_dir": db_dir}, verified) + except Exception as error: + raise click.ClickException(redact_sensitive_text(str(error))) from None + click.echo(f"[+] 已安全导入 {len(verified)} 个数据库密钥") + click.echo(f" 配置: {os.path.join(generation, 'config.json')}") diff --git a/wechat_cli/core/config.py b/wechat_cli/core/config.py index 0294d2b..0eb839d 100644 --- a/wechat_cli/core/config.py +++ b/wechat_cli/core/config.py @@ -1,7 +1,6 @@ """配置加载 — 从 ~/.wechat-cli/ 读取自包含配置""" import glob as glob_mod -import json import os import platform import sys @@ -148,15 +147,17 @@ def auto_detect_db_dir(): def load_config(config_path=None): """加载配置。默认从 ~/.wechat-cli/config.json 读取。""" if config_path is None: - config_path = CONFIG_FILE + from .state import resolve_config_path + + config_path = resolve_config_path(STATE_DIR, CONFIG_FILE) cfg = {} if os.path.exists(config_path): - try: - with open(config_path, encoding="utf-8") as f: - cfg = json.load(f) - except json.JSONDecodeError: - cfg = {} + from .secure_files import read_private_json + + cfg = read_private_json(config_path) + if not isinstance(cfg, dict): + raise ValueError("config.json 必须是对象") # db_dir 缺失时，自动检测 db_dir = cfg.get("db_dir", "") diff --git a/wechat_cli/core/context.py b/wechat_cli/core/context.py index f8cc412..86b8974 100644 --- a/wechat_cli/core/context.py +++ b/wechat_cli/core/context.py @@ -1,19 +1,20 @@ """应用上下文 — 单例持有配置、缓存、密钥等共享状态""" import atexit -import json import os from .config import load_config, STATE_DIR from .db_cache import DBCache -from .key_utils import strip_key_metadata +from .key_document import create_key_provider from .messages import find_msg_db_keys +from .secure_files import ensure_private_directory class AppContext: """每次 CLI 调用初始化一次，被所有命令共享。""" def __init__(self, config_path=None): + ensure_private_directory(STATE_DIR, tighten_existing=True) self.cfg = load_config(config_path) self.db_dir = self.cfg["db_dir"] self.decrypted_dir = self.cfg["decrypted_dir"] @@ -25,16 +26,16 @@ def __init__(self, config_path=None): "请运行: wechat-cli init" ) - with open(self.keys_file, encoding="utf-8") as f: - self.all_keys = strip_key_metadata(json.load(f)) + self.key_provider = create_key_provider( + self.keys_file, + backend=self.cfg.get("key_backend"), + namespace=self.cfg.get("key_namespace"), + ) - self.cache = DBCache(self.all_keys, self.db_dir) + self.cache = DBCache(self.key_provider, self.db_dir, STATE_DIR) atexit.register(self.cache.cleanup) - self.msg_db_keys = find_msg_db_keys(self.all_keys) - - # 确保状态目录存在 - os.makedirs(STATE_DIR, exist_ok=True) + self.msg_db_keys = find_msg_db_keys(self.key_provider.references()) def display_name_fn(self, username, names): from .contacts import display_name_for_username diff --git a/wechat_cli/core/crypto.py b/wechat_cli/core/crypto.py index 0285dc1..793e18a 100644 --- a/wechat_cli/core/crypto.py +++ b/wechat_cli/core/crypto.py @@ -1,77 +1,132 @@ -"""数据库解密 — SQLCipher 4, AES-256-CBC""" +"""数据库解密 — SQLCipher 4 AES-256-CBC、页 HMAC 与 SQLite WAL 校验。""" +import hashlib +import hmac import os import struct from Crypto.Cipher import AES + PAGE_SZ = 4096 KEY_SZ = 32 SALT_SZ = 16 RESERVE_SZ = 80 # IV(16) + HMAC-SHA512(64) -SQLITE_HDR = b'SQLite format 3\x00' +SQLITE_HDR = b"SQLite format 3\x00" WAL_HEADER_SZ = 32 WAL_FRAME_HEADER_SZ = 24 +_WAL_MAGIC_LITTLE = 0x377F0682 +_WAL_MAGIC_BIG = 0x377F0683 + + +def _derive_hmac_key(enc_key, database_salt): + if len(enc_key) != KEY_SZ or len(database_salt) != SALT_SZ: + raise ValueError("SQLCipher key 或 salt 长度无效") + hmac_salt = bytes(value ^ 0x3A for value in database_salt) + return hashlib.pbkdf2_hmac("sha512", enc_key, hmac_salt, 2, dklen=KEY_SZ) -def decrypt_page(enc_key, page_data, pgno): +def _verify_page_hmac(enc_key, database_salt, page_data, pgno): + if len(page_data) != PAGE_SZ or pgno <= 0: + raise ValueError("SQLCipher 页大小或页号无效") + hmac_key = _derive_hmac_key(enc_key, database_salt) + authenticated_end = PAGE_SZ - RESERVE_SZ + 16 + start = SALT_SZ if pgno == 1 else 0 + payload = page_data[start:authenticated_end] + struct.pack(" 0: - page = page + b'\x00' * (PAGE_SZ - len(page)) - else: - break - fout.write(decrypt_page(enc_key, page, pgno)) + with open(db_path, "rb") as source, open(out_path, "wb") as destination: + first_page = source.read(PAGE_SZ) + database_salt = first_page[:SALT_SZ] + destination.write(decrypt_page(enc_key, database_salt, first_page, 1)) + for pgno in range(2, total_pages + 1): + page = source.read(PAGE_SZ) + if len(page) != PAGE_SZ: + raise ValueError("加密数据库读取到截断页") + destination.write(decrypt_page(enc_key, database_salt, page, pgno)) return total_pages -def decrypt_wal(wal_path, out_path, enc_key): +def _wal_checksum(data, byte_order, checksum=(0, 0)): + if len(data) % 8: + raise ValueError("WAL checksum 输入长度无效") + values = struct.unpack(f"{byte_order}{len(data) // 4}I", data) + s0, s1 = checksum + for index in range(0, len(values), 2): + s0 = (s0 + values[index] + s1) & 0xFFFFFFFF + s1 = (s1 + values[index + 1] + s0) & 0xFFFFFFFF + return s0, s1 + + +def decrypt_wal(wal_path, out_path, enc_key, database_salt): if not os.path.exists(wal_path): return 0 wal_size = os.path.getsize(wal_path) if wal_size <= WAL_HEADER_SZ: return 0 - patched = 0 - with open(wal_path, 'rb') as wf, open(out_path, 'r+b') as df: - wal_hdr = wf.read(WAL_HEADER_SZ) - wal_salt1 = struct.unpack('>I', wal_hdr[16:20])[0] - wal_salt2 = struct.unpack('>I', wal_hdr[20:24])[0] + with open(wal_path, "rb") as wal: + header = wal.read(WAL_HEADER_SZ) + magic, version, page_size = struct.unpack(">III", header[:12]) + if magic not in (_WAL_MAGIC_LITTLE, _WAL_MAGIC_BIG): + raise ValueError("WAL magic 无效") + if version != 3_007_000 or page_size != PAGE_SZ: + raise ValueError("WAL 版本或页大小不受支持") + byte_order = "<" if magic == _WAL_MAGIC_LITTLE else ">" + checksum = _wal_checksum(header[:24], byte_order) + if checksum != struct.unpack(">II", header[24:32]): + raise ValueError("WAL header checksum 无效") + salt = header[16:24] + valid_frames = [] + last_commit_index = -1 + last_commit_size = 0 frame_size = WAL_FRAME_HEADER_SZ + PAGE_SZ - while wf.tell() + frame_size <= wal_size: - fh = wf.read(WAL_FRAME_HEADER_SZ) - if len(fh) < WAL_FRAME_HEADER_SZ: + while wal.tell() + frame_size <= wal_size: + frame_header = wal.read(WAL_FRAME_HEADER_SZ) + page = wal.read(PAGE_SZ) + pgno, database_size = struct.unpack(">II", frame_header[:8]) + if pgno == 0 or frame_header[8:16] != salt: break - pgno = struct.unpack('>I', fh[0:4])[0] - frame_salt1 = struct.unpack('>I', fh[8:12])[0] - frame_salt2 = struct.unpack('>I', fh[12:16])[0] - ep = wf.read(PAGE_SZ) - if len(ep) < PAGE_SZ: + checksum = _wal_checksum(frame_header[:8], byte_order, checksum) + checksum = _wal_checksum(page, byte_order, checksum) + if checksum != struct.unpack(">II", frame_header[16:24]): break - if pgno == 0 or pgno > 1000000: - continue - if frame_salt1 != wal_salt1 or frame_salt2 != wal_salt2: + valid_frames.append((pgno, page)) + if database_size: + last_commit_index = len(valid_frames) - 1 + last_commit_size = database_size + + if last_commit_index < 0: + return 0 + patched = 0 + with open(out_path, "r+b") as database: + for pgno, page in valid_frames[: last_commit_index + 1]: + if pgno > last_commit_size: continue - dec = decrypt_page(enc_key, ep, pgno) - df.seek((pgno - 1) * PAGE_SZ) - df.write(dec) + decrypted = decrypt_page(enc_key, database_salt, page, pgno) + database.seek((pgno - 1) * PAGE_SZ) + database.write(decrypted) patched += 1 + database.truncate(last_commit_size * PAGE_SZ) return patched diff --git a/wechat_cli/core/db_cache.py b/wechat_cli/core/db_cache.py index 2cf5f64..76553f5 100644 --- a/wechat_cli/core/db_cache.py +++ b/wechat_cli/core/db_cache.py @@ -1,91 +1,228 @@ -"""解密数据库缓存 — mtime 检测变化，跨会话复用""" +"""解密数据库缓存 — 稳定快照、进程租约和退出清理。""" import hashlib -import json import os -import tempfile +import re +import shutil +import time +import uuid -from .crypto import full_decrypt, decrypt_wal -from .key_utils import get_key_info +from .crypto import decrypt_wal, full_decrypt +from .key_document import DatabaseKeyProvider +from .key_utils import _is_safe_rel_path +from .secure_files import ensure_private_directory -class DBCache: - CACHE_DIR = os.path.join(tempfile.gettempdir(), "wechat_cli_cache") - MTIME_FILE = os.path.join(tempfile.gettempdir(), "wechat_cli_cache", "_mtimes.json") +_RUN_DIR_RE = re.compile(r"^run-[0-9]+-[0-9a-f]{32}$") +_SNAPSHOT_ATTEMPTS = 3 - def __init__(self, all_keys, db_dir): - self._all_keys = all_keys - self._db_dir = db_dir - self._cache = {} # rel_key -> (db_mtime, wal_mtime, tmp_path) - os.makedirs(self.CACHE_DIR, exist_ok=True) - self._load_persistent_cache() - def _cache_path(self, rel_key): - h = hashlib.md5(rel_key.encode()).hexdigest()[:12] - return os.path.join(self.CACHE_DIR, f"{h}.db") +def _lock_descriptor(descriptor, *, nonblocking): + if os.name == "nt": + import msvcrt + + os.lseek(descriptor, 0, os.SEEK_SET) + if os.fstat(descriptor).st_size == 0: + os.write(descriptor, b"0") + os.fsync(descriptor) + os.lseek(descriptor, 0, os.SEEK_SET) + mode = msvcrt.LK_NBLCK if nonblocking else msvcrt.LK_LOCK + msvcrt.locking(descriptor, mode, 1) + else: + import fcntl + + mode = fcntl.LOCK_EX | (fcntl.LOCK_NB if nonblocking else 0) + fcntl.flock(descriptor, mode) + + +def _unlock_descriptor(descriptor): + if os.name == "nt": + import msvcrt + + os.lseek(descriptor, 0, os.SEEK_SET) + msvcrt.locking(descriptor, msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(descriptor, fcntl.LOCK_UN) + - def _load_persistent_cache(self): - if not os.path.exists(self.MTIME_FILE): - return +def _source_signature(path): + try: + stat = os.stat(path, follow_symlinks=False) + except FileNotFoundError: + return None + if os.path.islink(path) or not os.path.isfile(path): + raise ValueError("数据库源必须是普通文件且不能是符号链接") + return (stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns) + + +def _cleanup_stale_runs(cache_root): + for name in os.listdir(cache_root): + if not _RUN_DIR_RE.fullmatch(name): + continue + run_dir = os.path.join(cache_root, name) + if os.path.islink(run_dir) or not os.path.isdir(run_dir): + continue + lease_path = os.path.join(run_dir, ".lease") try: - with open(self.MTIME_FILE, encoding="utf-8") as f: - saved = json.load(f) - except (json.JSONDecodeError, OSError): - return - for rel_key, info in saved.items(): - tmp_path = info["path"] - if not os.path.exists(tmp_path): - continue - rel_path = rel_key.replace('\\', os.sep) - db_path = os.path.join(self._db_dir, rel_path) - wal_path = db_path + "-wal" - try: - db_mtime = os.path.getmtime(db_path) - wal_mtime = os.path.getmtime(wal_path) if os.path.exists(wal_path) else 0 - except OSError: - continue - if db_mtime == info["db_mt"] and wal_mtime == info["wal_mt"]: - self._cache[rel_key] = (db_mtime, wal_mtime, tmp_path) - - def _save_persistent_cache(self): - data = {} - for rel_key, (db_mt, wal_mt, path) in self._cache.items(): - data[rel_key] = {"db_mt": db_mt, "wal_mt": wal_mt, "path": path} + descriptor = os.open(lease_path, os.O_CREAT | os.O_RDWR, 0o600) + except OSError: + continue + acquired = False + closed = False try: - with open(self.MTIME_FILE, 'w', encoding="utf-8") as f: - json.dump(data, f) + _lock_descriptor(descriptor, nonblocking=True) + acquired = True + if os.name == "nt": + # UUID 运行目录不会被复用；确认无人持锁后可先关闭再删除。 + _unlock_descriptor(descriptor) + os.close(descriptor) + closed = True + shutil.rmtree(run_dir) except OSError: pass + finally: + if acquired and not closed: + _unlock_descriptor(descriptor) + if not closed: + os.close(descriptor) + + +class DBCache: + def __init__(self, key_provider: DatabaseKeyProvider, db_dir, state_dir): + self._key_provider = key_provider + self._db_dir = os.path.realpath(db_dir) + self._cache = {} # rel_key -> (db_sig, wal_sig, key_digest, tmp_path) + cache_root = os.path.join(state_dir, "cache") + ensure_private_directory(cache_root, tighten_existing=True) + _cleanup_stale_runs(cache_root) + self._cache_dir = os.path.join( + cache_root, f"run-{os.getpid()}-{uuid.uuid4().hex}" + ) + os.mkdir(self._cache_dir, 0o700) + self._lease_descriptor = os.open( + os.path.join(self._cache_dir, ".lease"), os.O_CREAT | os.O_RDWR, 0o600 + ) + _lock_descriptor(self._lease_descriptor, nonblocking=False) + + def _cache_path(self, rel_key): + digest = hashlib.sha256(rel_key.encode()).hexdigest()[:24] + return os.path.join(self._cache_dir, f"{digest}.db") + + def _source_paths(self, rel_key): + if not _is_safe_rel_path(rel_key): + raise ValueError("数据库相对路径不安全") + rel_path = rel_key.replace("\\", "/").replace("/", os.sep) + db_path = os.path.realpath(os.path.join(self._db_dir, rel_path)) + try: + contained = os.path.commonpath((self._db_dir, db_path)) == self._db_dir + except ValueError: + contained = False + if not contained: + raise ValueError("数据库路径越出配置目录") + return db_path, db_path + "-wal" + + def _stable_snapshot(self, db_path, wal_path, token): + snapshot_db = os.path.join(self._cache_dir, f".{token}.encrypted-db") + snapshot_wal = os.path.join(self._cache_dir, f".{token}.encrypted-wal") + for attempt in range(_SNAPSHOT_ATTEMPTS): + before_db = _source_signature(db_path) + before_wal = _source_signature(wal_path) + if before_db is None: + return None + try: + shutil.copyfile(db_path, snapshot_db) + if before_wal is not None: + shutil.copyfile(wal_path, snapshot_wal) + elif os.path.exists(snapshot_wal): + os.remove(snapshot_wal) + after_db = _source_signature(db_path) + after_wal = _source_signature(wal_path) + if before_db == after_db and before_wal == after_wal: + return ( + snapshot_db, + snapshot_wal if before_wal is not None else None, + before_db, + before_wal, + ) + except FileNotFoundError: + pass + if attempt + 1 < _SNAPSHOT_ATTEMPTS: + time.sleep(0.05) + for path in (snapshot_db, snapshot_wal): + try: + os.remove(path) + except FileNotFoundError: + pass + raise RuntimeError("数据库或 WAL 持续变化，无法取得一致快照；请稍后重试") def get(self, rel_key): - key_info = get_key_info(self._all_keys, rel_key) + key_info = self._key_provider.resolve(rel_key) if not key_info: return None - rel_path = rel_key.replace('\\', '/').replace('/', os.sep) - db_path = os.path.join(self._db_dir, rel_path) - wal_path = db_path + "-wal" - if not os.path.exists(db_path): - return None - + db_path, wal_path = self._source_paths(rel_key) try: - db_mtime = os.path.getmtime(db_path) - wal_mtime = os.path.getmtime(wal_path) if os.path.exists(wal_path) else 0 + db_sig = _source_signature(db_path) + wal_sig = _source_signature(wal_path) except OSError: return None + if db_sig is None: + return None + enc_key = bytes.fromhex(key_info["enc_key"]) + key_digest = hashlib.sha256(enc_key).digest() if rel_key in self._cache: - c_db_mt, c_wal_mt, c_path = self._cache[rel_key] - if c_db_mt == db_mtime and c_wal_mt == wal_mtime and os.path.exists(c_path): - return c_path + cached_db, cached_wal, cached_key, cached_path = self._cache[rel_key] + if ( + cached_db == db_sig + and cached_wal == wal_sig + and cached_key == key_digest + and os.path.exists(cached_path) + ): + return cached_path tmp_path = self._cache_path(rel_key) - enc_key = bytes.fromhex(key_info["enc_key"]) - full_decrypt(db_path, tmp_path, enc_key) - if os.path.exists(wal_path): - decrypt_wal(wal_path, tmp_path, enc_key) - self._cache[rel_key] = (db_mtime, wal_mtime, tmp_path) - self._save_persistent_cache() + token = hashlib.sha256((rel_key + uuid.uuid4().hex).encode()).hexdigest()[:24] + snapshot = self._stable_snapshot(db_path, wal_path, token) + if snapshot is None: + return None + snapshot_db, snapshot_wal, db_sig, wal_sig = snapshot + try: + with open(snapshot_db, "rb") as source: + database_salt = source.read(16) + configured_salt = key_info.get("salt") + if configured_salt and database_salt.hex() != configured_salt.lower(): + raise ValueError("数据库 salt 与密钥引用不匹配") + if os.path.exists(tmp_path): + os.remove(tmp_path) + descriptor = os.open(tmp_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + os.close(descriptor) + full_decrypt(snapshot_db, tmp_path, enc_key) + if snapshot_wal: + decrypt_wal(snapshot_wal, tmp_path, enc_key, database_salt) + except Exception: + try: + os.remove(tmp_path) + except OSError: + pass + raise + finally: + for path in (snapshot_db, snapshot_wal): + if path: + try: + os.remove(path) + except FileNotFoundError: + pass + self._cache[rel_key] = (db_sig, wal_sig, key_digest, tmp_path) return tmp_path def cleanup(self): - self._save_persistent_cache() + self._cache.clear() + descriptor = getattr(self, "_lease_descriptor", None) + if descriptor is not None: + _unlock_descriptor(descriptor) + os.close(descriptor) + self._lease_descriptor = None + if os.path.isdir(self._cache_dir): + shutil.rmtree(self._cache_dir) diff --git a/wechat_cli/core/key_document.py b/wechat_cli/core/key_document.py new file mode 100644 index 0000000..9b923ea --- /dev/null +++ b/wechat_cli/core/key_document.py @@ -0,0 +1,524 @@ +"""微信数据库密钥文档及系统密钥后端。""" + +import base64 +import json +import os +import platform +import re +import subprocess +import uuid +from abc import ABC, abstractmethod + +from .key_utils import _is_safe_rel_path +from .secure_files import atomic_write_json, read_private_json + + +_KEY_HEX = re.compile(r"^[0-9a-fA-F]{64}$") +_SALT_HEX = re.compile(r"^[0-9a-fA-F]{32}$") + + +def validate_key_document(document, source="密钥文档"): + """返回规范化副本；拒绝含糊条目且永不在错误中包含值。""" + if not isinstance(document, dict): + raise ValueError(f"{source} 必须是对象") + + normalized = {} + for rel_path, info in document.items(): + if not isinstance(rel_path, str): + raise ValueError(f"{source} 包含非字符串路径") + if rel_path.startswith("_"): + continue + if not rel_path or not _is_safe_rel_path(rel_path): + raise ValueError(f"{source} 包含不安全的数据库相对路径") + if not isinstance(info, dict): + raise ValueError(f"{source} 中 {rel_path!r} 的条目必须是对象") + + enc_key = info.get("enc_key") + if not isinstance(enc_key, str) or not _KEY_HEX.fullmatch(enc_key): + raise ValueError(f"{source} 中 {rel_path!r} 的 enc_key 格式无效") + salt = info.get("salt") + if salt is not None and (not isinstance(salt, str) or not _SALT_HEX.fullmatch(salt)): + raise ValueError(f"{source} 中 {rel_path!r} 的 salt 格式无效") + + normalized[rel_path] = dict(info) + + if not normalized: + raise ValueError(f"{source} 没有可用的数据库密钥") + return normalized + + +class DatabaseKeyProvider(ABC): + """数据库密钥引用边界；调用方不能枚举或读取无关密钥值。""" + + @abstractmethod + def references(self): + """返回可用数据库相对路径，不返回密钥值。""" + + @abstractmethod + def resolve(self, rel_path): + """按一次操作解析一个数据库的密钥信息。""" + + @abstractmethod + def replace(self, document): + """原子替换 Provider 管理的完整密钥集合。""" + + @abstractmethod + def describe(self, rel_path): + """返回配置状态和来源，不返回密钥值。""" + + +class JsonDatabaseKeyProvider(DatabaseKeyProvider): + """兼容 `all_keys.json` 的文件 Provider；每次操作重新读取。""" + + def __init__(self, path): + self.path = path + + def _load(self): + return validate_key_document(read_private_json(self.path), self.path) + + def references(self): + return tuple(self._load()) + + def resolve(self, rel_path): + if not isinstance(rel_path, str) or not _is_safe_rel_path(rel_path): + return None + document = self._load() + for candidate in ( + rel_path, + rel_path.replace("\\", "/"), + rel_path.replace("/", "\\"), + ): + if candidate in document: + return dict(document[candidate]) + return None + + def replace(self, document): + normalized = validate_key_document(document) + atomic_write_json(self.path, normalized) + + def describe(self, rel_path): + return { + "configured": self.resolve(rel_path) is not None, + "source": "file", + "writable": True, + } + + +def _dpapi_transform(data, *, protect): + if os.name != "nt": + raise RuntimeError("DPAPI 仅支持 Windows") + import ctypes + from ctypes import wintypes + + class DataBlob(ctypes.Structure): + _fields_ = [("cbData", wintypes.DWORD), ("pbData", ctypes.POINTER(ctypes.c_byte))] + + source_buffer = ctypes.create_string_buffer(data) + source = DataBlob(len(data), ctypes.cast(source_buffer, ctypes.POINTER(ctypes.c_byte))) + destination = DataBlob() + description = wintypes.LPWSTR() + crypt32 = ctypes.WinDLL("crypt32", use_last_error=True) + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + crypt32.CryptProtectData.argtypes = [ + ctypes.POINTER(DataBlob), wintypes.LPCWSTR, ctypes.POINTER(DataBlob), + ctypes.c_void_p, ctypes.c_void_p, wintypes.DWORD, ctypes.POINTER(DataBlob), + ] + crypt32.CryptProtectData.restype = wintypes.BOOL + crypt32.CryptUnprotectData.argtypes = [ + ctypes.POINTER(DataBlob), ctypes.POINTER(wintypes.LPWSTR), ctypes.POINTER(DataBlob), + ctypes.c_void_p, ctypes.c_void_p, wintypes.DWORD, ctypes.POINTER(DataBlob), + ] + crypt32.CryptUnprotectData.restype = wintypes.BOOL + kernel32.LocalFree.argtypes = [ctypes.c_void_p] + kernel32.LocalFree.restype = ctypes.c_void_p + flags = 0x1 # CRYPTPROTECT_UI_FORBIDDEN + if protect: + ok = crypt32.CryptProtectData( + ctypes.byref(source), "wechat-cli", None, None, None, flags, + ctypes.byref(destination), + ) + else: + ok = crypt32.CryptUnprotectData( + ctypes.byref(source), ctypes.byref(description), None, None, None, flags, + ctypes.byref(destination), + ) + if not ok: + raise OSError(ctypes.get_last_error(), "DPAPI 操作失败") + try: + return ctypes.string_at(destination.pbData, destination.cbData) + finally: + if destination.pbData: + kernel32.LocalFree(ctypes.cast(destination.pbData, ctypes.c_void_p)) + if description: + kernel32.LocalFree(ctypes.cast(description, ctypes.c_void_p)) + + +class DpapiDatabaseKeyProvider(DatabaseKeyProvider): + """Windows CurrentUser DPAPI；清单只保存密文。""" + + backend = "dpapi-current-user" + + def __init__(self, path): + self.path = path + + def _load(self): + document = read_private_json(self.path) + if not isinstance(document, dict) or document.get("_backend") != self.backend: + raise ValueError("DPAPI 密钥清单格式无效") + entries = document.get("_entries") + if not isinstance(entries, dict) or not entries: + raise ValueError("DPAPI 密钥清单没有条目") + for rel_path, blob in entries.items(): + if not isinstance(rel_path, str) or not _is_safe_rel_path(rel_path): + raise ValueError("DPAPI 密钥清单包含不安全路径") + if not isinstance(blob, str): + raise ValueError("DPAPI 密钥清单包含无效密文") + return entries + + def references(self): + return tuple(self._load()) + + def resolve(self, rel_path): + if not isinstance(rel_path, str) or not _is_safe_rel_path(rel_path): + return None + entries = self._load() + for candidate in (rel_path, rel_path.replace("\\", "/"), rel_path.replace("/", "\\")): + blob = entries.get(candidate) + if blob is None: + continue + try: + encrypted = base64.b64decode(blob, validate=True) + payload = _dpapi_transform(encrypted, protect=False) + info = json.loads(payload.decode("utf-8")) + except Exception: + raise ValueError("DPAPI 密钥条目无法解密或校验") from None + return validate_key_document({candidate: info}, "DPAPI 密钥条目")[candidate] + return None + + def replace(self, document): + normalized = validate_key_document(document) + entries = {} + for rel_path, info in normalized.items(): + payload = json.dumps(info, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + encrypted = _dpapi_transform(payload, protect=True) + entries[rel_path] = base64.b64encode(encrypted).decode("ascii") + atomic_write_json(self.path, {"_backend": self.backend, "_entries": entries}) + + def describe(self, rel_path): + return { + "configured": self.resolve(rel_path) is not None, + "source": self.backend, + "writable": True, + } + + +class SecretServiceDatabaseKeyProvider(DatabaseKeyProvider): + """Linux Secret Service 后端；密钥通过 stdin 交给 secret-tool。""" + + backend = "secret-service" + tool_path = "/usr/bin/secret-tool" + + def __init__(self, path, namespace): + self.path = path + self.namespace = namespace + if not isinstance(namespace, str) or not _GENERATION_TOKEN.fullmatch(namespace): + raise ValueError("Secret Service namespace 无效") + + def _command(self, *args, input_text=None): + if not os.path.isfile(self.tool_path): + raise RuntimeError("未安装 /usr/bin/secret-tool") + result = subprocess.run( + [self.tool_path, *args], + input=input_text, + capture_output=True, + text=True, + timeout=30, + check=False, + ) + if result.returncode != 0: + raise RuntimeError("Secret Service 操作失败") + return result.stdout + + def _load(self): + document = read_private_json(self.path) + if not isinstance(document, dict) or document.get("_backend") != self.backend: + raise ValueError("Secret Service 密钥清单格式无效") + if document.get("_namespace") != self.namespace: + raise ValueError("Secret Service namespace 不匹配") + entries = document.get("_entries") + if not isinstance(entries, dict) or not entries: + raise ValueError("Secret Service 密钥清单没有条目") + for rel_path, token in entries.items(): + if not isinstance(rel_path, str) or not _is_safe_rel_path(rel_path): + raise ValueError("Secret Service 密钥清单包含不安全路径") + if not isinstance(token, str) or not _GENERATION_TOKEN.fullmatch(token): + raise ValueError("Secret Service 密钥清单包含无效引用") + return entries + + def _attributes(self, token): + return ["application", "wechat-cli", "namespace", self.namespace, "entry", token] + + def references(self): + return tuple(self._load()) + + def resolve(self, rel_path): + if not isinstance(rel_path, str) or not _is_safe_rel_path(rel_path): + return None + entries = self._load() + for candidate in (rel_path, rel_path.replace("\\", "/"), rel_path.replace("/", "\\")): + token = entries.get(candidate) + if token is None: + continue + payload = self._command("lookup", *self._attributes(token)) + try: + info = json.loads(payload) + except json.JSONDecodeError: + raise ValueError("Secret Service 密钥条目格式无效") from None + return validate_key_document({candidate: info}, "Secret Service 密钥条目")[candidate] + return None + + def replace(self, document): + normalized = validate_key_document(document) + previous = {} + if os.path.exists(self.path): + previous = self._load() + created = {} + try: + for rel_path, info in normalized.items(): + token = uuid.uuid4().hex + payload = json.dumps(info, ensure_ascii=False, separators=(",", ":")) + self._command( + "store", "--label=WeChat CLI database key", *self._attributes(token), + input_text=payload, + ) + created[rel_path] = token + atomic_write_json( + self.path, + {"_backend": self.backend, "_namespace": self.namespace, "_entries": created}, + ) + except Exception: + for token in created.values(): + try: + self._command("clear", *self._attributes(token)) + except Exception: + pass + raise + for token in previous.values(): + try: + self._command("clear", *self._attributes(token)) + except Exception: + pass + + def describe(self, rel_path): + return { + "configured": self.resolve(rel_path) is not None, + "source": self.backend, + "writable": True, + } + + +class MacKeychainDatabaseKeyProvider(DatabaseKeyProvider): + """macOS login Keychain；清单只保存随机引用。""" + + backend = "macos-keychain" + + def __init__(self, path, namespace): + self.path = path + self.namespace = namespace + if not isinstance(namespace, str) or not _GENERATION_TOKEN.fullmatch(namespace): + raise ValueError("Keychain namespace 无效") + + def _security(self): + if platform.system().lower() != "darwin": + raise RuntimeError("macOS Keychain 仅支持 macOS") + import ctypes + + security = ctypes.CDLL( + "/System/Library/Frameworks/Security.framework/Security" + ) + core_foundation = ctypes.CDLL( + "/System/Library/Frameworks/CoreFoundation.framework/CoreFoundation" + ) + security.SecKeychainAddGenericPassword.argtypes = [ + ctypes.c_void_p, ctypes.c_uint32, ctypes.c_char_p, ctypes.c_uint32, + ctypes.c_char_p, ctypes.c_uint32, ctypes.c_void_p, + ctypes.POINTER(ctypes.c_void_p), + ] + security.SecKeychainAddGenericPassword.restype = ctypes.c_int32 + security.SecKeychainFindGenericPassword.argtypes = [ + ctypes.c_void_p, ctypes.c_uint32, ctypes.c_char_p, ctypes.c_uint32, + ctypes.c_char_p, ctypes.POINTER(ctypes.c_uint32), + ctypes.POINTER(ctypes.c_void_p), ctypes.POINTER(ctypes.c_void_p), + ] + security.SecKeychainFindGenericPassword.restype = ctypes.c_int32 + security.SecKeychainItemFreeContent.argtypes = [ctypes.c_void_p, ctypes.c_void_p] + security.SecKeychainItemFreeContent.restype = ctypes.c_int32 + security.SecKeychainItemDelete.argtypes = [ctypes.c_void_p] + security.SecKeychainItemDelete.restype = ctypes.c_int32 + core_foundation.CFRelease.argtypes = [ctypes.c_void_p] + return ctypes, security, core_foundation + + def _names(self, token): + return f"wechat-cli:{self.namespace}".encode(), token.encode() + + def _store(self, token, payload): + ctypes, security, core_foundation = self._security() + service, account = self._names(token) + item = ctypes.c_void_p() + data = payload.encode("utf-8") + data_buffer = ctypes.create_string_buffer(data) + status = security.SecKeychainAddGenericPassword( + None, len(service), service, len(account), account, len(data), + ctypes.cast(data_buffer, ctypes.c_void_p), + ctypes.byref(item), + ) + if item: + core_foundation.CFRelease(item) + if status != 0: + raise RuntimeError(f"Keychain 写入失败 (OSStatus {status})") + + def _find(self, token): + ctypes, security, core_foundation = self._security() + service, account = self._names(token) + length = ctypes.c_uint32() + data = ctypes.c_void_p() + item = ctypes.c_void_p() + status = security.SecKeychainFindGenericPassword( + None, len(service), service, len(account), account, + ctypes.byref(length), ctypes.byref(data), ctypes.byref(item), + ) + if status != 0: + raise RuntimeError(f"Keychain 读取失败 (OSStatus {status})") + try: + return ctypes.string_at(data, length.value), item + except Exception: + if item: + core_foundation.CFRelease(item) + raise + finally: + if data: + security.SecKeychainItemFreeContent(None, data) + + def _lookup(self, token): + payload, item = self._find(token) + try: + return payload.decode("utf-8") + finally: + if item: + _ctypes, _security, core_foundation = self._security() + core_foundation.CFRelease(item) + + def _clear(self, token): + _payload, item = self._find(token) + try: + status = self._security()[1].SecKeychainItemDelete(item) + if status != 0: + raise RuntimeError(f"Keychain 删除失败 (OSStatus {status})") + finally: + if item: + self._security()[2].CFRelease(item) + + def _load(self): + document = read_private_json(self.path) + if not isinstance(document, dict) or document.get("_backend") != self.backend: + raise ValueError("Keychain 密钥清单格式无效") + if document.get("_namespace") != self.namespace: + raise ValueError("Keychain namespace 不匹配") + entries = document.get("_entries") + if not isinstance(entries, dict) or not entries: + raise ValueError("Keychain 密钥清单没有条目") + for rel_path, token in entries.items(): + if not isinstance(rel_path, str) or not _is_safe_rel_path(rel_path): + raise ValueError("Keychain 密钥清单包含不安全路径") + if not isinstance(token, str) or not _GENERATION_TOKEN.fullmatch(token): + raise ValueError("Keychain 密钥清单包含无效引用") + return entries + + def references(self): + return tuple(self._load()) + + def resolve(self, rel_path): + if not isinstance(rel_path, str) or not _is_safe_rel_path(rel_path): + return None + entries = self._load() + for candidate in (rel_path, rel_path.replace("\\", "/"), rel_path.replace("/", "\\")): + token = entries.get(candidate) + if token is None: + continue + try: + info = json.loads(self._lookup(token)) + except json.JSONDecodeError: + raise ValueError("Keychain 密钥条目格式无效") from None + return validate_key_document({candidate: info}, "Keychain 密钥条目")[candidate] + return None + + def replace(self, document): + normalized = validate_key_document(document) + previous = self._load() if os.path.exists(self.path) else {} + created = {} + try: + for rel_path, info in normalized.items(): + token = uuid.uuid4().hex + payload = json.dumps(info, ensure_ascii=False, separators=(",", ":")) + self._store(token, payload) + created[rel_path] = token + atomic_write_json( + self.path, + {"_backend": self.backend, "_namespace": self.namespace, "_entries": created}, + ) + except Exception: + for token in created.values(): + try: + self._clear(token) + except Exception: + pass + raise + for token in previous.values(): + try: + self._clear(token) + except Exception: + pass + + def describe(self, rel_path): + return { + "configured": self.resolve(rel_path) is not None, + "source": self.backend, + "writable": True, + } + + +_GENERATION_TOKEN = re.compile(r"^[0-9a-f]{32}$") + + +def select_default_key_backend(): + system = platform.system().lower() + if system == "windows": + return DpapiDatabaseKeyProvider.backend + if system == "linux" and os.path.isfile(SecretServiceDatabaseKeyProvider.tool_path): + return SecretServiceDatabaseKeyProvider.backend + if system == "darwin": + return MacKeychainDatabaseKeyProvider.backend + if os.environ.get("WECHAT_CLI_ALLOW_PLAINTEXT_KEYS") == "1": + return "file" + raise RuntimeError( + "未找到系统密钥库；安装 secret-tool，或显式设置 WECHAT_CLI_ALLOW_PLAINTEXT_KEYS=1" + ) + + +def create_key_provider(path, backend=None, namespace=None): + if backend is None and os.path.exists(path): + document = read_private_json(path) + backend = document.get("_backend") if isinstance(document, dict) else None + if backend is None and os.environ.get("WECHAT_CLI_ALLOW_PLAINTEXT_KEYS") != "1": + raise RuntimeError( + "检测到旧版明文密钥文件；请运行 wechat-cli init --force 迁移到系统密钥库" + ) + if backend in (None, "file"): + return JsonDatabaseKeyProvider(path) + if backend == DpapiDatabaseKeyProvider.backend: + return DpapiDatabaseKeyProvider(path) + if backend == SecretServiceDatabaseKeyProvider.backend: + return SecretServiceDatabaseKeyProvider(path, namespace) + if backend == MacKeychainDatabaseKeyProvider.backend: + return MacKeychainDatabaseKeyProvider(path, namespace) + raise ValueError("不支持的密钥后端") diff --git a/wechat_cli/core/key_utils.py b/wechat_cli/core/key_utils.py index e3b8f31..05393f6 100644 --- a/wechat_cli/core/key_utils.py +++ b/wechat_cli/core/key_utils.py @@ -1,6 +1,7 @@ """密钥工具 — 路径匹配、元数据剥离""" import os +import ntpath import posixpath @@ -9,8 +10,16 @@ def strip_key_metadata(keys): def _is_safe_rel_path(path): + if not isinstance(path, str) or not path or "\x00" in path: + return False normalized = path.replace("\\", "/") - return ".." not in posixpath.normpath(normalized).split("/") + components = normalized.split("/") + return ( + not ntpath.splitdrive(path)[0] + and not posixpath.isabs(normalized) + and not ntpath.isabs(path) + and all(component not in ("", ".", "..") for component in components) + ) def key_path_variants(rel_path): diff --git a/wechat_cli/core/messages.py b/wechat_cli/core/messages.py index d62ef33..75e2e02 100644 --- a/wechat_cli/core/messages.py +++ b/wechat_cli/core/messages.py @@ -4,17 +4,15 @@ import os import re import sqlite3 -import xml.etree.ElementTree as ET from contextlib import closing from datetime import datetime import zstandard as zstd from .key_utils import key_path_variants +from .xml_utils import parse_untrusted_xml _zstd_dctx = zstd.ZstdDecompressor() -_XML_UNSAFE_RE = re.compile(r' _XML_PARSE_MAX_LEN or _XML_UNSAFE_RE.search(content): - return None - try: - return ET.fromstring(content) - except ET.ParseError: - return None + return parse_untrusted_xml(content) def _parse_int(value, fallback=0): diff --git a/wechat_cli/core/secure_files.py b/wechat_cli/core/secure_files.py new file mode 100644 index 0000000..34c5771 --- /dev/null +++ b/wechat_cli/core/secure_files.py @@ -0,0 +1,154 @@ +"""私有状态文件：严格权限、原子替换和无密钥诊断。""" + +import json +import os +import re +import stat +import tempfile +from subprocess import run as _run + + +PRIVATE_DIR_MODE = 0o700 +PRIVATE_FILE_MODE = 0o600 +_SECRET_PATTERNS = ( + re.compile(r"(?i)(enc_key\s*[=:]\s*)[0-9a-f]{64,192}"), + re.compile(r"(?i)(\"enc_key\"\s*:\s*\")[0-9a-f]{64,192}(\")"), + re.compile(r"(?i)x'[0-9a-f]{64,192}'"), + re.compile(r"(?i)(?= deadline: + raise TimeoutError("另一个 wechat-cli 状态事务仍在运行") from None + time.sleep(0.05) + yield + finally: + if acquired: + if os.name == "nt": + import msvcrt + + os.lseek(descriptor, 0, os.SEEK_SET) + msvcrt.locking(descriptor, msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(descriptor, fcntl.LOCK_UN) + os.close(descriptor) + + +def commit_generation(state_dir, config, key_document, backend=None): + """完整物化新 generation 后，以单文件原子切换激活。""" + from .key_document import validate_key_document + + key_document = validate_key_document(key_document, "初始化密钥文档") + generation_root = _generation_root(state_dir) + ensure_private_directory(generation_root) + generation_id = uuid.uuid4().hex + pending = tempfile.mkdtemp(prefix=f".{generation_id}.", dir=generation_root) + final = os.path.join(generation_root, generation_id) + try: + ensure_private_directory(pending, tighten_existing=True) + from .key_document import create_key_provider, select_default_key_backend + + backend = backend or select_default_key_backend() + key_path = os.path.join(pending, "all_keys.json") + provider = create_key_provider(key_path, backend=backend, namespace=generation_id) + provider.replace(key_document) + generation_config = dict(config) + generation_config["keys_file"] = "all_keys.json" + generation_config["generation"] = generation_id + generation_config["key_backend"] = backend + if backend == "secret-service": + generation_config["key_namespace"] = generation_id + atomic_write_json(os.path.join(pending, "config.json"), generation_config) + + # 完整读回后才允许换入。 + if set(provider.references()) != set(key_document): + raise ValueError("密钥后端读回条目不完整") + for rel_path in key_document: + if provider.resolve(rel_path) != key_document[rel_path]: + raise ValueError("密钥后端读回校验失败") + read_private_json(os.path.join(pending, "config.json")) + os.replace(pending, final) + atomic_write_json( + active_pointer_path(state_dir), + {"version": 1, "generation": generation_id}, + ) + return final + finally: + if os.path.isdir(pending): + shutil.rmtree(pending) diff --git a/wechat_cli/core/xml_utils.py b/wechat_cli/core/xml_utils.py new file mode 100644 index 0000000..ae0559c --- /dev/null +++ b/wechat_cli/core/xml_utils.py @@ -0,0 +1,19 @@ +"""不可信微信 XML 的有界解析。""" + +import re +import xml.etree.ElementTree as ET + + +XML_PARSE_MAX_LEN = 20_000 +_UNSAFE_DECLARATION = re.compile(r" max_length: + return None + if _UNSAFE_DECLARATION.search(content): + return None + try: + return ET.fromstring(content) + except (ET.ParseError, RecursionError): + return None diff --git a/wechat_cli/keys/__init__.py b/wechat_cli/keys/__init__.py index 602079d..4627f72 100644 --- a/wechat_cli/keys/__init__.py +++ b/wechat_cli/keys/__init__.py @@ -1,6 +1,14 @@ -"""密钥提取模块 — 根据平台调用对应的 scanner""" +"""密钥提取模块 — 根据平台调用对应的 scanner。""" +import json +import os import platform +import subprocess +import sys + +from ..core.key_document import validate_key_document +from ..core.secure_files import redact_sensitive_text +from .capabilities import require_supported_memory_scan def extract_keys(db_dir, output_path, pid=None): @@ -17,6 +25,7 @@ def extract_keys(db_dir, output_path, pid=None): Raises: RuntimeError: 提取失败 """ + require_supported_memory_scan() system = platform.system().lower() if system == "darwin": from .scanner_macos import extract_keys as _extract @@ -29,3 +38,56 @@ def extract_keys(db_dir, output_path, pid=None): return _extract(db_dir, output_path, pid=pid) else: raise RuntimeError(f"不支持的平台: {platform.system()}") + + +def extract_keys_for_init(db_dir): + """以最小权限提取密钥，返回经过校验的密钥文档但不负责持久化。""" + require_supported_memory_scan() + system = platform.system().lower() + if system == "windows": + from tempfile import TemporaryDirectory + + from ..core.secure_files import read_private_json + + with TemporaryDirectory(prefix="wechat-cli-scan-") as temp_dir: + output_path = os.path.join(temp_dir, "all_keys.json") + extract_keys(db_dir, output_path) + return validate_key_document(read_private_json(output_path), "扫描结果") + + if system not in {"darwin", "linux"}: + raise RuntimeError(f"不支持的平台: {platform.system()}") + if hasattr(os, "geteuid") and os.geteuid() == 0 and os.environ.get("SUDO_USER"): + raise RuntimeError("请以普通用户运行 wechat-cli init；工具会仅为内存扫描请求 sudo") + + sudo_path = "/usr/bin/sudo" + if not os.path.isfile(sudo_path): + raise RuntimeError("找不到 /usr/bin/sudo，无法启动最小权限扫描 helper") + + if getattr(sys, "frozen", False): + command = [sudo_path, "--", sys.executable, "_elevated-scan-helper", db_dir] + else: + command = [ + sudo_path, "--", sys.executable, "-m", + "wechat_cli.keys.elevated_helper", db_dir, + ] + try: + result = subprocess.run( + command, + capture_output=True, + text=True, + timeout=300, + check=False, + ) + except subprocess.TimeoutExpired: + raise RuntimeError("提权扫描超时") from None + + diagnostic = redact_sensitive_text(result.stderr or "").strip() + if diagnostic: + print(diagnostic, file=sys.stderr) + if result.returncode != 0: + raise RuntimeError("提权扫描失败；请检查上方已脱敏诊断") + try: + document = json.loads(result.stdout) + except json.JSONDecodeError: + raise RuntimeError("提权扫描返回了无效结果") from None + return validate_key_document(document, "提权扫描结果") diff --git a/wechat_cli/keys/capabilities.py b/wechat_cli/keys/capabilities.py new file mode 100644 index 0000000..dc1ba25 --- /dev/null +++ b/wechat_cli/keys/capabilities.py @@ -0,0 +1,164 @@ +"""内存扫描能力清单与只读预检。 + +清单只允许精确匹配已经完成真实客户端验收的组合。未知版本必须在申请 +进程内存读取权限之前失败关闭。 +""" + +import json +import os +import platform +import subprocess +from dataclasses import asdict, dataclass + + +@dataclass(frozen=True) +class ScannerProfile: + profile_id: str + platform: str + executable: str + client_version: str + publisher: str + scanner_algorithm: str + acceptance: str + + +# 安全默认值：尚无任何客户端版本完成平台真实验收。 +# 新条目必须绑定精确版本、发布者和算法，并把 acceptance 设为 verified。 +SCANNER_PROFILES: tuple[ScannerProfile, ...] = () + + +def _process_running_without_memory_access(system): + """只查询进程列表，不申请进程句柄或读取进程内存。""" + try: + if system == "windows": + tasklist = os.path.join( + os.environ.get("SystemRoot", r"C:\Windows"), "System32", "tasklist.exe" + ) + result = subprocess.run( + [tasklist, "/FI", "IMAGENAME eq Weixin.exe", "/FO", "CSV", "/NH"], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + return "Weixin.exe" in result.stdout + executable = "WeChat" if system == "darwin" else "wechat" + result = subprocess.run( + ["/usr/bin/pgrep", "-x", executable], + capture_output=True, + timeout=10, + check=False, + ) + return result.returncode == 0 + except (OSError, subprocess.SubprocessError): + return False + + +def _windows_signed_identity(): + """读取进程路径对应文件的版本和 Authenticode 签名，不读取进程内存。""" + windows_powershell = os.path.join( + os.environ.get("SystemRoot", r"C:\Windows"), + "System32", + "WindowsPowerShell", + "v1.0", + "powershell.exe", + ) + powershell = windows_powershell + script = ( + "$p=Get-Process -Name Weixin -ErrorAction SilentlyContinue|" + "Select-Object -First 1;" + "if($null -ne $p -and $p.Path){" + "$f=Get-Item -LiteralPath $p.Path;" + "$s=Get-AuthenticodeSignature -LiteralPath $p.Path;" + "$u=$(if($null -ne $s.SignerCertificate){$s.SignerCertificate.Subject}else{'unknown'});" + "[pscustomobject]@{version=$f.VersionInfo.FileVersion;publisher=$u;" + "signature_status=$($s.Status.ToString())}|ConvertTo-Json -Compress}" + ) + try: + child_env = os.environ.copy() + # PowerShell 7 的模块路径会使 Windows PowerShell 5.1 加载不兼容模块。 + for name in tuple(child_env): + if name.lower() in {"psmodulepath", "pshome"}: + child_env.pop(name, None) + result = subprocess.run( + [powershell, "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", script], + capture_output=True, + text=True, + timeout=15, + check=False, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + env=child_env, + ) + if result.returncode != 0 or not result.stdout.strip(): + return None + document = json.loads(result.stdout) + version = document.get("version") + publisher = document.get("publisher") + status = document.get("signature_status") + if not isinstance(version, str) or not isinstance(publisher, str): + return None + return { + "client_version": version.strip() or "unknown", + "publisher": publisher.strip() or "unknown", + "signature_valid": status == "Valid", + } + except (OSError, subprocess.SubprocessError, json.JSONDecodeError): + return None + + +def detect_client_identity(): + """返回不含密钥且不读进程内存的客户端身份摘要。""" + system = platform.system().lower() + executable = {"windows": "Weixin.exe", "darwin": "WeChat", "linux": "wechat"}.get( + system, "unknown" + ) + identity = { + "platform": system, + "executable": executable, + "client_version": "unknown", + "publisher": "unknown", + "signature_valid": False, + "process_running": _process_running_without_memory_access(system), + } + if system == "windows" and identity["process_running"]: + signed = _windows_signed_identity() + if signed: + identity.update(signed) + return identity + + +def evaluate_scan_capability(identity=None, profiles=SCANNER_PROFILES): + identity = dict(identity or detect_client_identity()) + for profile in profiles: + if ( + profile.acceptance == "verified" + and identity.get("platform") == profile.platform + and identity.get("executable") == profile.executable + and identity.get("client_version") == profile.client_version + and identity.get("publisher") == profile.publisher + and identity.get("signature_valid") is True + ): + return { + "supported": True, + "reason_code": "WXCLI_SCAN_PROFILE_VERIFIED", + "profile": asdict(profile), + "identity": identity, + } + return { + "supported": False, + "reason_code": "WXCLI_SCAN_UNSUPPORTED_CLIENT", + "profile": None, + "identity": identity, + } + + +def require_supported_memory_scan(): + result = evaluate_scan_capability() + if not result["supported"]: + identity = result["identity"] + raise RuntimeError( + "[WXCLI_SCAN_UNSUPPORTED_CLIENT] 当前客户端没有通过内存扫描验收：" + f"platform={identity['platform']}, executable={identity['executable']}, " + f"version={identity['client_version']}, publisher={identity['publisher']}" + ) + return result diff --git a/wechat_cli/keys/common.py b/wechat_cli/keys/common.py index c281b67..ca103f2 100644 --- a/wechat_cli/keys/common.py +++ b/wechat_cli/keys/common.py @@ -6,11 +6,13 @@ import hashlib import hmac as hmac_mod -import json import os import re import struct +from ..core.key_document import JsonDatabaseKeyProvider +from ..core.key_utils import _is_safe_rel_path + PAGE_SZ = 4096 KEY_SZ = 32 SALT_SZ = 16 @@ -25,7 +27,7 @@ def verify_enc_key(enc_key, db_page1): stored_hmac = db_page1[PAGE_SZ - 64: PAGE_SZ] hm = hmac_mod.new(mac_key, hmac_data, hashlib.sha512) hm.update(struct.pack("