NoCannoBB

外观

02 · 只读读库层 ​

有了明文库，这一章把「怎么安全地读它」写成一组只读函数。配套代码： code/wechat_win/db_reader_win.py。这层是后面 MCP 的唯一数据来源，所以三条纪律从这里 就定下来：只读打开、参数化查询、标识符白名单校验。

1. 只读打开明文库 ​

解密后先落成明文临时文件，再用 stdlib sqlite3 以只读 URI打开：

python
def decrypt_to_temp(enc_path, hex_key) -> str:
    fd, tmp = tempfile.mkstemp(suffix=".db", prefix="wxplain_")
    os.close(fd)
    sqlcipher4.decrypt_db(enc_path, hex_key, tmp)   # 见第 1 章
    return tmp                                        # 调用方负责删除

@contextmanager
def _connect(plain_path):
    conn = sqlite3.connect(f"file:{plain_path}?mode=ro", uri=True)  # 只读
    try:
        yield conn
    finally:
        conn.close()

mode=ro 保证这层物理上不可能改动数据。临时文件由调用方（CLI 或 MCP）在 finally 里删除——第 5 章会强调这条清理纪律。

2. 防注入：标识符白名单 + 参数化 ​

微信的消息表名是动态的（Msg_<md5(username)>），表名不能用占位符 ?，只能拼进 SQL。 所以对表名/列名一律走白名单正则校验，对值一律用参数化：

python
_SAFE_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")

def _validate_identifier(name: str) -> str:
    if not _SAFE_IDENTIFIER.match(name):
        raise ValueError(f"不安全的标识符: {name!r}")
    return name

读消息时，表名先过 _validate_identifier，条数用 ? 绑定：

python
safe_table = _validate_identifier(table)
cur = conn.execute(
    f"SELECT local_id, server_id, local_type, create_time, "
    f"real_sender_id, message_content, WCDB_CT_message_content "
    f"FROM [{safe_table}] ORDER BY local_id DESC LIMIT ?",
    (limit,),
)

这条纪律是 MCP 只暴露固定工具、不暴露 execute_sql 的技术前提：值永远参数化， 标识符永远白名单，模型就没有任何注入面。

3. 消息表的结构与解码 ​

微信 4.x 每个会话一张 Msg_<md5(username)> 表。几个要点：

  • 发现会话：discover_message_tables() 从 sqlite_master 找 Msg_% 且只含一个下划线 的表；Name2Id 表提供 rowid → username 映射； Msg_<md5(username)> 反查得到「表 → 会话」对应关系。
  • 正文解压：message_content 可能是 zstd 压缩的字节（WCDB_CT_message_content 是标志）：
python
def _decode_content(raw, compress_flag) -> str:
    if compress_flag and isinstance(raw, (bytes, bytearray)):
        return zstandard.ZstdDecompressor().decompress(bytes(raw)).decode("utf-8", "replace")
    if isinstance(raw, (bytes, bytearray)):
        return bytes(raw).decode("utf-8", "replace")
    return str(raw)
  • 群消息发送者：群里正文常是 wxid:\n真正内容，read_recent_messages 会把发送者 从正文里剥出来放到 sender_in_group，并判断 is_sender（是不是「我」发的）。
  • 返回的是不可变 Message dataclass（local_id / create_time / talker / content / msg_type / is_sender / sender_in_group），结果按时间升序。

4. 联系人：表名/列名都做兼容 ​

不同版本联系人表名和列名不一样，所以先探测表名，再探测列名，全部经过白名单校验：

python
contact_table = next((n for n in ["Contact","contact","rcontact","Friend"] if n in tables), None)
username_col = _find_col(col_names, ["username", "userName", "wxid"])
nickname_col = _find_col(col_names, ["nickname", "nickName", "nick_name"])
remark_col   = _find_col(col_names, ["remark", "remarkName", "conRemark"])

返回 {username: Contact(username, nickname, remark)}。

5. 这一层暴露给上层的「四个动作」 ​

到这里，读库层能干的事收敛成四个只读动作，正好对应第 3 章要暴露的 MCP 工具：

动作函数
列会话discover_message_tables + build_table_conversation_map
列联系人read_contacts
读某会话历史read_recent_messages
关键字搜消息在多张表上跑 read_recent_messages + Python 层子串匹配

注意上层能调用的只有这几个固定动作——没有「传一段 SQL 进来执行」的口子。这不是 偷懒，是刻意的安全设计：把能力面缩到最小，模型再聪明也越不过去。

上一章：01 · 取密钥与解密 ｜ 下一章：03 · 封装为 MCP