diff --git a/.github/workflows/windows-wechat-vault.yml b/.github/workflows/windows-wechat-vault.yml new file mode 100644 index 0000000..379ee84 --- /dev/null +++ b/.github/workflows/windows-wechat-vault.yml @@ -0,0 +1,46 @@ +name: Windows WeChat vault + +on: + push: + paths: + - 'yichen-wechat-windows-vault/**' + - '.github/workflows/windows-wechat-vault.yml' + pull_request: + paths: + - 'yichen-wechat-windows-vault/**' + - '.github/workflows/windows-wechat-vault.yml' + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: windows-wechat-vault-${{ github.ref }} + cancel-in-progress: true + +jobs: + test: + runs-on: windows-2025 + timeout-minutes: 15 + steps: + - name: Check out repository + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + + - name: Set up Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: '3.11' + cache: pip + cache-dependency-path: yichen-wechat-windows-vault/requirements-dev.txt + + - name: Install pinned dependencies + run: python -m pip install --requirement yichen-wechat-windows-vault/requirements-dev.txt + + - name: Run synthetic tests + run: python -m pytest yichen-wechat-windows-vault/tests -q + + - name: Run dependency self-test + run: python yichen-wechat-windows-vault/scripts/self_test.py + + - name: Verify environment + run: python -m pip check diff --git a/README.md b/README.md index 4f83240..6a88f16 100644 --- a/README.md +++ b/README.md @@ -88,6 +88,15 @@ WeChat digital-asset assistant for macOS: - Requirements: macOS, WeChat Mac 4.x, Python 3.9+, `pycryptodome`, `zstandard` - See [yichen-wechat-local-vault/README.md](./yichen-wechat-local-vault/README.md) for full documentation +### 4W) `yichen-wechat-windows-vault` +Independent Windows counterpart to the Mac local-vault Skill: +- Uses finite, exact-version key capture and page-HMAC-verified SQLCipher decryption +- Supports sessions, contacts, members, history, search, statistics, exports, Favorites, Moments, and digest-source packs +- Stores verified keys with Windows DPAPI and keeps all decrypted data in a private local vault +- Requires an explicit database path and never changes Codex, Hermes, MCP, startup, or WeChat configuration +- Requirements: Windows 10/11, WeChat 4.x with a supported profile, Python 3.11+, `frida`, `pycryptodome`, `zstandard` +- See [yichen-wechat-windows-vault/README.md](./yichen-wechat-windows-vault/README.md) for full documentation + ### 5–6) Social fetchers integrated into `yichen-content-archive` The former standalone Douyin and Xiaohongshu fetchers now have one source of truth: - `douyin_download.py` reads metadata or downloads a known Douyin video through Playwright interception @@ -261,6 +270,12 @@ yichen-skills/ │ ├─ list_contacts.py │ ├─ search_sns.py │ └─ wechat_digest.py +├─ yichen-wechat-windows-vault/ +│ ├─ SKILL.md +│ ├─ README.md +│ ├─ references/ +│ ├─ scripts/ +│ └─ tests/ ├─ yichen-mac-wechat-dual-open/ │ ├─ SKILL.md │ ├─ scripts/ @@ -368,6 +383,7 @@ yichen-skills/ - Dependencies: - X article drafts: `pip install playwright pycryptodome && python3 -m playwright install chromium` - WeChat local vault: `pip install pycryptodome zstandard` + - Windows WeChat vault: run `yichen-wechat-windows-vault/scripts/setup.ps1` - WeChat dual open: `pip install Pillow` - Content archive (Douyin): `pip install playwright requests && python3 -m playwright install chromium` - Content archive (Xiaohongshu): `pip install requests` @@ -392,6 +408,7 @@ Keep directory names unchanged: - `yichen-summary` - `yichen-x-article-draft-uploader` - `yichen-wechat-local-vault` +- `yichen-wechat-windows-vault` - `yichen-mac-wechat-dual-open` - `yichen-volc-asr` - `yichen-video-content` diff --git a/README.zh.md b/README.zh.md index 1202953..6d0875e 100644 --- a/README.zh.md +++ b/README.zh.md @@ -88,6 +88,15 @@ Mac 微信双开——无需第三方工具，一条命令搞定： - 依赖：macOS、微信 Mac 4.x、Python 3.9+、`pycryptodome`、`zstandard` - 详细文档见 [yichen-wechat-local-vault/README.md](./yichen-wechat-local-vault/README.md) +### 4W) `yichen-wechat-windows-vault` +Mac 本地 Vault Skill 的独立 Windows 对应实现： +- 采用限时、精确版本密钥捕获和逐页 HMAC 验证的 SQLCipher 解密 +- 支持会话、联系人、群成员、历史、搜索、统计、导出、收藏、朋友圈和摘要素材包 +- 使用 Windows DPAPI 保存已验证密钥，明文数据只进入本机私有 Vault +- 必须显式指定数据库路径，不修改 Codex、Hermes、MCP、启动项或微信配置 +- 依赖：Windows 10/11、受支持的微信 4.x、Python 3.11+、`frida`、`pycryptodome`、`zstandard` +- 详细文档见 [yichen-wechat-windows-vault/README.md](./yichen-wechat-windows-vault/README.md) + ### 5–6) 已融合进 `yichen-content-archive` 的社交平台抓取器 原先独立的抖音和小红书抓取器现在只保留一个事实源： - `douyin_download.py` 通过 Playwright 拦截读取元数据或下载已知抖音视频 @@ -260,6 +269,12 @@ yichen-skills/ │ ├─ list_contacts.py │ ├─ search_sns.py │ └─ wechat_digest.py +├─ yichen-wechat-windows-vault/ +│ ├─ SKILL.md +│ ├─ README.md +│ ├─ references/ +│ ├─ scripts/ +│ └─ tests/ ├─ yichen-mac-wechat-dual-open/ │ ├─ SKILL.md │ ├─ scripts/ @@ -367,6 +382,7 @@ yichen-skills/ - 依赖： - X 文章草稿：`pip install playwright pycryptodome && python3 -m playwright install chromium` - 微信本地解析：`pip install pycryptodome zstandard` + - Windows 微信本地 Vault：运行 `yichen-wechat-windows-vault/scripts/setup.ps1` - 微信双开：`pip install Pillow` - 内容归档（抖音）：`pip install playwright requests && python3 -m playwright install chromium` - 内容归档（小红书）：`pip install requests` @@ -391,6 +407,7 @@ yichen-skills/ - `yichen-summary` - `yichen-x-article-draft-uploader` - `yichen-wechat-local-vault` +- `yichen-wechat-windows-vault` - `yichen-mac-wechat-dual-open` - `yichen-volc-asr` - `yichen-video-content` diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index 60cf1f1..6796d33 100644 --- a/THIRD_PARTY_NOTICES.md +++ b/THIRD_PARTY_NOTICES.md @@ -127,6 +127,22 @@ This repository references and adapts ideas/workflows from external projects. - The protocol and implementation in this repository have been adapted for this Skill family and are not represented as an official upstream release. - The upstream copyright and complete MIT license text are preserved in the local license copy above. +## 11) Windows WeChat vault runtime and protocol references + +- Usage in this repo: `yichen-wechat-windows-vault`. +- Runtime dependencies: + - Frida Python bindings 17.15.4, wxWindows Library Licence 3.1. + - PyCryptodome 3.23.0, BSD / public-domain components. + - python-zstandard 0.23.0, BSD-3-Clause. +- Reference-only sources: + - SQLCipher official source and verification utility for public page-layout and HMAC behavior. + - OpenSSL 1.1.1 source for identification of the public `PKCS5_PBKDF2_HMAC` function structure. + - Frida official JavaScript API documentation for finite `Interceptor.attach` usage. +- What was copied: + - No third-party source is vendored in this Skill. + - The Windows process adapter, exact-version profile, DPAPI store, page verifier, schema queries, and synthetic tests are independently implemented. +- Full dependency links and scope notes are recorded in `yichen-wechat-windows-vault/THIRD_PARTY_NOTICES.md` and `PROVENANCE.md`. + ## Notes - This repository maintains its own license (`LICENSE`) for original contributions. It is personal-learning and non-commercial only. diff --git a/docs/superpowers/specs/2026-08-18-windows-wechat-vault-design.md b/docs/superpowers/specs/2026-08-18-windows-wechat-vault-design.md new file mode 100644 index 0000000..f6588e4 --- /dev/null +++ b/docs/superpowers/specs/2026-08-18-windows-wechat-vault-design.md @@ -0,0 +1,53 @@ +# Windows WeChat Vault Skill Design + +## Goal + +Publish the supplied Windows WeChat local-vault skill in `mcncarl/yichen-skills` without altering the maintained macOS vault or exposing personal data. + +## Scope + +- Add a new, self-contained `yichen-wechat-windows-vault/` skill directory from the supplied v0.3.0 archive. +- Preserve the archive's Windows-only implementation, privacy rules, compatibility reference, MCP configuration, and self-test. +- Add concise English and Chinese entries to the repository indexes so users can discover the Windows skill and its OS boundary. +- Add the new directory to the README project trees and the documented stable directory-name lists. + +## Deliberate boundaries + +- Do not merge this implementation into `yichen-wechat-local-vault`; that directory is macOS-only and uses a different database format, key extraction flow, filesystem layout, and dependencies. +- Do not change existing macOS behavior or installation guidance. +- Do not include keys, live databases, decrypted exports, media caches, credentials, user paths, or chat content. +- Do not claim support for an unlisted WeChat build; the bundled compatibility checks remain authoritative. + +## Layout + +```text +yichen-wechat-windows-vault/ +├─ SKILL.md +├─ README.md +├─ agents/openai.yaml +├─ references/ +│ ├─ compatibility.md +│ └─ profile-fixtures.json +└─ scripts/ + ├─ setup.ps1 + ├─ uninstall.ps1 + ├─ capture_keys.py + ├─ decrypt_databases.py + ├─ refresh_vault.py + ├─ diagnose.py + ├─ self_test.py + ├─ vault_cli.py + ├─ vault_common.py + ├─ wechat_media.py + ├─ decode_silk.cjs + ├─ gateway_query.py + ├─ mcp_server.py + └─ requirements.txt +``` + +## Validation + +1. Check that the imported file set matches the archive and that no tracked file matches repository secret/database/media exclusions. +2. Run Python syntax compilation for the bundled Windows scripts on the current host; this checks syntax only and does not attach to WeChat or read local user data. +3. Run the archive's profile-fixture validation in an isolated import context; full `self_test.py` is intentionally not expected to pass on macOS because it validates Windows-only dependencies and paths. +4. Review the final diff and README links before committing and opening the PR. diff --git a/yichen-wechat-windows-vault/PROVENANCE.md b/yichen-wechat-windows-vault/PROVENANCE.md new file mode 100644 index 0000000..e4a471e --- /dev/null +++ b/yichen-wechat-windows-vault/PROVENANCE.md @@ -0,0 +1,34 @@ +# Provenance + +## Relationship to the Mac Skill + +This Skill follows the existing `yichen-wechat-local-vault` user-facing concepts: local private vault, sessions, contacts, members, history, search, statistics, export, Favorites, Moments, and digest-source generation. That same-repository Skill was used as the functional and interface reference. + +The following Windows components were implemented independently for this contribution: + +- exact Windows PE fingerprint and prologue validation; +- finite Frida attachment to one explicitly selected Weixin process; +- derived-key candidate handling and page-HMAC verification; +- Windows DPAPI key storage; +- snapshot-based incremental decryption; +- Windows WeChat 4.x contact, session, message-shard, Favorites, Moments, and resource-index adapters; +- synthetic fixtures and automated tests. + +## Excluded sources + +No source code, implementation details, binary, package, fork, mirror, or generated output from wx-cli, wxcli, jackwener, or related projects was consulted, copied, adapted, or invoked. The implementation does not detect or import those packages at runtime. + +No code from the previously closed Windows v0.1/v0.3 contribution was reused. Existing private vault tooling was used only as a black-box source of schema availability and query-count comparisons; its source was not used to implement this Skill. + +## Independent compatibility derivation + +The supported Windows profile was derived from the user's installed `Weixin.dll` by: + +1. hashing the complete DLL; +2. reading PE exception-function boundaries; +3. locating an embedded OpenSSL source assertion reference for `crypto/evp/p5_crpt2.c`; +4. identifying the adjacent x64 function by the documented `PKCS5_PBKDF2_HMAC` argument and loop structure; +5. recording and checking the exact function prologue; +6. accepting captured results only when the SQLCipher page HMAC validates against an explicit local database. + +Official OpenSSL, SQLCipher, and Frida documentation were used as protocol and API references. No third-party implementation was copied into the repository. diff --git a/yichen-wechat-windows-vault/README.md b/yichen-wechat-windows-vault/README.md new file mode 100644 index 0000000..450d62a --- /dev/null +++ b/yichen-wechat-windows-vault/README.md @@ -0,0 +1,47 @@ +# Yichen WeChat Windows Vault + +An independent Windows implementation of the repository's WeChat local-vault experience. It supports finite, version-pinned key capture; verified SQLCipher decryption; incremental private-vault refresh; and read-only queries for sessions, contacts, group members, history, search, statistics, exports, Favorites, Moments, and message-resource metadata. + +## Requirements + +- Windows 10 or 11 +- Python 3.11+ +- Desktop WeChat 4.x with an exact supported profile +- Data belonging to the current user, with explicit authorization + +## Install + +```powershell +Set-Location yichen-wechat-windows-vault +.\scripts\setup.ps1 +``` + +This creates an isolated environment under `%LOCALAPPDATA%\yichen-wechat-windows-vault`. It does not modify Codex, Hermes, MCP, WeChat, logon tasks, or system configuration. + +## Diagnose, capture, and refresh + +Always provide the database root explicitly: + +```powershell +$Python = Join-Path $env:LOCALAPPDATA "yichen-wechat-windows-vault\venv\Scripts\python.exe" +& $Python .\scripts\diagnose.py --db-root "C:\explicit\path\to\db_storage" +.\scripts\capture_keys_on_demand.ps1 -DbRoot "C:\explicit\path\to\db_storage" -Duration 20 +``` + +Keep WeChat running and open the needed area during the finite capture window. If no new database derivation occurs, the command may correctly report zero captured candidates; retry during a normal WeChat restart or while opening a database-backed feature. Unknown DLL hashes fail closed. + +## Query + +```powershell +$Root = Join-Path $env:LOCALAPPDATA "yichen-wechat-windows-vault\vault\decrypted" +& $Python .\scripts\vault_cli.py --decrypted-root $Root sessions --limit 20 +& $Python .\scripts\vault_cli.py --decrypted-root $Root history "chat name" --limit 50 +& $Python .\scripts\vault_cli.py --decrypted-root $Root search "keyword" --start-time 2026-08-01 +& $Python .\scripts\vault_cli.py --decrypted-root $Root stats "chat name" +``` + +Run `vault_cli.py --help` and the subcommand help for all options. Exports must point to a deliberate private destination. + +## Scope + +The Windows implementation keeps the Mac Skill's command-level concepts and interaction model. Its Windows process integration, version profile, DPAPI storage, database refresh, schema adapters, and tests were implemented independently. See [PROVENANCE.md](PROVENANCE.md) and [THIRD_PARTY_NOTICES.md](THIRD_PARTY_NOTICES.md). diff --git a/yichen-wechat-windows-vault/SKILL.md b/yichen-wechat-windows-vault/SKILL.md new file mode 100644 index 0000000..1f4756b --- /dev/null +++ b/yichen-wechat-windows-vault/SKILL.md @@ -0,0 +1,39 @@ +--- +name: yichen-wechat-windows-vault +description: Capture, decrypt, refresh, and query the current user's authorized local Windows WeChat 4.x vault. Use for Windows chat sessions, contacts, group members, history, search, statistics, exports, Favorites, Moments, or digest-source requests; key capture is finite and version-pinned. +--- + +# Yichen WeChat Windows Vault + +Operate only on the current user's authorized Windows WeChat data. Treat the live database tree as read-only and keep keys, decrypted databases, state, and exports out of repositories. + +## Workflow + +1. Require an explicit `db_storage` path. Do not search Documents, Desktop, OneDrive, or other user folders. +2. Run `scripts/diagnose.py --db-root ` before capture. +3. Run `scripts/setup.ps1` once when the isolated runtime is missing. It does not edit Codex, Hermes, MCP, startup, or WeChat configuration. +4. Use `scripts/capture_keys_on_demand.ps1 -DbRoot -Duration 20` only with authorization to attach to the user's own Weixin process. Keep the window finite. The wrapper selects exactly one process that has `Weixin.dll` loaded. +5. Stop when the DLL fingerprint or function prologue is unknown. Read [references/compatibility.md](references/compatibility.md) before adding a profile. +6. Query only the private decrypted copy with `scripts/vault_cli.py --decrypted-root `. + +Use bounded limits and explicit dates for large searches. Prefer `digest-source` for a reusable group-chat material pack and `export` for a direct transcript. Use `resources` only for metadata; this Skill does not decode images or transcribe voice. + +## Commands + +- `status`, `sessions`, `unread`, `new-messages` +- `contacts`, `members` +- `history`, `search`, `stats` +- `favorites`, `moments`, `resources` +- `export`, `digest-source` + +## Invariants + +- Never print, return, commit, or upload captured keys. +- Accept a captured candidate only after SQLCipher page-HMAC verification. +- Store verified keys with Windows DPAPI for the current user. +- Never modify a live WeChat database or control the WeChat process. +- Never create persistent hooks, watchers, services, or scheduled tasks. +- Do not claim capture success when the result reports zero verified databases. +- Do not commit private vault files, real chat data, account identifiers, or absolute personal paths. + +Read [references/privacy.md](references/privacy.md) when handling real data or exports. diff --git a/yichen-wechat-windows-vault/THIRD_PARTY_NOTICES.md b/yichen-wechat-windows-vault/THIRD_PARTY_NOTICES.md new file mode 100644 index 0000000..60946b3 --- /dev/null +++ b/yichen-wechat-windows-vault/THIRD_PARTY_NOTICES.md @@ -0,0 +1,16 @@ +# Third-Party Notices + +Runtime dependencies are pinned in `requirements.txt` and installed into an isolated local environment. + +| Dependency | Version | Purpose | License / source | +|---|---:|---|---| +| Frida Python bindings | 17.15.4 | Finite local process instrumentation | wxWindows Library Licence 3.1; https://github.com/frida/frida | +| PyCryptodome | 3.23.0 | AES-256-CBC operations | BSD / public-domain components; https://github.com/Legrandin/pycryptodome | +| python-zstandard | 0.23.0 | WeChat compressed message decoding | BSD-3-Clause; https://github.com/indygreg/python-zstandard | +| pytest | 8.4.1 | Development tests only | MIT; https://github.com/pytest-dev/pytest | + +Reference-only specifications and source navigation: + +- SQLCipher official source and verification utility (BSD-3-Clause) were consulted for page layout, HMAC salt masking, key derivation, and little-endian page numbering. No SQLCipher source is vendored. +- OpenSSL 1.1.1 source was consulted to identify the public `PKCS5_PBKDF2_HMAC` function structure in a local binary. No OpenSSL source is vendored. +- The same-repository Mac Skill was consulted only for the product interface and functional vocabulary under this repository's license. diff --git a/yichen-wechat-windows-vault/agents/openai.yaml b/yichen-wechat-windows-vault/agents/openai.yaml new file mode 100644 index 0000000..2d1fe78 --- /dev/null +++ b/yichen-wechat-windows-vault/agents/openai.yaml @@ -0,0 +1,4 @@ +interface: + display_name: "Yichen WeChat Windows Vault" + short_description: "Windows WeChat local vault capture and analysis" + default_prompt: "Use $yichen-wechat-windows-vault to refresh and query my authorized local Windows WeChat vault." diff --git a/yichen-wechat-windows-vault/references/compatibility.md b/yichen-wechat-windows-vault/references/compatibility.md new file mode 100644 index 0000000..193d1fb --- /dev/null +++ b/yichen-wechat-windows-vault/references/compatibility.md @@ -0,0 +1,22 @@ +# Windows compatibility profiles + +Key capture is supported only when every profile check succeeds. + +| WeChat version | Weixin.dll SHA-256 | PBKDF2 RVA | Required prologue | +|---|---|---:|---| +| 4.1.10.53 | `AB35CFBD7AA9514EC0530747CFC59CAE9DEB0DD46D953548D3CF01919C62A577` | `0x68831B0` | `40535556574154415541564157B898000000E82955F4FF482BE0488B052FD197` | + +The runtime table is `scripts/profiles.json`. The profile pins the complete DLL hash and the bytes at the target RVA. A matching version label alone is insufficient. + +## Adding a version + +Do not guess or copy an offset from another build. For each new DLL: + +1. record the complete SHA-256 and file version; +2. independently identify the PBKDF2 function within that exact PE image; +3. record at least 16 bytes of its prologue; +4. validate captured 32-byte outputs against multiple encrypted database first-page HMACs; +5. add a synthetic regression and document the derivation evidence; +6. keep unknown builds fail-closed until review is complete. + +The current profile was derived independently from PE metadata, embedded OpenSSL source references, x64 calling convention, and the OpenSSL PBKDF2 control flow. It was not obtained from an older Windows implementation. diff --git a/yichen-wechat-windows-vault/references/privacy.md b/yichen-wechat-windows-vault/references/privacy.md new file mode 100644 index 0000000..cbbf5b0 --- /dev/null +++ b/yichen-wechat-windows-vault/references/privacy.md @@ -0,0 +1,11 @@ +# Privacy and operational boundaries + +- Work only with the current user's account and explicit local paths. +- Source databases are read-only. Snapshot before decryption. +- Keep `keys.dpapi`, `manifest.json`, decrypted databases, query state, and exports under a private local directory. +- DPAPI protects keys for the current Windows user; it does not make exported plaintext chat data safe to share. +- Never paste keys into command-line arguments, environment variables, logs, issues, commits, or pull requests. +- Never commit real databases, messages, contacts, account identifiers, media, exports, or personal absolute paths. +- Key capture is on-demand and finite. Never install a watcher, service, scheduled task, or persistent hook. +- If a capture reports zero verified databases, report that outcome and retry only when normal WeChat activity can cause the required database derivation. +- This Skill indexes media metadata only. It does not decode image files, transcribe voice, or recover server-only content. diff --git a/yichen-wechat-windows-vault/requirements-dev.txt b/yichen-wechat-windows-vault/requirements-dev.txt new file mode 100644 index 0000000..df60fe9 --- /dev/null +++ b/yichen-wechat-windows-vault/requirements-dev.txt @@ -0,0 +1,2 @@ +-r requirements.txt +pytest==8.4.1 diff --git a/yichen-wechat-windows-vault/requirements.txt b/yichen-wechat-windows-vault/requirements.txt new file mode 100644 index 0000000..f086a49 --- /dev/null +++ b/yichen-wechat-windows-vault/requirements.txt @@ -0,0 +1,3 @@ +pycryptodome==3.23.0 +zstandard==0.23.0 +frida==17.15.4 diff --git a/yichen-wechat-windows-vault/scripts/capture_keys.py b/yichen-wechat-windows-vault/scripts/capture_keys.py new file mode 100644 index 0000000..6feaf0c --- /dev/null +++ b/yichen-wechat-windows-vault/scripts/capture_keys.py @@ -0,0 +1,164 @@ +"""Capture and store only page-verified keys during a finite Frida session.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import time +from pathlib import Path + +from key_store import load_keys, save_keys +from sqlcipher import SQLITE_HEADER, verify_database +from vault_common import VaultError, require_explicit_dir, vault_home + + +AGENT = r""" +'use strict'; +let listener = null; +rpc.exports = { + inspectmodule(moduleName) { + const wanted = moduleName.toLowerCase(); + const module = Process.enumerateModules().find(m => m.name.toLowerCase() === wanted); + if (module === undefined) throw new Error(moduleName + ' is not loaded'); + return {path: module.path, base: module.base.toString(), size: module.size}; + }, + install(moduleName, rva, expectedHex) { + const wanted = moduleName.toLowerCase(); + const module = Process.enumerateModules().find(m => m.name.toLowerCase() === wanted); + if (module === undefined) throw new Error(moduleName + ' is not loaded'); + const target = module.base.add(rva); + const actual = target.readByteArray(expectedHex.length / 2); + const actualHex = Array.from(new Uint8Array(actual), b => b.toString(16).padStart(2, '0')).join(''); + if (actualHex !== expectedHex.toLowerCase()) throw new Error('function prologue mismatch'); + listener = Interceptor.attach(target, { + onEnter(args) { + this.output = args[7]; + this.keyLength = args[6].toInt32(); + }, + onLeave(retval) { + if (!retval.isNull() && this.keyLength === 32 && !this.output.isNull()) { + send({kind: 'derived-key'}, this.output.readByteArray(32)); + } + } + }); + return true; + }, + uninstall() { + if (listener !== null) { listener.detach(); listener = null; } + } +}; +""" + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest().upper() + + +def encrypted_databases(root: Path) -> list[Path]: + result = [] + for path in sorted(root.rglob("*.db")): + try: + with path.open("rb") as handle: + header = handle.read(16) + if header != SQLITE_HEADER: + result.append(path) + except OSError: + continue + return result + + +def load_profiles() -> dict: + path = Path(__file__).with_name("profiles.json") + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise VaultError(f"invalid compatibility profile file: {path}") from exc + + +def capture(db_root: Path, key_file: Path, duration: int, pid: int | None = None) -> dict: + try: + import frida + except ImportError as exc: + raise VaultError("frida is required; install requirements.txt first") from exc + device = frida.get_local_device() + if pid is None: + matches = [item for item in device.enumerate_processes() if item.name.casefold() == "weixin.exe"] + if not matches: + raise VaultError("Weixin.exe is not running") + if len(matches) != 1: + raise VaultError("multiple Weixin.exe processes found; pass the PID that loads Weixin.dll") + pid = matches[0].pid + session = device.attach(pid) + script = session.create_script(AGENT) + candidates: set[bytes] = set() + + def on_message(message, data) -> None: + payload = message.get("payload", {}) + if message.get("type") == "send" and payload.get("kind") == "derived-key": + if data is not None and len(data) == 32: + candidates.add(bytes(data)) + + script.on("message", on_message) + script.load() + try: + info = script.exports_sync.inspectmodule("Weixin.dll") + dll_path = Path(info["path"]) + digest = sha256_file(dll_path) + profile = load_profiles().get(digest) + if profile is None: + raise VaultError(f"unsupported Weixin.dll SHA-256: {digest}") + script.exports_sync.install( + profile["module"], int(profile["rva"]), profile["prologue"] + ) + deadline = time.monotonic() + max(1, min(int(duration), 120)) + while time.monotonic() < deadline: + time.sleep(0.1) + finally: + try: + script.exports_sync.uninstall() + except Exception: + pass + script.unload() + session.detach() + + databases = encrypted_databases(db_root) + keys = load_keys(key_file) + matched_paths: set[str] = set() + for candidate in candidates: + for database in databases: + if verify_database(database, candidate): + relative = database.relative_to(db_root).as_posix() + keys[relative] = candidate + matched_paths.add(relative) + if matched_paths: + save_keys(key_file, keys) + return { + "capture_seconds": max(1, min(int(duration), 120)), + "candidate_count": len(candidates), + "verified_database_count": len(matched_paths), + "stored_database_key_count": len(keys), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--db-root", required=True, help="Explicit WeChat db_storage root") + parser.add_argument("--vault-home") + parser.add_argument("--duration", type=int, default=20) + parser.add_argument("--pid", type=int) + args = parser.parse_args() + try: + root = require_explicit_dir(args.db_root, "--db-root") + result = capture(root, vault_home(args.vault_home) / "keys.dpapi", args.duration, args.pid) + print(json.dumps(result, ensure_ascii=False)) + except VaultError as exc: + parser.error(str(exc)) + + +if __name__ == "__main__": + main() diff --git a/yichen-wechat-windows-vault/scripts/capture_keys_on_demand.ps1 b/yichen-wechat-windows-vault/scripts/capture_keys_on_demand.ps1 new file mode 100644 index 0000000..2a27e57 --- /dev/null +++ b/yichen-wechat-windows-vault/scripts/capture_keys_on_demand.ps1 @@ -0,0 +1,38 @@ +[CmdletBinding()] +param( + [Parameter(Mandatory = $true)] + [string]$DbRoot, + [int]$Duration = 20, + [int]$Pid = 0, + [string]$VaultHome = "" +) + +$ErrorActionPreference = "Stop" +$ScriptRoot = Split-Path -Parent $MyInvocation.MyCommand.Path +$Python = Join-Path $env:LOCALAPPDATA "yichen-wechat-windows-vault\venv\Scripts\python.exe" +if (-not (Test-Path -LiteralPath $Python -PathType Leaf)) { + throw "Runtime not found. Run setup.ps1 first." +} + +$captureArgs = @((Join-Path $ScriptRoot "capture_keys.py"), "--db-root", $DbRoot, "--duration", $Duration) +$refreshArgs = @((Join-Path $ScriptRoot "refresh_vault.py"), "--db-root", $DbRoot) +if ($Pid -eq 0) { + $candidates = @(Get-Process -Name Weixin -ErrorAction Stop | Where-Object { + try { @($_.Modules | Where-Object { $_.ModuleName -ieq "Weixin.dll" }).Count -eq 1 } + catch { $false } + }) + if ($candidates.Count -ne 1) { + throw "Expected exactly one Weixin.exe process with Weixin.dll loaded; pass -Pid explicitly." + } + $Pid = $candidates[0].Id +} +$captureArgs += @("--pid", $Pid) +if ($VaultHome) { + $captureArgs += @("--vault-home", $VaultHome) + $refreshArgs += @("--vault-home", $VaultHome) +} + +& $Python @captureArgs +if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE } +& $Python @refreshArgs +exit $LASTEXITCODE diff --git a/yichen-wechat-windows-vault/scripts/diagnose.py b/yichen-wechat-windows-vault/scripts/diagnose.py new file mode 100644 index 0000000..ea694c2 --- /dev/null +++ b/yichen-wechat-windows-vault/scripts/diagnose.py @@ -0,0 +1,72 @@ +"""Diagnose only an explicitly supplied Windows WeChat data directory.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from pathlib import Path + +from vault_common import VaultError, require_explicit_dir + + +def diagnose(root: Path) -> dict: + databases = [] + for path in sorted(root.rglob("*.db")): + with path.open("rb") as handle: + header = handle.read(16) + kind = "plaintext" if header == b"SQLite format 3\x00" else "encrypted" + databases.append({ + "relative_path": path.relative_to(root).as_posix(), + "bytes": path.stat().st_size, + "kind": kind, + "salt_fingerprint": None if kind == "plaintext" else hashlib.sha256(header).hexdigest()[:12], + }) + return { + "database_count": len(databases), + "encrypted_count": sum(item["kind"] == "encrypted" for item in databases), + "plaintext_count": sum(item["kind"] == "plaintext" for item in databases), + "databases": databases, + } + + +def inspect_dll(path: Path) -> dict: + if not path.is_file(): + raise VaultError(f"DLL not found: {path}") + hasher = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + hasher.update(chunk) + digest = hasher.hexdigest().upper() + profiles_path = Path(__file__).with_name("profiles.json") + profiles = json.loads(profiles_path.read_text(encoding="utf-8")) + profile = profiles.get(digest) + return { + "sha256": digest, + "supported": profile is not None, + "profile_version": None if profile is None else profile["version"], + } + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--db-root", required=True) + parser.add_argument("--dll", help="Optional explicit Weixin.dll path") + parser.add_argument("--json", action="store_true") + args = parser.parse_args() + try: + result = diagnose(require_explicit_dir(args.db_root, "--db-root")) + if args.dll: + result["dll"] = inspect_dll(Path(args.dll).expanduser().resolve()) + except (OSError, VaultError) as exc: + parser.error(str(exc)) + if args.json: + print(json.dumps(result, ensure_ascii=False, indent=2)) + else: + print(f"Databases: {result['database_count']} (encrypted {result['encrypted_count']}, plaintext {result['plaintext_count']})") + for item in result["databases"]: + print(f"- {item['kind']:9} {item['relative_path']}") + + +if __name__ == "__main__": + main() diff --git a/yichen-wechat-windows-vault/scripts/key_store.py b/yichen-wechat-windows-vault/scripts/key_store.py new file mode 100644 index 0000000..1a5c580 --- /dev/null +++ b/yichen-wechat-windows-vault/scripts/key_store.py @@ -0,0 +1,77 @@ +"""Windows DPAPI-backed storage for verified database keys.""" + +from __future__ import annotations + +import ctypes +import json +import os +import tempfile +from ctypes import wintypes +from pathlib import Path + +from vault_common import VaultError, ensure_private_dir + + +class _Blob(ctypes.Structure): + _fields_ = [("cbData", wintypes.DWORD), ("pbData", ctypes.POINTER(ctypes.c_byte))] + + +def _input_blob(data: bytes) -> tuple[_Blob, ctypes.Array]: + buffer = ctypes.create_string_buffer(data) + blob = _Blob(len(data), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_byte))) + return blob, buffer + + +def _crypt(data: bytes, protect: bool) -> bytes: + if os.name != "nt": + raise VaultError("DPAPI key storage is available only on Windows") + source, keepalive = _input_blob(data) + destination = _Blob() + crypt32 = ctypes.WinDLL("crypt32", use_last_error=True) + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + if protect: + ok = crypt32.CryptProtectData( + ctypes.byref(source), "Yichen WeChat vault key", None, None, None, + 0x01, ctypes.byref(destination), + ) + else: + ok = crypt32.CryptUnprotectData( + ctypes.byref(source), None, None, None, None, 0x01, + ctypes.byref(destination), + ) + if not ok: + raise ctypes.WinError(ctypes.get_last_error()) + try: + return ctypes.string_at(destination.pbData, destination.cbData) + finally: + kernel32.LocalFree(destination.pbData) + + +def save_keys(path: Path, keys: dict[str, bytes]) -> None: + ensure_private_dir(path.parent) + payload = json.dumps({name: value.hex() for name, value in keys.items()}).encode("ascii") + protected = _crypt(payload, True) + fd, temp_name = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent) + try: + with os.fdopen(fd, "wb") as handle: + handle.write(protected) + handle.flush() + os.fsync(handle.fileno()) + os.chmod(temp_name, 0o600) + os.replace(temp_name, path) + finally: + if os.path.exists(temp_name): + os.unlink(temp_name) + + +def load_keys(path: Path) -> dict[str, bytes]: + if not path.exists(): + return {} + try: + raw = json.loads(_crypt(path.read_bytes(), False).decode("ascii")) + result = {str(name): bytes.fromhex(str(value)) for name, value in raw.items()} + except (OSError, ValueError, json.JSONDecodeError) as exc: + raise VaultError(f"unable to read DPAPI key store: {path}") from exc + if any(len(value) != 32 for value in result.values()): + raise VaultError("DPAPI key store contains an invalid key") + return result diff --git a/yichen-wechat-windows-vault/scripts/profiles.json b/yichen-wechat-windows-vault/scripts/profiles.json new file mode 100644 index 0000000..dcd30f2 --- /dev/null +++ b/yichen-wechat-windows-vault/scripts/profiles.json @@ -0,0 +1,8 @@ +{ + "AB35CFBD7AA9514EC0530747CFC59CAE9DEB0DD46D953548D3CF01919C62A577": { + "version": "4.1.10.53", + "module": "Weixin.dll", + "rva": 109588912, + "prologue": "40535556574154415541564157B898000000E82955F4FF482BE0488B052FD197" + } +} diff --git a/yichen-wechat-windows-vault/scripts/refresh_vault.py b/yichen-wechat-windows-vault/scripts/refresh_vault.py new file mode 100644 index 0000000..775bd30 --- /dev/null +++ b/yichen-wechat-windows-vault/scripts/refresh_vault.py @@ -0,0 +1,126 @@ +"""Incrementally copy and decrypt an explicit WeChat database root.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shutil +import sqlite3 +import tempfile +from pathlib import Path + +from key_store import load_keys, save_keys +from sqlcipher import SQLITE_HEADER, copy_plain_database, decrypt_database, verify_database +from vault_common import VaultError, atomic_json, ensure_private_dir, load_json, require_explicit_dir, vault_home + + +def fingerprint(path: Path) -> dict: + stat = path.stat() + with path.open("rb") as handle: + first = handle.read(16) + return { + "size": stat.st_size, + "mtime_ns": stat.st_mtime_ns, + "header_sha256": hashlib.sha256(first).hexdigest(), + } + + +def validate_plaintext(path: Path) -> None: + try: + uri = path.resolve().as_uri() + "?mode=ro&immutable=1" + with sqlite3.connect(uri, uri=True) as con: + con.execute("PRAGMA query_only=ON") + con.execute("SELECT name FROM sqlite_master LIMIT 1").fetchone() + except sqlite3.DatabaseError as exc: + raise VaultError(f"decrypted SQLite validation failed: {path.name}") from exc + + +def matching_key(snapshot: Path, relative: str, keys: dict[str, bytes]) -> bytes | None: + preferred = keys.get(relative) + if preferred is not None and verify_database(snapshot, preferred): + return preferred + seen: set[bytes] = set() + for candidate in keys.values(): + if candidate in seen: + continue + seen.add(candidate) + if verify_database(snapshot, candidate): + return candidate + return None + + +def refresh(db_root: Path, home: Path) -> dict: + ensure_private_dir(home) + decrypted_root = ensure_private_dir(home / "vault" / "decrypted") + staging = ensure_private_dir(home / "staging") + key_file = home / "keys.dpapi" + manifest_path = home / "manifest.json" + keys = load_keys(key_file) + manifest = load_json(manifest_path, {"databases": {}}) + previous = manifest.get("databases", {}) + current: dict[str, dict] = {} + result = {"ok": 0, "unchanged": 0, "missing_key": 0, "failed": 0, "databases": []} + + for source in sorted(db_root.rglob("*.db")): + relative = source.relative_to(db_root).as_posix() + state = fingerprint(source) + destination = decrypted_root / Path(relative) + if previous.get(relative) == state and destination.is_file(): + current[relative] = state + result["unchanged"] += 1 + result["databases"].append({"path": relative, "status": "unchanged"}) + continue + token = hashlib.sha256(relative.encode("utf-8")).hexdigest()[:16] + fd, snapshot_name = tempfile.mkstemp(prefix=token + ".", suffix=".db", dir=staging) + os.close(fd) + snapshot = Path(snapshot_name) + try: + shutil.copyfile(source, snapshot) + with snapshot.open("rb") as handle: + plain = handle.read(16) == SQLITE_HEADER + if plain: + copy_plain_database(snapshot, destination) + else: + key = matching_key(snapshot, relative, keys) + if key is None: + result["missing_key"] += 1 + result["databases"].append({"path": relative, "status": "missing_key"}) + continue + decrypt_database(snapshot, destination, key) + keys[relative] = key + validate_plaintext(destination) + current[relative] = state + result["ok"] += 1 + result["databases"].append({"path": relative, "status": "ok"}) + except (OSError, VaultError) as exc: + result["failed"] += 1 + result["databases"].append({"path": relative, "status": "failed", "error": str(exc)}) + finally: + snapshot.unlink(missing_ok=True) + + save_keys(key_file, keys) + atomic_json(manifest_path, {"databases": current}, private=True) + return result + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--db-root", required=True, help="Explicit WeChat db_storage root") + parser.add_argument("--vault-home") + args = parser.parse_args() + try: + result = refresh( + require_explicit_dir(args.db_root, "--db-root"), + vault_home(args.vault_home), + ) + print(json.dumps(result, ensure_ascii=False, indent=2)) + if result["failed"] or result["missing_key"]: + raise SystemExit(2) + except VaultError as exc: + parser.error(str(exc)) + + +if __name__ == "__main__": + main() diff --git a/yichen-wechat-windows-vault/scripts/self_test.py b/yichen-wechat-windows-vault/scripts/self_test.py new file mode 100644 index 0000000..0ff7310 --- /dev/null +++ b/yichen-wechat-windows-vault/scripts/self_test.py @@ -0,0 +1,34 @@ +"""Dependency and static profile checks that do not access WeChat data.""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + + +def main() -> int: + try: + import Crypto + import frida + import zstandard + except ImportError as exc: + print(f"missing dependency: {exc.name}", file=sys.stderr) + return 1 + profiles = json.loads(Path(__file__).with_name("profiles.json").read_text(encoding="utf-8")) + for digest, profile in profiles.items(): + assert len(digest) == 64 and int(digest, 16) >= 0 + assert profile["module"].casefold() == "weixin.dll" + assert profile["rva"] > 0 + assert len(bytes.fromhex(profile["prologue"])) >= 16 + print(json.dumps({ + "ok": True, + "python": sys.version.split()[0], + "frida": frida.__version__, + "profiles": len(profiles), + })) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/yichen-wechat-windows-vault/scripts/setup.ps1 b/yichen-wechat-windows-vault/scripts/setup.ps1 new file mode 100644 index 0000000..c0568cf --- /dev/null +++ b/yichen-wechat-windows-vault/scripts/setup.ps1 @@ -0,0 +1,25 @@ +[CmdletBinding()] +param( + [string]$Python = "python" +) + +$ErrorActionPreference = "Stop" +$ScriptRoot = Split-Path -Parent $MyInvocation.MyCommand.Path +$SkillRoot = Split-Path -Parent $ScriptRoot +$RuntimeRoot = Join-Path $env:LOCALAPPDATA "yichen-wechat-windows-vault" +$Venv = Join-Path $RuntimeRoot "venv" +$VenvPython = Join-Path $Venv "Scripts\python.exe" + +New-Item -ItemType Directory -Path $RuntimeRoot -Force | Out-Null +if (-not (Test-Path -LiteralPath $VenvPython -PathType Leaf)) { + & $Python -m venv $Venv + if ($LASTEXITCODE -ne 0) { throw "Unable to create virtual environment." } +} + +& $VenvPython -m pip install --disable-pip-version-check -r (Join-Path $SkillRoot "requirements.txt") +if ($LASTEXITCODE -ne 0) { throw "Dependency installation failed." } +& $VenvPython (Join-Path $ScriptRoot "self_test.py") +if ($LASTEXITCODE -ne 0) { throw "Self-test failed." } + +Write-Output "Runtime ready: $RuntimeRoot" +Write-Output "No Codex, Hermes, MCP, startup-task, or WeChat configuration was changed." diff --git a/yichen-wechat-windows-vault/scripts/sqlcipher.py b/yichen-wechat-windows-vault/scripts/sqlcipher.py new file mode 100644 index 0000000..3664357 --- /dev/null +++ b/yichen-wechat-windows-vault/scripts/sqlcipher.py @@ -0,0 +1,122 @@ +"""SQLCipher 4 page verification and decryption for authorized local copies.""" + +from __future__ import annotations + +import hashlib +import hmac +import os +import shutil +import struct +import tempfile +from dataclasses import dataclass +from pathlib import Path + +from Crypto.Cipher import AES + +from vault_common import VaultError + + +SQLITE_HEADER = b"SQLite format 3\x00" + + +@dataclass(frozen=True) +class CipherProfile: + page_size: int = 4096 + reserve_size: int = 80 + hmac_size: int = 64 + hmac_salt_mask: int = 0x3A + hmac_kdf_iter: int = 2 + + +PROFILE = CipherProfile() + + +def _hmac_key(raw_key: bytes, salt: bytes, profile: CipherProfile = PROFILE) -> bytes: + hmac_salt = bytes(byte ^ profile.hmac_salt_mask for byte in salt) + return hashlib.pbkdf2_hmac( + "sha512", raw_key, hmac_salt, profile.hmac_kdf_iter, dklen=32 + ) + + +def verify_page( + page: bytes, + page_number: int, + raw_key: bytes, + salt: bytes, + profile: CipherProfile = PROFILE, +) -> bool: + if len(raw_key) != 32 or len(salt) != 16 or len(page) != profile.page_size: + return False + offset = 16 if page_number == 1 else 0 + payload_end = profile.page_size - profile.reserve_size + 16 + stored = page[-profile.hmac_size :] + calculated = hmac.new( + _hmac_key(raw_key, salt, profile), + page[offset:payload_end] + struct.pack(" bool: + with path.open("rb") as handle: + page = handle.read(profile.page_size) + if len(page) != profile.page_size or page.startswith(SQLITE_HEADER): + return False + return verify_page(page, 1, raw_key, page[:16], profile) + + +def decrypt_database( + source: Path, + destination: Path, + raw_key: bytes, + profile: CipherProfile = PROFILE, +) -> int: + if not source.is_file(): + raise VaultError(f"encrypted database not found: {source}") + if len(raw_key) != 32: + raise VaultError("database key must contain exactly 32 bytes") + destination.parent.mkdir(parents=True, exist_ok=True) + fd, temp_name = tempfile.mkstemp(prefix=destination.name + ".", dir=destination.parent) + os.close(fd) + pages = 0 + try: + with source.open("rb") as src, open(temp_name, "wb") as dst: + salt = src.read(16) + src.seek(0) + if len(salt) != 16: + raise VaultError(f"database is too short: {source}") + while True: + page = src.read(profile.page_size) + if not page: + break + pages += 1 + if len(page) != profile.page_size: + raise VaultError(f"truncated page {pages}: {source}") + if not verify_page(page, pages, raw_key, salt, profile): + raise VaultError(f"page HMAC verification failed at page {pages}: {source.name}") + offset = 16 if pages == 1 else 0 + usable_end = profile.page_size - profile.reserve_size + iv = page[usable_end : usable_end + 16] + clear = AES.new(raw_key, AES.MODE_CBC, iv).decrypt(page[offset:usable_end]) + if pages == 1: + dst.write(SQLITE_HEADER) + dst.write(clear) + dst.write(page[usable_end:]) + dst.flush() + os.fsync(dst.fileno()) + if pages == 0: + raise VaultError(f"empty database: {source}") + os.replace(temp_name, destination) + return pages + finally: + if os.path.exists(temp_name): + os.unlink(temp_name) + + +def copy_plain_database(source: Path, destination: Path) -> None: + with source.open("rb") as handle: + if handle.read(16) != SQLITE_HEADER: + raise VaultError(f"not a plaintext SQLite database: {source}") + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, destination) diff --git a/yichen-wechat-windows-vault/scripts/vault_cli.py b/yichen-wechat-windows-vault/scripts/vault_cli.py new file mode 100644 index 0000000..8b5fd14 --- /dev/null +++ b/yichen-wechat-windows-vault/scripts/vault_cli.py @@ -0,0 +1,480 @@ +"""Read-only queries for decrypted Windows WeChat 4.x vault copies.""" + +from __future__ import annotations + +import argparse +import datetime as dt +import hashlib +import json +import re +import xml.etree.ElementTree as ET +from collections import Counter +from pathlib import Path + +from vault_common import ( + VaultError, atomic_json, find_db, iter_message_dbs, load_json, + quote_identifier, readonly_connect, require_explicit_dir, table_columns, + table_names, vault_home, +) + + +TYPE_LABELS = { + 1: "text", 3: "image", 34: "voice", 37: "contact-card", 42: "contact-card", + 43: "video", 47: "sticker", 48: "location", 49: "app", 50: "call", + 10000: "system", +} + + +def parse_time(value: str | None, end: bool = False) -> int | None: + if not value: + return None + text = value.strip() + if text.isdigit(): + return int(text) + formats = ["%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"] + for fmt in formats: + try: + parsed = dt.datetime.strptime(text, fmt) + if fmt == "%Y-%m-%d" and end: + parsed += dt.timedelta(days=1, microseconds=-1) + return int(parsed.timestamp()) + except ValueError: + pass + raise VaultError(f"invalid time: {value}") + + +def decode_blob(value) -> str: + if value is None: + return "" + if isinstance(value, str): + return value + if isinstance(value, memoryview): + value = value.tobytes() + if not isinstance(value, bytes): + return str(value) + for raw in (value,): + try: + return raw.decode("utf-8") + except UnicodeDecodeError: + pass + try: + import zstandard + return zstandard.ZstdDecompressor().decompress(value).decode("utf-8", "replace") + except Exception: + return value.decode("utf-8", "replace") + + +def emit(value, fmt: str) -> None: + if fmt == "json": + print(json.dumps(value, ensure_ascii=False, indent=2)) + return + if isinstance(value, list): + for item in value: + print(" | ".join(f"{key}={val}" for key, val in item.items() if val not in (None, ""))) + elif isinstance(value, dict): + for key, val in value.items(): + print(f"{key}: {val}") + else: + print(value) + + +def contact_db(root: Path) -> Path: + path = find_db(root, ["contact.db"]) + if path is None: + raise VaultError("contact.db not found") + return path + + +def load_contacts(root: Path) -> list[dict]: + with readonly_connect(contact_db(root)) as con: + if "contact" not in table_names(con): + raise VaultError("contact table not found") + rows = con.execute( + "SELECT id,username,alias,remark,nick_name,local_type,delete_flag," + "is_in_chat_room,chat_room_type FROM contact" + ).fetchall() + return [dict(row) for row in rows] + + +def display_name(contact: dict) -> str: + return contact.get("remark") or contact.get("nick_name") or contact.get("alias") or contact.get("username") or "" + + +def resolve_contact(root: Path, query: str) -> dict: + needle = query.casefold() + contacts = load_contacts(root) + exact = [item for item in contacts if needle in { + str(item.get(key) or "").casefold() for key in ("username", "alias", "remark", "nick_name") + }] + if len(exact) == 1: + return exact[0] + partial = [item for item in contacts if any( + needle in str(item.get(key) or "").casefold() + for key in ("username", "alias", "remark", "nick_name") + )] + if len(partial) == 1: + return partial[0] + if not partial: + raise VaultError(f"contact not found: {query}") + names = ", ".join(display_name(item) for item in partial[:8]) + raise VaultError(f"contact is ambiguous: {query} ({names})") + + +def command_status(root: Path) -> dict: + expected = ["contact.db", "session.db", "favorite.db", "sns.db", "message_resource.db"] + found = sorted(path.relative_to(root).as_posix() for path in root.rglob("*.db")) + return { + "decrypted_root": str(root), + "database_count": len(found), + "message_shards": len(iter_message_dbs(root)), + "components": {name: any(path.endswith(name) for path in found) for name in expected}, + "databases": found, + } + + +def command_contacts(root: Path, query: str | None, limit: int) -> list[dict]: + rows = load_contacts(root) + if query: + needle = query.casefold() + rows = [row for row in rows if any( + needle in str(row.get(key) or "").casefold() + for key in ("username", "alias", "remark", "nick_name") + )] + rows.sort(key=lambda row: display_name(row).casefold()) + return [{**row, "display_name": display_name(row)} for row in rows[:limit]] + + +def command_members(root: Path, group: str) -> dict: + contact = resolve_contact(root, group) + with readonly_connect(contact_db(root)) as con: + room = con.execute("SELECT id,username,owner FROM chat_room WHERE username=?", (contact["username"],)).fetchone() + if room is None: + raise VaultError(f"not a group chat: {group}") + members = con.execute( + "SELECT c.username,c.alias,c.remark,c.nick_name FROM chatroom_member m " + "JOIN contact c ON c.id=m.member_id WHERE m.room_id=?", (room["id"],) + ).fetchall() + items = [dict(row) for row in members] + return {"group": contact["username"], "name": display_name(contact), "owner": room["owner"], "members": items} + + +def session_db(root: Path) -> Path: + path = find_db(root, ["session.db"]) + if path is None: + raise VaultError("session.db not found") + return path + + +def command_sessions(root: Path, unread: bool, limit: int) -> list[dict]: + contacts = {row["username"]: row for row in load_contacts(root)} + where = "WHERE unread_count>0" if unread else "" + with readonly_connect(session_db(root)) as con: + rows = con.execute( + "SELECT username,unread_count,summary,last_timestamp,sort_timestamp,last_msg_type," + "last_msg_sub_type,last_sender_display_name FROM SessionTable " + where + + " ORDER BY sort_timestamp DESC LIMIT ?", (limit,) + ).fetchall() + result = [] + for row in rows: + item = dict(row) + item["name"] = display_name(contacts.get(item["username"], {"username": item["username"]})) + item["summary"] = decode_blob(item["summary"]) + result.append(item) + return result + + +def _message_table(username: str) -> str: + return "Msg_" + hashlib.md5(username.encode("utf-8")).hexdigest() + + +def _candidate_tables(root: Path, contact: dict | None) -> list[tuple[Path, str, str]]: + selected = _message_table(contact["username"]) if contact else None + result = [] + contacts = load_contacts(root) if contact is None else [contact] + by_table = {_message_table(item["username"]).casefold(): item["username"] for item in contacts} + for path in iter_message_dbs(root): + with readonly_connect(path) as con: + for table in table_names(con, "Msg_"): + if selected is None or table.casefold() == selected.casefold(): + result.append((path, table, by_table.get(table.casefold(), ""))) + return result + + +def _message_type(local_type: int) -> tuple[int, int]: + value = int(local_type or 0) + if value > 0xFFFF: + return value & 0xFFFF, value >> 16 + return value, 0 + + +def _rows_from_table( + path: Path, table: str, username: str, start: int | None, end: int | None, + keyword: str | None, type_name: str | None, +) -> list[dict]: + conditions = [] + params: list[object] = [] + if start is not None: + conditions.append("create_time>=?") + params.append(start) + if end is not None: + conditions.append("create_time<=?") + params.append(end) + sql = ( + "SELECT local_id,server_id,local_type,sort_seq,real_sender_id,create_time,status," + "message_content,compress_content,source,packed_info_data FROM " + quote_identifier(table) + ) + if conditions: + sql += " WHERE " + " AND ".join(conditions) + with readonly_connect(path) as con: + rows = con.execute(sql, params).fetchall() + result = [] + for row in rows: + item = dict(row) + content = decode_blob(item.get("message_content")) or decode_blob(item.get("compress_content")) + base_type, subtype = _message_type(item.get("local_type") or 0) + label = TYPE_LABELS.get(base_type, f"type-{base_type}") + if keyword and keyword.casefold() not in content.casefold(): + continue + if type_name and label != type_name: + continue + item.update({ + "chat": username, "content": content, "type": label, "subtype": subtype, + "database": path.name, "table": table, + }) + for binary in ("message_content", "compress_content", "source", "packed_info_data"): + item.pop(binary, None) + result.append(item) + return result + + +def collect_messages( + root: Path, chat: str | None, start: int | None, end: int | None, + keyword: str | None, type_name: str | None, limit: int, offset: int, +) -> list[dict]: + contact = resolve_contact(root, chat) if chat else None + rows = [] + for path, table, username in _candidate_tables(root, contact): + rows.extend(_rows_from_table(path, table, username or (contact or {}).get("username", ""), start, end, keyword, type_name)) + rows.sort(key=lambda row: (row.get("sort_seq") or row.get("create_time") or 0, row.get("local_id") or 0), reverse=True) + return rows[offset : offset + limit] + + +def command_stats(root: Path, chat: str, start: int | None, end: int | None) -> dict: + rows = collect_messages(root, chat, start, end, None, None, 1_000_000, 0) + by_type = Counter(row["type"] for row in rows) + by_sender = Counter(str(row.get("real_sender_id") or "unknown") for row in rows) + return { + "chat": resolve_contact(root, chat)["username"], + "message_count": len(rows), + "by_type": dict(by_type.most_common()), + "by_sender_id": dict(by_sender.most_common()), + "first_timestamp": min((row["create_time"] for row in rows), default=None), + "last_timestamp": max((row["create_time"] for row in rows), default=None), + } + + +def command_favorites(root: Path, query: str | None, limit: int) -> list[dict]: + path = find_db(root, ["favorite.db"]) + if path is None: + raise VaultError("favorite.db not found") + with readonly_connect(path) as con: + rows = con.execute( + "SELECT local_id,server_id,type,update_time,content,fromusr,realchatname " + "FROM fav_db_item ORDER BY update_time DESC" + ).fetchall() + result = [] + for row in rows: + item = dict(row) + item["content"] = decode_blob(item["content"]) + if query and query.casefold() not in item["content"].casefold(): + continue + result.append(item) + if len(result) >= limit: + break + return result + + +def _xml_value(root: ET.Element, *paths: str) -> str: + for path in paths: + value = root.findtext(path) + if value: + return value + return "" + + +def command_moments(root: Path, username: list[str] | None, keyword: str | None, limit: int) -> list[dict]: + path = find_db(root, ["sns.db"]) + if path is None: + raise VaultError("sns.db not found") + conditions = [] + params: list[object] = [] + if username: + conditions.append("user_name IN (" + ",".join("?" for _ in username) + ")") + params.extend(username) + # pack_info_buf is an opaque protobuf and may be declared as TEXT in some + # builds despite containing arbitrary bytes. It is intentionally excluded. + sql = "SELECT tid,user_name,content FROM SnsTimeLine" + if conditions: + sql += " WHERE " + " AND ".join(conditions) + with readonly_connect(path) as con: + rows = con.execute(sql, params).fetchall() + result = [] + for row in rows: + raw = decode_blob(row["content"]) + if keyword and keyword.casefold() not in raw.casefold(): + continue + item = {"tid": row["tid"], "username": row["user_name"], "raw": raw} + try: + xml = ET.fromstring(raw) + location = xml.find(".//location") + item.update({ + "text": _xml_value(xml, ".//contentDesc", ".//content"), + "create_time": _xml_value(xml, ".//createTime"), + "location": "" if location is None else (location.get("poiName") or location.text or ""), + }) + except ET.ParseError: + item["text"] = raw + result.append(item) + if len(result) >= limit: + break + return result + + +def command_new_messages(root: Path, state_file: Path, limit: int) -> dict: + state = load_json(state_file, {"last_timestamp": 0}) + since = int(state.get("last_timestamp", 0)) + rows = collect_messages(root, None, since + 1, None, None, None, limit, 0) + newest = max((int(row["create_time"]) for row in rows), default=since) + atomic_json(state_file, {"last_timestamp": newest}, private=True) + return {"since": since, "newest": newest, "messages": rows} + + +def command_resources(root: Path, chat: str, local_id: int | None, limit: int) -> dict: + path = find_db(root, ["message_resource.db"]) + if path is None: + raise VaultError("message_resource.db not found") + with readonly_connect(path) as con: + chat_row = con.execute( + "SELECT rowid,user_name FROM ChatName2Id WHERE user_name=?", (chat,) + ).fetchone() + if chat_row is None: + try: + contact = resolve_contact(root, chat) + except VaultError: + return {"chat": chat, "resources": []} + chat_row = con.execute( + "SELECT rowid,user_name FROM ChatName2Id WHERE user_name=?", (contact["username"],) + ).fetchone() + if chat_row is None: + return {"chat": chat, "resources": []} + conditions = ["i.chat_id=?"] + params: list[object] = [chat_row[0]] + if local_id is not None: + conditions.append("i.message_local_id=?") + params.append(local_id) + params.append(limit) + rows = con.execute( + "SELECT i.message_id,i.message_local_type,i.message_create_time," + "i.message_local_id,i.message_svr_id,d.resource_id,d.type AS resource_type," + "d.size,d.status,d.data_index FROM MessageResourceInfo i " + "LEFT JOIN MessageResourceDetail d ON d.message_id=i.message_id WHERE " + + " AND ".join(conditions) + + " ORDER BY i.message_create_time DESC LIMIT ?", + params, + ).fetchall() + return {"chat": chat_row["user_name"], "resources": [dict(row) for row in rows]} + + +def command_digest(root: Path, args: argparse.Namespace) -> dict: + rows = collect_messages( + root, args.group, parse_time(args.start), parse_time(args.end, True), + None, None, args.limit, 0, + ) + contact = resolve_contact(root, args.group) + destination = Path(args.output).expanduser().resolve() + destination.parent.mkdir(parents=True, exist_ok=True) + counts = Counter(row["type"] for row in rows) + lines = [ + f"# {display_name(contact)} digest source", + "", + f"- Messages: {len(rows)}", + f"- Types: {json.dumps(dict(counts), ensure_ascii=False)}", + "", + "## Timeline", + "", + ] + for row in reversed(rows): + stamp = dt.datetime.fromtimestamp(row["create_time"]).isoformat(sep=" ", timespec="seconds") + lines.append(f"- {stamp} [{row['type']}] {row['content']}") + destination.write_text("\n".join(lines) + "\n", encoding="utf-8") + return {"output": str(destination), "message_count": len(rows), "by_type": dict(counts)} + + +def export_messages(root: Path, args: argparse.Namespace) -> dict: + rows = collect_messages( + root, args.chat, parse_time(args.start_time), parse_time(args.end_time, True), + None, args.type, args.limit, 0, + ) + destination = Path(args.output).expanduser().resolve() + destination.parent.mkdir(parents=True, exist_ok=True) + title = display_name(resolve_contact(root, args.chat)) + lines = [f"# {title}", ""] if args.export_format == "markdown" else [] + for row in reversed(rows): + stamp = dt.datetime.fromtimestamp(row["create_time"]).isoformat(sep=" ", timespec="seconds") + lines.append(f"- {stamp} [{row['type']}] {row['content']}" if args.export_format == "markdown" else f"{stamp}\t{row['type']}\t{row['content']}") + destination.write_text("\n".join(lines) + "\n", encoding="utf-8") + return {"output": str(destination), "message_count": len(rows)} + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--decrypted-root", required=True, help="Explicit decrypted vault root") + parser.add_argument("--format", choices=["json", "text"], default="json") + sub = parser.add_subparsers(dest="command", required=True) + sub.add_parser("status") + for name, unread in (("sessions", False), ("unread", True)): + p = sub.add_parser(name); p.add_argument("--limit", type=int, default=20); p.set_defaults(unread=unread) + p = sub.add_parser("new-messages"); p.add_argument("--limit", type=int, default=100); p.add_argument("--vault-home") + p = sub.add_parser("contacts"); p.add_argument("--query"); p.add_argument("--limit", type=int, default=50) + p = sub.add_parser("members"); p.add_argument("group") + for name in ("history", "search"): + p = sub.add_parser(name) + if name == "history": p.add_argument("chat") + else: p.add_argument("keyword"); p.add_argument("--chat") + p.add_argument("--start-time"); p.add_argument("--end-time"); p.add_argument("--type", choices=sorted(set(TYPE_LABELS.values()))) + p.add_argument("--limit", type=int, default=50); p.add_argument("--offset", type=int, default=0) + p = sub.add_parser("stats"); p.add_argument("chat"); p.add_argument("--start-time"); p.add_argument("--end-time") + p = sub.add_parser("favorites"); p.add_argument("--query"); p.add_argument("--limit", type=int, default=20) + p = sub.add_parser("moments"); p.add_argument("--username", action="append"); p.add_argument("--keyword"); p.add_argument("--limit", type=int, default=50) + p = sub.add_parser("resources"); p.add_argument("chat"); p.add_argument("--local-id", type=int); p.add_argument("--limit", type=int, default=50) + p = sub.add_parser("export"); p.add_argument("chat"); p.add_argument("--output", required=True); p.add_argument("--export-format", choices=["markdown", "txt"], default="markdown"); p.add_argument("--start-time"); p.add_argument("--end-time"); p.add_argument("--type", choices=sorted(set(TYPE_LABELS.values()))); p.add_argument("--limit", type=int, default=500) + p = sub.add_parser("digest-source"); p.add_argument("group"); p.add_argument("--output", required=True); p.add_argument("--start"); p.add_argument("--end"); p.add_argument("--limit", type=int, default=5000) + return parser + + +def main(argv: list[str] | None = None) -> None: + parser = build_parser() + args = parser.parse_args(argv) + try: + root = require_explicit_dir(args.decrypted_root, "--decrypted-root") + if args.command == "status": result = command_status(root) + elif args.command in ("sessions", "unread"): result = command_sessions(root, args.unread, args.limit) + elif args.command == "new-messages": result = command_new_messages(root, vault_home(args.vault_home) / "query-state.json", args.limit) + elif args.command == "contacts": result = command_contacts(root, args.query, args.limit) + elif args.command == "members": result = command_members(root, args.group) + elif args.command in ("history", "search"): + result = collect_messages(root, getattr(args, "chat", None), parse_time(args.start_time), parse_time(args.end_time, True), getattr(args, "keyword", None), args.type, args.limit, args.offset) + elif args.command == "stats": result = command_stats(root, args.chat, parse_time(args.start_time), parse_time(args.end_time, True)) + elif args.command == "favorites": result = command_favorites(root, args.query, args.limit) + elif args.command == "moments": result = command_moments(root, args.username, args.keyword, args.limit) + elif args.command == "resources": result = command_resources(root, args.chat, args.local_id, args.limit) + elif args.command == "export": result = export_messages(root, args) + elif args.command == "digest-source": result = command_digest(root, args) + else: raise VaultError(f"unsupported command: {args.command}") + emit(result, args.format) + except (OSError, VaultError) as exc: + parser.error(str(exc)) + + +if __name__ == "__main__": + main() diff --git a/yichen-wechat-windows-vault/scripts/vault_common.py b/yichen-wechat-windows-vault/scripts/vault_common.py new file mode 100644 index 0000000..29e295d --- /dev/null +++ b/yichen-wechat-windows-vault/scripts/vault_common.py @@ -0,0 +1,115 @@ +"""Shared safety and filesystem helpers for the Windows vault.""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import tempfile +from pathlib import Path +from typing import Any, Iterable + + +APP_NAME = "yichen-wechat-windows-vault" +DEFAULT_HOME = Path(os.environ.get("LOCALAPPDATA", tempfile.gettempdir())) / APP_NAME + + +class VaultError(RuntimeError): + """A user-facing vault error.""" + + +def vault_home(value: str | Path | None = None) -> Path: + path = Path(value).expanduser() if value else DEFAULT_HOME + return path.resolve() + + +def ensure_private_dir(path: Path) -> Path: + path.mkdir(parents=True, exist_ok=True) + if os.name == "nt": + # Keep this best-effort and dependency-free. Windows ACL inheritance remains + # authoritative; files containing keys are never written outside this root. + os.chmod(path, 0o700) + return path + + +def require_explicit_dir(value: str | Path | None, label: str) -> Path: + if value is None or not str(value).strip(): + raise VaultError(f"{label} is required; automatic folder scanning is disabled") + path = Path(value).expanduser().resolve() + if not path.is_dir(): + raise VaultError(f"{label} is not a directory: {path}") + return path + + +def require_under(child: Path, parent: Path, label: str) -> Path: + child = child.resolve() + parent = parent.resolve() + if child != parent and parent not in child.parents: + raise VaultError(f"{label} must stay under {parent}") + return child + + +def atomic_json(path: Path, value: Any, *, private: bool = False) -> None: + ensure_private_dir(path.parent) + fd, temp_name = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(value, handle, ensure_ascii=False, indent=2) + handle.write("\n") + if private: + os.chmod(temp_name, 0o600) + os.replace(temp_name, path) + finally: + if os.path.exists(temp_name): + os.unlink(temp_name) + + +def load_json(path: Path, default: Any) -> Any: + if not path.exists(): + return default + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise VaultError(f"invalid JSON file: {path}") from exc + + +def readonly_connect(path: Path) -> sqlite3.Connection: + if not path.is_file(): + raise VaultError(f"database not found: {path}") + uri = path.resolve().as_uri() + "?mode=ro&immutable=1" + con = sqlite3.connect(uri, uri=True) + con.row_factory = sqlite3.Row + con.execute("PRAGMA query_only=ON") + return con + + +def quote_identifier(value: str) -> str: + return '"' + value.replace('"', '""') + '"' + + +def table_names(con: sqlite3.Connection, prefix: str | None = None) -> list[str]: + sql = "SELECT name FROM sqlite_master WHERE type='table'" + params: tuple[str, ...] = () + if prefix is not None: + sql += " AND name LIKE ? ESCAPE '\\'" + params = (prefix.replace("%", "\\%").replace("_", "\\_") + "%",) + return [row[0] for row in con.execute(sql, params)] + + +def table_columns(con: sqlite3.Connection, table: str) -> set[str]: + return {row[1] for row in con.execute(f"PRAGMA table_info({quote_identifier(table)})")} + + +def find_db(root: Path, names: Iterable[str]) -> Path | None: + wanted = {name.casefold() for name in names} + for path in sorted(root.rglob("*.db")): + if path.name.casefold() in wanted: + return path + return None + + +def iter_message_dbs(root: Path) -> list[Path]: + return sorted( + path for path in root.rglob("message_*.db") + if path.stem.removeprefix("message_").isdigit() + ) diff --git a/yichen-wechat-windows-vault/tests/conftest.py b/yichen-wechat-windows-vault/tests/conftest.py new file mode 100644 index 0000000..fa4a9f8 --- /dev/null +++ b/yichen-wechat-windows-vault/tests/conftest.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import hashlib +import sqlite3 +import sys +from pathlib import Path + +import pytest + + +SCRIPTS = Path(__file__).resolve().parents[1] / "scripts" +sys.path.insert(0, str(SCRIPTS)) + + +def create_db(path: Path, statements: list[str], rows: list[tuple[str, tuple]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with sqlite3.connect(path) as con: + for statement in statements: + con.execute(statement) + for sql, params in rows: + con.execute(sql, params) + + +@pytest.fixture() +def synthetic_vault(tmp_path: Path) -> Path: + root = tmp_path / "vault" + create_db( + root / "contact" / "contact.db", + [ + "CREATE TABLE contact(id INTEGER PRIMARY KEY,username TEXT,local_type INTEGER,alias TEXT,encrypt_username TEXT,flag INTEGER,delete_flag INTEGER,verify_flag INTEGER,remark TEXT,remark_quan_pin TEXT,remark_pin_yin_initial TEXT,nick_name TEXT,pin_yin_initial TEXT,quan_pin TEXT,big_head_url TEXT,small_head_url TEXT,head_img_md5 TEXT,chat_room_notify INTEGER,is_in_chat_room INTEGER,description TEXT,extra_buffer BLOB,chat_room_type INTEGER)", + "CREATE TABLE chat_room(id INTEGER PRIMARY KEY,username TEXT,owner TEXT,ext_buffer BLOB)", + "CREATE TABLE chatroom_member(room_id INTEGER,member_id INTEGER)", + ], + [ + ("INSERT INTO contact(id,username,alias,remark,nick_name,local_type,delete_flag,is_in_chat_room,chat_room_type) VALUES(?,?,?,?,?,?,?,?,?)", (1,"alice","alice_alias","Alice Remark","Alice",1,0,0,0)), + ("INSERT INTO contact(id,username,alias,remark,nick_name,local_type,delete_flag,is_in_chat_room,chat_room_type) VALUES(?,?,?,?,?,?,?,?,?)", (2,"group@chatroom","","Synthetic Group","Group",2,0,1,1)), + ("INSERT INTO chat_room(id,username,owner) VALUES(?,?,?)", (7,"group@chatroom","alice")), + ("INSERT INTO chatroom_member(room_id,member_id) VALUES(?,?)", (7,1)), + ], + ) + create_db( + root / "session" / "session.db", + ["CREATE TABLE SessionTable(username TEXT,type INTEGER,unread_count INTEGER,unread_first_msg_srv_id INTEGER,unread_first_pat_msg_local_id INTEGER,unread_first_pat_msg_sort_seq INTEGER,is_hidden INTEGER,summary BLOB,draft TEXT,status INTEGER,last_timestamp INTEGER,sort_timestamp INTEGER,last_clear_unread_timestamp INTEGER,last_msg_locald_id INTEGER,last_msg_type INTEGER,last_msg_sub_type INTEGER,last_msg_sender INTEGER,last_sender_display_name TEXT,last_msg_ext_type INTEGER)"], + [("INSERT INTO SessionTable(username,unread_count,summary,last_timestamp,sort_timestamp,last_msg_type,last_msg_sub_type,last_sender_display_name) VALUES(?,?,?,?,?,?,?,?)", ("alice",2,"hello",1700000000,1700000000,1,0,"Alice"))], + ) + table = "Msg_" + hashlib.md5(b"alice").hexdigest() + create_db( + root / "message" / "message_0.db", + [f"CREATE TABLE {table}(local_id INTEGER,server_id INTEGER,local_type INTEGER,sort_seq INTEGER,real_sender_id INTEGER,create_time INTEGER,status INTEGER,upload_status INTEGER,download_status INTEGER,server_seq INTEGER,origin_source BLOB,source BLOB,message_content BLOB,compress_content BLOB,packed_info_data BLOB,WCDB_CT_message_content INTEGER,WCDB_CT_source INTEGER)"], + [ + (f"INSERT INTO {table}(local_id,server_id,local_type,sort_seq,real_sender_id,create_time,status,message_content) VALUES(?,?,?,?,?,?,?,?)", (1,101,1,1,1,1700000000,0,"hello synthetic")), + (f"INSERT INTO {table}(local_id,server_id,local_type,sort_seq,real_sender_id,create_time,status,message_content) VALUES(?,?,?,?,?,?,?,?)", (2,102,3,2,1,1700000010,0,"image synthetic")), + ], + ) + create_db( + root / "favorite" / "favorite.db", + ["CREATE TABLE fav_db_item(local_id INTEGER,server_id INTEGER,type INTEGER,update_seq INTEGER,flag INTEGER,update_time INTEGER,version INTEGER,content BLOB,source_id TEXT,sync_status INTEGER,upload_status INTEGER,upload_error_code INTEGER,trans_res_status INTEGER,trans_res_error_code INTEGER,fromusr TEXT,fromusr_id INTEGER,realchatname TEXT,realchatname_id INTEGER,ext_buf BLOB)"], + [("INSERT INTO fav_db_item(local_id,server_id,type,update_time,content,fromusr,realchatname) VALUES(?,?,?,?,?,?,?)", (1,2,1,1700000000,"favorite synthetic","alice","alice"))], + ) + create_db( + root / "message" / "message_resource.db", + [ + "CREATE TABLE ChatName2Id(user_name TEXT,update_time INTEGER)", + "CREATE TABLE MessageResourceInfo(message_id INTEGER,chat_id INTEGER,sender_id INTEGER,message_local_type INTEGER,message_create_time INTEGER,message_local_id INTEGER,message_svr_id INTEGER,message_origin_source BLOB,packed_info BLOB)", + "CREATE TABLE MessageResourceDetail(resource_id INTEGER,message_id INTEGER,type INTEGER,size INTEGER,create_time INTEGER,access_time INTEGER,status INTEGER,data_index TEXT,packed_info BLOB)", + ], + [ + ("INSERT INTO ChatName2Id(rowid,user_name) VALUES(?,?)", (3,"alice")), + ("INSERT INTO MessageResourceInfo(message_id,chat_id,message_local_type,message_create_time,message_local_id,message_svr_id) VALUES(?,?,?,?,?,?)", (9,3,3,1700000010,2,102)), + ("INSERT INTO MessageResourceDetail(resource_id,message_id,type,size,status,data_index) VALUES(?,?,?,?,?,?)", (11,9,1,128,0,"synthetic-index")), + ], + ) + create_db( + root / "sns" / "sns.db", + ["CREATE TABLE SnsTimeLine(tid TEXT,user_name TEXT,content BLOB,pack_info_buf BLOB)"], + [("INSERT INTO SnsTimeLine(tid,user_name,content) VALUES(?,?,?)", ("t1","alice","moment synthetic1700000000"))], + ) + return root diff --git a/yichen-wechat-windows-vault/tests/test_capture_and_refresh.py b/yichen-wechat-windows-vault/tests/test_capture_and_refresh.py new file mode 100644 index 0000000..7140eec --- /dev/null +++ b/yichen-wechat-windows-vault/tests/test_capture_and_refresh.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +import hashlib +import os +from pathlib import Path + +import pytest + +import capture_keys +import diagnose +import refresh_vault +from key_store import load_keys, save_keys +from test_sqlcipher import make_encrypted + + +@pytest.mark.skipif(os.name != "nt", reason="DPAPI is Windows-only") +def test_dpapi_key_store_roundtrip(tmp_path: Path) -> None: + path = tmp_path / "keys.dpapi" + expected = {"message/message_0.db": bytes(range(32))} + save_keys(path, expected) + assert load_keys(path) == expected + assert bytes(range(32)).hex().encode("ascii") not in path.read_bytes() + + +class FakeExports: + def __init__(self, script) -> None: + self.script = script + + def inspectmodule(self, name: str) -> dict: + return {"path": str(self.script.dll), "base": "0x1000", "size": 64} + + def install(self, module: str, rva: int, prologue: str) -> bool: + self.script.callback({"type": "send", "payload": {"kind": "derived-key"}}, self.script.key) + return True + + def uninstall(self) -> None: + pass + + +class FakeScript: + def __init__(self, dll: Path, key: bytes) -> None: + self.dll = dll + self.key = key + self.callback = None + self.exports_sync = FakeExports(self) + + def on(self, event: str, callback) -> None: + self.callback = callback + + def load(self) -> None: + pass + + def unload(self) -> None: + pass + + +class FakeSession: + def __init__(self, script: FakeScript) -> None: + self.script = script + + def create_script(self, source: str) -> FakeScript: + return self.script + + def detach(self) -> None: + pass + + +class FakeDevice: + def __init__(self, script: FakeScript) -> None: + self.script = script + + def attach(self, pid: int) -> FakeSession: + return FakeSession(self.script) + + +class FakeFrida: + def __init__(self, script: FakeScript) -> None: + self.device = FakeDevice(script) + + def get_local_device(self) -> FakeDevice: + return self.device + + +def test_capture_stores_only_verified_candidate(tmp_path: Path, monkeypatch) -> None: + key = bytes(range(32)) + db_root = tmp_path / "db" + db_root.mkdir() + make_encrypted(db_root / "message_0.db", key, 1) + dll = tmp_path / "Weixin.dll" + dll.write_bytes(b"synthetic-dll") + script = FakeScript(dll, key) + digest = hashlib.sha256(dll.read_bytes()).hexdigest().upper() + monkeypatch.setitem(__import__("sys").modules, "frida", FakeFrida(script)) + monkeypatch.setattr(capture_keys, "load_profiles", lambda: {digest: {"module":"Weixin.dll", "rva":1, "prologue":"00"}}) + stored = {} + monkeypatch.setattr(capture_keys, "load_keys", lambda path: {}) + monkeypatch.setattr(capture_keys, "save_keys", lambda path, keys: stored.update(keys)) + result = capture_keys.capture(db_root, tmp_path / "keys.dpapi", 1, 123) + assert result["candidate_count"] == 1 + assert result["verified_database_count"] == 1 + assert stored == {"message_0.db": key} + assert key.hex() not in str(result) + + +def test_refresh_uses_verified_key_and_incremental_state(tmp_path: Path, monkeypatch) -> None: + key = bytes(range(32)) + root = tmp_path / "source" + root.mkdir() + make_encrypted(root / "message_0.db", key, 1) + home = tmp_path / "home" + monkeypatch.setattr(refresh_vault, "load_keys", lambda path: {"message_0.db": key}) + monkeypatch.setattr(refresh_vault, "save_keys", lambda path, keys: None) + monkeypatch.setattr(refresh_vault, "validate_plaintext", lambda path: None) + first = refresh_vault.refresh(root, home) + second = refresh_vault.refresh(root, home) + assert first["ok"] == 1 and first["failed"] == 0 + assert second["unchanged"] == 1 + assert (home / "vault" / "decrypted" / "message_0.db").is_file() + + +def test_unknown_dll_is_reported_without_guessing(tmp_path: Path) -> None: + dll = tmp_path / "Weixin.dll" + dll.write_bytes(b"unknown-build") + result = diagnose.inspect_dll(dll) + assert result["supported"] is False + assert len(result["sha256"]) == 64 diff --git a/yichen-wechat-windows-vault/tests/test_provenance.py b/yichen-wechat-windows-vault/tests/test_provenance.py new file mode 100644 index 0000000..c9e2547 --- /dev/null +++ b/yichen-wechat-windows-vault/tests/test_provenance.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +import re +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def test_runtime_has_no_excluded_dependency_names() -> None: + files = list((ROOT / "scripts").glob("*")) + [ROOT / "requirements.txt", ROOT / "requirements-dev.txt"] + text = "\n".join( + path.read_text(encoding="utf-8", errors="ignore") + for path in files if path.is_file() + ).casefold() + assert "wx-cli" not in text + assert "wxcli" not in text + assert "jackwener" not in text + + +def test_repository_files_have_no_personal_absolute_paths() -> None: + candidates = [path for path in ROOT.rglob("*") if path.is_file() and "__pycache__" not in path.parts] + text = "\n".join(path.read_text(encoding="utf-8", errors="ignore") for path in candidates) + assert not re.search(r"C:\\Users\\\d{5,}", text, re.IGNORECASE) + assert not re.search(r"qq\d{6,}", text, re.IGNORECASE) diff --git a/yichen-wechat-windows-vault/tests/test_sqlcipher.py b/yichen-wechat-windows-vault/tests/test_sqlcipher.py new file mode 100644 index 0000000..1f3d75d --- /dev/null +++ b/yichen-wechat-windows-vault/tests/test_sqlcipher.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import hashlib +import hmac +import os +import struct +from pathlib import Path + +import pytest +from Crypto.Cipher import AES + +from sqlcipher import PROFILE, SQLITE_HEADER, _hmac_key, decrypt_database, verify_database +from vault_common import VaultError + + +def make_encrypted(path: Path, key: bytes, pages: int = 2) -> list[bytes]: + salt = bytes(range(16)) + expected = [] + encrypted = [] + for number in range(1, pages + 1): + prefix = SQLITE_HEADER if number == 1 else b"" + clear = bytes(((index + number) % 251 for index in range(PROFILE.page_size - PROFILE.reserve_size - len(prefix)))) + iv = bytes((number + index) % 256 for index in range(16)) + cipher = AES.new(key, AES.MODE_CBC, iv).encrypt(clear) + page = (salt if number == 1 else b"") + cipher + iv + tag = hmac.new(_hmac_key(key, salt), page[(16 if number == 1 else 0):] + struct.pack(" None: + key = bytes(range(32)) + source = tmp_path / "encrypted.db" + expected = make_encrypted(source, key) + assert verify_database(source, key) + output = tmp_path / "clear.db" + assert decrypt_database(source, output, key) == 2 + assert output.read_bytes() == b"".join(expected) + + +def test_hmac_tamper_fails_closed(tmp_path: Path) -> None: + key = bytes(range(32)) + source = tmp_path / "encrypted.db" + make_encrypted(source, key, 1) + data = bytearray(source.read_bytes()) + data[100] ^= 1 + source.write_bytes(data) + assert not verify_database(source, key) + with pytest.raises(VaultError, match="HMAC verification failed"): + decrypt_database(source, tmp_path / "clear.db", key) diff --git a/yichen-wechat-windows-vault/tests/test_vault_cli.py b/yichen-wechat-windows-vault/tests/test_vault_cli.py new file mode 100644 index 0000000..aa3b68c --- /dev/null +++ b/yichen-wechat-windows-vault/tests/test_vault_cli.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +import vault_cli +from vault_common import VaultError, require_explicit_dir + + +def test_explicit_root_required() -> None: + with pytest.raises(VaultError, match="automatic folder scanning is disabled"): + require_explicit_dir(None, "--decrypted-root") + + +def test_status_contacts_sessions_and_members(synthetic_vault: Path) -> None: + assert vault_cli.command_status(synthetic_vault)["message_shards"] == 1 + assert vault_cli.command_contacts(synthetic_vault, "remark", 10)[0]["username"] == "alice" + assert vault_cli.command_sessions(synthetic_vault, True, 10)[0]["unread_count"] == 2 + assert vault_cli.command_members(synthetic_vault, "Synthetic Group")["members"][0]["username"] == "alice" + + +def test_history_search_stats(synthetic_vault: Path) -> None: + history = vault_cli.collect_messages(synthetic_vault, "Alice", None, None, None, None, 10, 0) + assert [row["type"] for row in history] == ["image", "text"] + search = vault_cli.collect_messages(synthetic_vault, None, None, None, "hello", None, 10, 0) + assert len(search) == 1 and search[0]["content"] == "hello synthetic" + stats = vault_cli.command_stats(synthetic_vault, "Alice", None, None) + assert stats["message_count"] == 2 + assert stats["by_type"] == {"image": 1, "text": 1} + + +def test_favorites_moments_and_export(synthetic_vault: Path, tmp_path: Path) -> None: + assert vault_cli.command_favorites(synthetic_vault, "synthetic", 10)[0]["type"] == 1 + assert vault_cli.command_moments(synthetic_vault, ["alice"], "synthetic", 10)[0]["text"] == "moment synthetic" + output = tmp_path / "export.md" + args = type("Args", (), {"chat":"Alice", "start_time":None, "end_time":None, "type":None, "limit":10, "output":str(output), "export_format":"markdown"})() + result = vault_cli.export_messages(synthetic_vault, args) + assert result["message_count"] == 2 + assert "hello synthetic" in output.read_text(encoding="utf-8") + + +def test_resource_index_and_digest(synthetic_vault: Path, tmp_path: Path) -> None: + resources = vault_cli.command_resources(synthetic_vault, "Alice", 2, 10) + assert resources["resources"][0]["size"] == 128 + output = tmp_path / "digest.md" + args = type("Args", (), {"group":"Alice", "start":None, "end":None, "limit":10, "output":str(output)})() + result = vault_cli.command_digest(synthetic_vault, args) + assert result["message_count"] == 2 + assert "hello synthetic" in output.read_text(encoding="utf-8")