ROAD-354: expose contact labels, phone, and stable message IDs

.gitignore

@@ -33,6 +33,11 @@ Thumbs.db
 *.db
 *.db-wal
 *.db-shm
+wechat_ent.plist
+
+# Chat history exports (personal data — never commit)
+exports/*
+!exports/.gitkeep
 
 # Sensitive data — NEVER commit
 *.json

AGENTS.md

@@ -0,0 +1,71 @@
+# AGENTS.md
+
+Read-only WeChat data query CLI (fork of `huohuoer/wechat-cli`) — decrypts local
+WeChat 4.x databases and exposes messages, contacts, sessions, etc. as JSON for
+LLM/agent consumption.
+
+## Hard rules
+
+- **Read-only by design.** No send/write capability — do not add one (security
+  posture per LANDIT ROAD-336 review). UI automation of WeChat is out of scope.
+- **Sensitive data everywhere.** `~/.wechat-cli/all_keys.json` holds SQLCipher
+  keys; decrypted DBs land in `$TMPDIR/wechat_cli_cache` and
+  `~/.wechat-cli/decrypted`. Never commit keys, `*.db*`, or `*.json` output dumps
+  (already gitignored). Don't echo key material or bulk personal data into logs.
+- **Exports go in `exports/`.** Dump query/export output there — the dir is
+  gitignored (except `.gitkeep`) and is the designated place for personal data
+  on disk, e.g. `wechat-cli history "X" --limit 500 > exports/x.json`.
+- **Chinese comments/docstrings** are the codebase convention — match them.
+
+## Setup
+
+```bash
+python3 -m venv .venv && .venv/bin/pip install -e .
+.venv/bin/wechat-cli init        # one-time: extract keys (WeChat must be running + logged in)
+.venv/bin/wechat-cli sessions    # smoke test
+```
+
+`init` runs a bundled C binary (`wechat_cli/bin/find_all_keys_macos.*`) that
+reads WeChat process memory via `task_for_pid`. If blocked, it re-signs
+WeChat preserving entitlements (adds `get-task-allow`), then you must restart
+WeChat and re-run `init`. Works without sudo once re-signed.
+`wechat_ent.plist` produced during re-sign is a temp artifact — delete it.
+
+## Layout
+
+- `wechat_cli/commands/` — one click command per file, registered in `main.py`
+- `wechat_cli/core/` — `context` (AppContext singleton), `config` (`~/.wechat-cli`),
+  `db_cache` (mtime-keyed decrypt cache), `crypto` (SQLCipher AES-256-CBC
+  page/WAL decrypt), `contacts`, `messages`, `key_utils`
+- `wechat_cli/keys/` — platform key scanners; `output/formatter.py` — `output(data, fmt)`
+
+## Data model gotchas
+
+- `contact.db`: `contact` table + `contact_label` (label id→name only; membership
+  is **not** a table — it's `contact.extra_buffer` protobuf field 30).
+- `extra_buffer` protobuf: field 30 = comma-separated label_ids; field 14→2→1 =
+  mobile number. Shared decoder lives in `core/contacts.py` — extend there, don't
+  fork a second parser.
+- Message tables: `Msg_<md5(username)>` across `message/message_*.db`;
+  `Name2Id` maps `real_sender_id`→username. Message identity = `local_id`/`server_id`.
+- Live DBs are read while WeChat writes — `db_cache` validates decrypts with
+  `PRAGMA integrity_check` and retries; don't bypass it by copying DB files.
+- `init` reuses `config.json`'s `db_dir`; machines may have several
+  `xwechat_files/*/db_storage` accounts — never auto-switch.
+
+## Verify
+
+Hermetic suite (synthetic fixture DBs — never touches real WeChat data):
+
+```bash
+.venv/bin/pip install -e ".[dev]" && .venv/bin/python -m pytest tests/
+```
+
+Covers the `extra_buffer` decoder, contact loading/detail, history message IDs,
+db_cache torn-read poisoning, and init `db_dir` preservation. For end-to-end
+checks against the live DB:
+
+```bash
+.venv/bin/wechat-cli contacts --detail "<wxid>"   # labels/phone
+.venv/bin/wechat-cli history "<chat>" --limit 3    # local_id/server_id in JSON
+```

CLAUDE.md

@@ -0,0 +1 @@
+@AGENTS.md

pyproject.toml

@@ -13,6 +13,11 @@ dependencies = [
     "zstandard>=0.22,<1",
 ]
 
+[project.optional-dependencies]
+dev = [
+    "pytest>=8,<9",
+]
+
 [project.scripts]
 wechat-cli = "wechat_cli.main:cli"
 

tests/conftest.py

@@ -0,0 +1,100 @@
+"""测试辅助 — 合成 protobuf 编码与 SQLite fixture 库（不触碰真实微信数据）"""
+
+import hashlib
+import sqlite3
+
+import pytest
+
+
+# ---- protobuf 编码辅助 ----
+
+def encode_varint(value):
+    out = bytearray()
+    while True:
+        b = value & 0x7F
+        value >>= 7
+        if value:
+            out.append(b | 0x80)
+        else:
+            out.append(b)
+            return bytes(out)
+
+
+def field_bytes(fno, data):
+    return encode_varint((fno << 3) | 2) + encode_varint(len(data)) + data
+
+
+def field_varint(fno, value):
+    return encode_varint(fno << 3) + encode_varint(value)
+
+
+def make_extra_buffer(labels_raw=None, phone=None):
+    """构造 contact.extra_buffer：field 30 = 标签 ID 字符串，field 14→2→1 = 手机号。"""
+    buf = b""
+    if phone is not None:
+        inner = field_bytes(1, phone.encode())
+        buf += field_bytes(14, field_varint(1, 1) + field_bytes(2, inner))
+    if labels_raw is not None:
+        buf += field_bytes(30, labels_raw.encode())
+    return buf
+
+
+# ---- SQLite fixture ----
+
+CONTACT_SCHEMA = """
+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 contact_label(label_id_ INTEGER PRIMARY KEY, label_name_ TEXT, sort_order_ INTEGER);
+"""
+
+
+def msg_table_name(username):
+    return "Msg_" + hashlib.md5(username.encode()).hexdigest()
+
+
+MSG_SCHEMA = """
+CREATE TABLE {table}(
+  local_id INTEGER PRIMARY KEY, 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 INTEGER, source TEXT, message_content TEXT,
+  compress_content TEXT, packed_info_data BLOB,
+  WCDB_CT_message_content INTEGER, WCDB_CT_source INTEGER);
+CREATE TABLE Name2Id(user_name TEXT);
+"""
+
+
+def insert_contact(conn, username, nick_name="", remark="", extra_buffer=None, **kw):
+    cols = {"username": username, "nick_name": nick_name, "remark": remark,
+            "extra_buffer": extra_buffer}
+    cols.update(kw)
+    keys = ", ".join(cols)
+    conn.execute(
+        f"INSERT INTO contact({keys}) VALUES ({', '.join('?' * len(cols))})",
+        list(cols.values()),
+    )
+
+
+@pytest.fixture
+def contact_db_path(tmp_path):
+    """带真实 schema 的 contact.db，预置 2 个标签 + 2 个联系人。"""
+    path = tmp_path / "contact.db"
+    conn = sqlite3.connect(path)
+    conn.executescript(CONTACT_SCHEMA)
+    conn.executemany(
+        "INSERT INTO contact_label VALUES (?, ?, ?)",
+        [(1, "客户", 0), (5, "Sydney", 1)],
+    )
+    insert_contact(
+        conn, "wxid_alice", nick_name="Alice", remark="爱丽丝",
+        extra_buffer=make_extra_buffer(labels_raw="1,5", phone="0412345678"),
+    )
+    insert_contact(conn, "wxid_bob", nick_name="Bob")
+    conn.commit()
+    conn.close()
+    return path

tests/test_contacts.py

@@ -0,0 +1,41 @@
+"""contacts 加载/详情测试 — 使用合成 contact.db，不触碰真实微信数据"""
+
+from wechat_cli.core.contacts import _load_contacts_from, get_contact_detail
+
+
+class _NullCache:
+    def get(self, rel_key):
+        return None
+
+
+def test_load_contacts_labels_and_phone(contact_db_path):
+    names, full = _load_contacts_from(str(contact_db_path))
+    assert names["wxid_alice"] == "爱丽丝"
+    alice = next(c for c in full if c["username"] == "wxid_alice")
+    assert alice["labels"] == ["客户", "Sydney"]
+    assert alice["phone"] == "0412345678"
+    bob = next(c for c in full if c["username"] == "wxid_bob")
+    assert bob["labels"] == []
+    assert bob["phone"] == ""
+
+
+def test_get_contact_detail(contact_db_path, tmp_path):
+    # get_contact_detail 优先读 decrypted_dir/contact/contact.db
+    decrypted_dir = tmp_path / "decrypted"
+    (decrypted_dir / "contact").mkdir(parents=True)
+    import shutil
+    shutil.copy(contact_db_path, decrypted_dir / "contact" / "contact.db")
+
+    info = get_contact_detail("wxid_alice", _NullCache(), str(decrypted_dir))
+    assert info["labels"] == ["客户", "Sydney"]
+    assert info["label_ids"] == [1, 5]
+    assert info["phone"] == "0412345678"
+    assert info["is_group"] is False
+
+
+def test_get_contact_detail_missing(contact_db_path, tmp_path):
+    decrypted_dir = tmp_path / "decrypted"
+    (decrypted_dir / "contact").mkdir(parents=True)
+    import shutil
+    shutil.copy(contact_db_path, decrypted_dir / "contact" / "contact.db")
+    assert get_contact_detail("wxid_nobody", _NullCache(), str(decrypted_dir)) is None

tests/test_db_cache.py

@@ -0,0 +1,106 @@
+"""DBCache 回归测试 — 防止撕裂解密结果被缓存（曾导致全部查询失败的 bug）"""
+
+import json
+import os
+import sqlite3
+
+import pytest
+
+import wechat_cli.core.db_cache as db_cache_mod
+from wechat_cli.core.db_cache import DBCache, _has_sqlite_header, _is_valid_sqlite
+
+
+REL_KEY = "contact/contact.db"
+ENC_KEY_HEX = "ab" * 32
+
+
+def _make_valid_db(path):
+    if os.path.exists(path):
+        os.remove(path)
+    conn = sqlite3.connect(path)
+    conn.executescript(
+        "DROP TABLE IF EXISTS t;"
+        "CREATE TABLE t(id INTEGER PRIMARY KEY, v TEXT);"
+        "INSERT INTO t(v) VALUES ('x');"
+    )
+    conn.commit()
+    conn.close()
+
+
+@pytest.fixture
+def env(tmp_path, monkeypatch):
+    """独立 db_dir + cache 目录，full_decrypt/decrypt_wal 可注入。"""
+    cache_dir = tmp_path / "cache"
+    monkeypatch.setattr(DBCache, "CACHE_DIR", str(cache_dir))
+    monkeypatch.setattr(DBCache, "MTIME_FILE", str(cache_dir / "_mtimes.json"))
+    monkeypatch.setattr(db_cache_mod, "_DECRYPT_RETRY_DELAY", 0)
+
+    db_dir = tmp_path / "db_storage"
+    (db_dir / "contact").mkdir(parents=True)
+    (db_dir / "contact" / "contact.db").write_bytes(b"encrypted")
+
+    all_keys = {REL_KEY: {"enc_key": ENC_KEY_HEX}}
+    cache = DBCache(all_keys, str(db_dir))
+    calls = {"decrypt": 0}
+
+    def set_decrypt_result(valid):
+        def fake_full_decrypt(db_path, out_path, enc_key):
+            calls["decrypt"] += 1
+            if valid:
+                _make_valid_db(out_path)
+            else:
+                with open(out_path, "wb") as f:
+                    f.write(b"\x00" * 8192)
+        monkeypatch.setattr(db_cache_mod, "full_decrypt", fake_full_decrypt)
+
+    monkeypatch.setattr(db_cache_mod, "decrypt_wal", lambda *a: 0)
+    return cache, set_decrypt_result, calls, db_dir
+
+
+def test_get_decrypts_and_caches(env):
+    cache, set_decrypt, calls, _ = env
+    set_decrypt(True)
+    p = cache.get(REL_KEY)
+    assert p and _is_valid_sqlite(p)
+    # 第二次命中缓存，不再解密
+    assert cache.get(REL_KEY) == p
+    assert calls["decrypt"] == 1
+
+
+def test_torn_decrypt_not_cached(env):
+    """撕裂读取（微信写入中）不得进入缓存 — 曾导致缓存中毒的核心回归。"""
+    cache, set_decrypt, calls, _ = env
+    set_decrypt(False)
+    assert cache.get(REL_KEY) is None
+    assert calls["decrypt"] == db_cache_mod._DECRYPT_ATTEMPTS
+    # 失败结果不写入持久缓存
+    assert not os.path.exists(DBCache.MTIME_FILE) or REL_KEY not in json.load(
+        open(DBCache.MTIME_FILE)
+    )
+
+
+def test_poisoned_cache_file_not_served(env):
+    """已存在的损坏缓存文件不得被直接返回。"""
+    cache, set_decrypt, calls, db_dir = env
+    set_decrypt(True)
+    good = cache.get(REL_KEY)
+    assert _is_valid_sqlite(good)
+
+    # 模拟中毒：缓存文件被撕裂内容覆盖（mtime 记录不变）
+    with open(good, "wb") as f:
+        f.write(b"\x00" * 8192)
+
+    p2 = cache.get(REL_KEY)
+    assert p2 is not None
+    assert _is_valid_sqlite(p2)
+
+
+def test_is_valid_sqlite(tmp_path):
+    good = tmp_path / "good.db"
+    _make_valid_db(str(good))
+    assert _is_valid_sqlite(str(good))
+    bad = tmp_path / "bad.db"
+    bad.write_bytes(b"\x00" * 8192)
+    assert not _is_valid_sqlite(str(bad))
+    assert not _has_sqlite_header(str(bad))
+    assert not _is_valid_sqlite(str(tmp_path / "nonexistent.db"))