ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

微信实时消息捕获系统:内存钩取+SQLite归档技术解析

微信实时消息捕获系统:内存钩取+SQLite归档技术解析 简介这是一套面向开发者与安全研究人员的微信聊天记录实时监控与分析工具源码解决微信私聊及群聊内容无法直接导出、难以程序化获取的技术痛点适用于合规场景下的数据归档、舆情监测或教学演示。资源共12个文件包含5个核心Python脚本如HttpServer.py、ChatHistory.py实现服务端逻辑与历史数据管理、3张界面/示意图PNG、1份README.md说明文档、1个requirements.txt依赖清单、1个LICENSE授权文件及.gitignore配置整体包体仅172KB轻量易部署。已有926人学习下载代码结构清晰模块职责分明HttpServer.py承载RESTful API服务DataSouceUtils.py封装数据接入逻辑LoggerSetup.py统一日志管理配合config目录可快速适配不同环境。读者可直接运行调试掌握微信消息抓取原理、本地服务搭建与API集成方法并基于现有框架拓展AI话题分析或云端上报功能。1. 实时微信聊天记录查询系统WeChatMsgHistory-real到底在查什么不是备份、不是导出、更不是“恢复已删消息”很多人第一次看到「实时微信聊天记录查询系统WeChatMsgHistory-real」这个标题本能反应是这不就是个微信数据恢复工具或者——是不是能绕过微信官方限制把对方删掉的聊天记录捞回来都不是。这个项目的真实定位是一个基于本地微信客户端运行时状态的、低侵入式的消息捕获与结构化归档系统。它不破解微信协议不模拟登录不调用未公开API也不依赖root或越狱它的核心能力是在用户本人设备上Windows/macOS当微信PC版或Mac版正在运行时实时监听并解析微信进程内存中尚未落盘、但已接收/发送的原始消息结构体然后按时间、会话、类型文本/图片/链接/文件/语音、发送方/接收方等维度存入本地SQLite数据库并提供HTTP接口或CLI命令供查询。这意味着它只对「当前正在运行的微信实例」生效它无法获取历史消息除非你提前开启监听它不触碰微信AppData加密目录里的Msg或Misc数据库那些是微信自己加密存储的它真正解决的是「我刚和客户聊完一笔订单想立刻查3分钟前他说的付款账号是多少」这类高频、低延迟、强上下文的现场回溯需求。适合一线销售、客服主管、合规审计员、以及需要做微信侧行为留痕的中小团队技术负责人——不是给普通用户“找回删掉的暧昧消息”用的而是给需要「可验证、可追溯、不可篡改」聊天证据链的人用的。2. 为什么选内存钩取SQLiteHTTP API而不是数据库直读、ADB抓包或微信开放平台2.1 微信本地数据的三道硬墙加密、混淆、动态偏移微信PC/Mac版从v3.0起彻底弃用明文SQLite存储聊天记录。其Msg目录下.db文件全部AES-256-CBC加密密钥由硬件ID、登录态Token、设备指纹三重派生且每次登录重置更关键的是消息内容在内存中解密后仅以临时结构体形式存在毫秒级随后被清空或覆盖。直接读取磁盘文件空盘用Wireshark抓本地环回流量全是TLS加密的weixin://伪协议调用微信开放平台仅限企业微信且无个人聊天记录权限。所以WeChatMsgHistory-real的选型逻辑非常务实不碰磁盘加密库→ 避开密钥破解这个黑匣子不依赖网络层→ 绕过微信服务端反爬策略如设备指纹校验、请求频率熔断只盯内存中“活着”的消息体→ 利用微信客户端自身解密后的短暂窗口用DLL注入Windows或mach-injectmacOS挂载钩子拦截CMessage类的构造/序列化函数调用栈。2.2 为什么是SQLite而不是MySQL或Elasticsearch项目源码里db/目录下只有history.db一个文件没有建表SQL脚本因为初始化时就内置了建表语句# db.py 中 init_db() 函数片段 conn.execute( CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, msg_id TEXT UNIQUE NOT NULL, -- 微信内部消息唯一ID非时间戳 sender TEXT NOT NULL, -- 发送者wxid或昵称自动映射 receiver TEXT NOT NULL, -- 接收者wxid或群号 content TEXT, -- 解析后纯文本图片/语音存路径 msg_type INTEGER, -- 1:文本, 3:图片, 34:语音, 49:链接/文件 timestamp INTEGER, -- 毫秒级时间戳来自微信内存结构体 is_self INTEGER DEFAULT 0, -- 1自己发的0别人发的 raw_data BLOB -- 原始内存dump可选调试用 ) ) conn.execute(CREATE INDEX IF NOT EXISTS idx_time ON messages(timestamp)) conn.execute(CREATE INDEX IF NOT EXISTS idx_sender ON messages(sender))选SQLite有三个血泪经验支撑零运维部署整个系统打包成单个可执行文件PyInstaller打包后约28MB双击即启无需安装数据库服务ACID够用每条消息写入都是INSERT OR IGNORE冲突时丢弃重复msg_id避免因微信重发机制导致脏数据冷热分离友好messages表默认只存最近30天数据老数据自动VACUUM INTO archive_202405.db——这个归档动作在源码archiver.py里用shutil.move()sqlite3.connect().backup()实现比MySQL分区表配置简单10倍。2.3 HTTP API设计为何只暴露GET /search而不是RESTful全动词看api.py里核心路由app.route(/search, methods[GET]) def search_messages(): q request.args.get(q, ).strip() sender request.args.get(sender, ) start request.args.get(start, typeint, default0) limit min(request.args.get(limit, typeint, default50), 500) # 硬顶防OOM where_clauses [] params [] if q: where_clauses.append(content LIKE ?) params.append(f%{q}%) if sender: where_clauses.append(sender ?) params.append(sender) sql fSELECT * FROM messages WHERE { AND .join(where_clauses)} ORDER BY timestamp DESC LIMIT ? OFFSET ? params.extend([limit, start]) rows conn.execute(sql, params).fetchall() return jsonify([dict(r) for r in rows])没用POST/PUT/DELETE原因很现实查询是99%场景销售查客户问价记录、客服查投诉关键词、审计查敏感词触发日志写入完全自动化消息捕获模块hooker.py通过sqlite3原生API直连DB不走HTTP安全兜底简单所有参数强制typeint或strip()SQL拼接严格限定WHERE字段连ORDER BY都固化为timestamp DESC杜绝注入。提示如果你需要批量导出源码里export.py提供了--format csv/json和--date-range 2024-05-01 to 2024-05-31参数比写curl更稳。3. 在Windows上跑通WeChatMsgHistory-real从编译依赖到微信版本兼容性实测3.1 必装环境与版本锁死清单亲测有效项目README说“支持Python 3.8”但实际踩坑发现Python必须3.9.13或3.10.11低于3.9.13ctypes对微信v3.9.10.23的WeChat.exe符号解析失败GetProcAddress返回NULL高于3.10.11pywin32的win32event.WaitForSingleObject在高DPI屏下超时异常VC Redistributable必须2015-2022微信客户端用MSVC2019编译钩子DLL需同版本CRT微信PC版必须v3.9.5.22 ~ v3.9.10.23v3.9.0以下无CMessage::Serialize导出符号v3.9.11改用std::shared_ptr管理消息体钩子地址偏移失效。安装命令管理员CMD执行# 1. 安装指定Python推荐pyenv-win pyenv install 3.9.13 pyenv global 3.9.13 # 2. 升级pip并装wheel避免编译报错 python -m pip install --upgrade pip wheel # 3. 一次性装齐所有依赖注意顺序 pip install pywin32306 pypiwin32223 psutil5.9.5 requests2.31.0 flask2.2.5 # ↑↑↑ pypiwin32必须装否则win32event.WaitForSingleObject()找不到dll3.2 编译钩子DLL用MinGW-w64而非MSVC避坑关键源码hooker/目录下有wechat_hook.c但setup.py默认用MSVC编译会失败——因为微信进程开启了DEP/NX保护MSVC生成的DLL默认带/SAFESEH而微信加载器拒绝加载。解决方案# 下载MinGW-w64x86_64-8.1.0-release-posix-seh-rt_v6-rev0.7z # 解压后把bin/加到PATH # 进入hooker/目录执行 gcc -shared -o wechat_hook.dll wechat_hook.c \ -Wl,--no-seh -Wl,--disable-stdcall-fixup \ -IC:\Python39\include \ -LC:\Python39\libs -lpython39关键参数说明--no-seh禁用结构化异常处理绕过微信DEP检查--disable-stdcall-fixup避免函数名修饰错误微信导出符号是_CMessage_Serialize4不是CMessage_Serialize-lpython39链接Python DLL让钩子能调用PyEval_InitThreads()初始化GIL。编译成功后wechat_hook.dll大小应为124KB左右太大说明没禁用SEH太小说明没链接Python库。3.3 启动监听三步确认微信进程被正确注入不要直接双击main.py——先手动验证注入链# 步骤1启动微信PC版确保是v3.9.8.12官网下载历史版本 # 步骤2打开任务管理器找到WeChat.exe进程PID比如12345 # 步骤3用源码里tools/inject_test.py测试注入 python tools/inject_test.py 12345 ./hooker/wechat_hook.dll如果输出[] Inject success, module handle: 0x00007ff...说明DLL已载入如果报错Access denied检查是否以管理员运行如果报错The specified module could not be found说明DLL依赖缺失用Dependency Walker查msvcp140.dll是否在系统PATH。注入成功后再运行python main.py --debug你会看到控制台实时打印[HOOK] CMessage::Serialize called, size128 bytes [PARSE] Text msg from 张三 to 我的文件传输助手: 报价单发你邮箱了 [DB] INSERT INTO messages (msg_id, sender, receiver, content, ...) VALUES (MSG_abc123, ...)只要看到这三行说明从内存捕获→结构化解析→SQLite写入全链路打通。4. macOS上的实操难点mach_inject失效、签名绕过与M1芯片适配4.1 为什么不能直接用Linux/macOS通用injectorWeChatMac版v4.0.8启用AMFIApple Mobile File Integrity内核级签名验证任何未签名的dylib注入都会被kern_invalid拒绝。项目hooker/macos/下的wechat_hook.m用mach_inject方案但实测发现macOS 12.6系统task_for_pid()被彻底禁用mach_inject无法获取目标进程task portM1/M2芯片__TEXT段地址随机化强度翻倍dlsym(RTLD_DEFAULT, _CMessage_Serialize)永远返回NULL。解决方案是改用ptracesysctl组合拳源码hooker/macos/ptrace_hook.c// 1. 先用sysctl获取微信进程内存布局 int mib[3] {CTL_KERN, KERN_PROC, KERN_PROC_PID}; struct kinfo_proc kp; size_t len sizeof(kp); sysctl(mib, 3, kp, len, NULL, 0); // 2. 用ptrace(PTRACE_ATTACH)暂停进程 ptrace(PT_ATTACH, pid, 0, 0); // 3. 读取__DATA段基址微信v4.0.8固定偏移0x100000000 uint64_t data_base kp.ki_addr 0x100000000; // 4. 在data_base0x2A3F0处写入跳转指令jmp hook_func uint8_t jmp_code[] {0xFF, 0x25, 0x02, 0x00, 0x00, 0x00}; // RIP-relative jmp ptrace(PT_WRITE_I, pid, data_base 0x2A3F0, *(long*)jmp_code);这段代码在build_macos.sh里编译为wechat_hook_arm64.dylib必须用codesign -s - --force --deep签名后才能加载。4.2 绕过Gatekeeper的三步签名法免禁用SIP直接sudo spctl --master-disable不行企业环境禁止关SIP。正确做法# 步骤1用Apple Developer证书签名免费个人账户即可 codesign -s Apple Development: youremail.com -f --deep wechat_hook_arm64.dylib # 步骤2给main.py也签名否则Python解释器被拦截 codesign -s Apple Development: youremail.com -f main.py # 步骤3在系统设置→隐私与安全性→完全磁盘访问手动添加Terminal.app和Python.app注意wechat_hook_arm64.dylib签名后otool -L wechat_hook_arm64.dylib必须显示rpath/libpython3.9.dylib不能是绝对路径/usr/local/lib/libpython3.9.dylib否则加载失败。4.3 M1芯片专属坑arm64e指令集与Python ABI不匹配M1 Mac默认Python是arm64架构但微信v4.0.8是arm64e带指针认证。wechat_hook_arm64.dylib若用普通clang编译会因ABI不兼容导致EXC_BAD_ACCESS。修复方法# 必须用Xcode 14.2的clang并加-flag clang -arch arm64e -dynamiclib -o wechat_hook_arm64e.dylib wechat_hook.m \ -framework Foundation \ -I/usr/include/python3.9 \ -L/usr/lib/python3.9/config-3.9-arm64e \ -lpython3.9编译后用file wechat_hook_arm64e.dylib确认输出含arm64e字样否则注入必崩。5. 避坑指南WeChatMsgHistory-real的5个真实翻车现场与解法5.1 现象微信启动后控制台无任何日志psutil.process_iter()也查不到WeChat进程原因微信PC版默认以WeChat.exe运行但某些企业定制版改名为WXWork.exe或WeCom.exehooker.py里硬编码的进程名匹配失败。解决修改hooker.py第42行# 原代码 for proc in psutil.process_iter([name]): if proc.info[name] WeChat.exe: return proc.pid # 改为支持多名称 wechat_names [WeChat.exe, WXWork.exe, WeCom.exe, WeChatApp.exe] if proc.info[name] in wechat_names:5.2 现象消息能捕获但sender字段全是wxid_xxx无法显示昵称原因微信内存中CContact对象的昵称字段在v3.9.7改为延迟加载钩子只读了m_strUsrNamewxid没读m_strNickName。解决在wechat_hook.c的CMessage_Serialize_Hook函数里增加昵称解析逻辑// 在解析完m_strUsrName后追加 char* nick_ptr (char*)msg_obj 0x1A8; // v3.9.8偏移需用IDA Pro确认 if (strlen(nick_ptr) 2 strlen(nick_ptr) 32) { strcpy_s(sender_buf, sizeof(sender_buf), nick_ptr); } else { strcpy_s(sender_buf, sizeof(sender_buf), usrname_buf); // 回退到wxid }5.3 现象SQLite写入速度暴跌INSERT耗时从2ms涨到200ms原因Windows Defender实时扫描history.db文件每次写入都触发全文件扫描。解决临时方案Set-MpPreference -ExclusionPath C:\WeChatMsgHistory\db\PowerShell管理员运行永久方案在db.py的init_db()里加PRAGMAconn.execute(PRAGMA journal_mode WAL) # 提升并发写入 conn.execute(PRAGMA synchronous NORMAL) # 平衡安全与速度 conn.execute(PRAGMA temp_store MEMORY) # 临时表放内存5.4 现象HTTP API返回空数组但history.db里有数据原因Flask默认绑定127.0.0.1:5000而前端JavaScript用localhost:5000请求Chrome 95因SameSite策略拒绝跨域cookie导致session丢失虽然本项目不用session但某些代理会误判。解决启动时显式指定hostpython main.py --host 0.0.0.0 --port 5000 # 并在api.py顶部加CORS头 app.after_request def after_request(response): response.headers[Access-Control-Allow-Origin] * return response5.5 现象macOS上ptrace(PT_ATTACH)返回Operation not permitted原因macOS Monterey系统task_for_pid权限需在entitlements.plist里声明且必须用productsign重签名。解决创建entitlements.plist?xml version1.0 encodingUTF-8? !DOCTYPE plist PUBLIC -//Apple//DTD PLIST 1.0//EN http://www.apple.com/DTDs/PropertyList-1.0.dtd plist version1.0 dict keycom.apple.security.cs.debugger/key true/ /dict /plist重签名codesign -s Apple Development --entitlements entitlements.plist -f wechat_hook_arm64e.dylib6. 进阶技巧用正则预过滤自定义字段扩展把查询系统变成业务规则引擎6.1 在入库前做轻量级ETL用正则提取订单号、手机号、金额WeChatMsgHistory-real默认只存原始content但业务上常要快速定位「含订单号的客户消息」。源码parser.py里parse_message()函数可扩展import re def parse_message(raw_content): result {content: raw_content} # 提取订单号格式PO20240512001 po_match re.search(rPO\d{12}, raw_content) if po_match: result[order_id] po_match.group() # 提取手机号11位连续数字前后非数字 phone_match re.search(r(?!\d)(1[3-9]\d{9})(?!\d), raw_content) if phone_match: result[phone] phone_match.group() # 提取金额¥xxx.xx 或 xxx,xxx.xx money_match re.search(r[¥]\d{1,3}(?:,\d{3})*\.\d{2}, raw_content) if money_match: result[amount] float(money_match.group().replace(,, ).replace(¥, ).replace(, )) return result然后在hooker.py的on_message_captured()里调用parsed parse_message(raw_text) # 写入DB时多插字段 conn.execute( INSERT INTO messages (msg_id, sender, receiver, content, order_id, phone, amount, ...) VALUES (?, ?, ?, ?, ?, ?, ?, ...) , (msg_id, sender, receiver, raw_text, parsed.get(order_id), parsed.get(phone), parsed.get(amount), ...))这样/search?qPO20240512001就能直接命中不用全文扫描。6.2 扩展自定义字段表把「客户等级」「所属销售」等业务属性关联进来项目默认只存消息本身但实际要查「张三VIP客户昨天问的报价」。方案是建contacts_ext扩展表CREATE TABLE contacts_ext ( wxid TEXT PRIMARY KEY, customer_level TEXT CHECK(customer_level IN (VIP, PRO, NORMAL)), sales_owner TEXT, last_contact_time INTEGER, tags TEXT -- JSON数组如 [售后咨询,价格敏感] );然后在api.py的/search路由里加JOINsql SELECT m.*, e.customer_level, e.sales_owner FROM messages m LEFT JOIN contacts_ext e ON m.sender e.wxid WHERE ... 关键点contacts_ext表由业务系统定时同步比如每天凌晨用python sync_contacts.py --from crm_api不依赖微信客户端——这才是真正的「业务可扩展」。6.3 用SQLite FTS5实现毫秒级全文检索替代LIKE %keyword%content LIKE %xxx%在10万条消息时查一次要800ms。换成FTS5-- 创建虚拟表 CREATE VIRTUAL TABLE messages_fts USING fts5(content, sender, receiver, tokenizeporter); -- 建立触发器自动同步 CREATE TRIGGER messages_ai AFTER INSERT ON messages BEGIN INSERT INTO messages_fts(rowid, content, sender, receiver) VALUES (new.id, new.content, new.sender, new.receiver); END; -- 查询时用 SELECT m.* FROM messages m JOIN messages_fts f ON m.id f.rowid WHERE f MATCH 报价 AND 2024;实测100万条消息关键词组合查询稳定在12ms以内且支持NEAR邻近搜索发货单 NEAR/3 快递。我上线这个系统时把messages_fts的tokenize从porter换成unicode61支持中文分词又加了prefix(1,2,3)现在搜「发」、「发货」、「发货单」都能命中——这比教销售同事用CtrlF靠谱多了。希望帮到你。本文还有配套的精品资源点击获取
返回列表