595 lines
22 KiB
Python
595 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
wxwork_key.py - 从 WXWork.exe 进程内存中提取数据库加密密钥
|
||
============================================================
|
||
策略（多模式扫描 + salt/明文头校验）：
|
||
1. 枚举所有 WXWork.exe 进程
|
||
2. VirtualQueryEx 枚举可读内存区，ReadProcessMemory 分块读取
|
||
3. 模式 A: 32 位 hex 字符串（小写/大写）→ 16 字节
|
||
4. 模式 B: b"raw:" 前缀 + 16 字节二进制
|
||
5. 模式 C: SQLCipher x'...' 语法
|
||
6. 模式 D: 16 字节二进制滑窗（兜底，慢）
|
||
每个候选密钥用真实 db 文件的页 1 明文头做密码学验证（verify_key），
|
||
只有能成功解密出 SQLite 头的候选才被采纳。
|
||
|
||
用法:
|
||
from wxwork_key import extract_keys_from_running, find_wxwork_pids
|
||
keys = extract_keys_from_running(db_paths=[...]) # {db_path: [key_hex,...]}
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import ctypes
|
||
import ctypes.wintypes as wt
|
||
import os
|
||
import re
|
||
import struct
|
||
import sys
|
||
|
||
from wxwork_crypto import verify_key, read_page_header, derive_key_pdf
|
||
|
||
# --- Windows API 常量 ---
|
||
PROCESS_QUERY_INFORMATION = 0x0400
|
||
PROCESS_VM_READ = 0x0010
|
||
MEM_COMMIT = 0x1000
|
||
PAGE_NOACCESS = 0x01
|
||
PAGE_GUARD = 0x100
|
||
PAGE_READWRITE = 0x04
|
||
PAGE_READONLY = 0x02
|
||
PAGE_WRITECOPY = 0x08
|
||
PAGE_EXECUTE_READ = 0x20
|
||
PAGE_EXECUTE_READWRITE = 0x40
|
||
PAGE_EXECUTE_WRITECOPY = 0x80
|
||
|
||
READABLE_PROTECTS = (PAGE_READONLY, PAGE_READWRITE, PAGE_WRITECOPY,
|
||
PAGE_EXECUTE_READ, PAGE_EXECUTE_READWRITE, PAGE_EXECUTE_WRITECOPY)
|
||
|
||
_CHUNK = 1 << 20 # 1MB 读块
|
||
|
||
_HEX32_RE = re.compile(rb"[0-9a-fA-F]{32}")
|
||
_HEX32_X_RE = re.compile(rb"x'[0-9a-fA-F]{32}'")
|
||
_RAW16_RE = re.compile(rb"raw:.{16}") # raw: + 16 字节（可能含不可打印字符）
|
||
|
||
|
||
class _MEMORY_BASIC_INFORMATION(ctypes.Structure):
|
||
_fields_ = [
|
||
("BaseAddress", ctypes.c_void_p),
|
||
("AllocationBase", ctypes.c_void_p),
|
||
("AllocationProtect", wt.DWORD),
|
||
("RegionSize", ctypes.c_size_t),
|
||
("State", wt.DWORD),
|
||
("Protect", wt.DWORD),
|
||
("Type", wt.DWORD),
|
||
]
|
||
|
||
|
||
def find_wxwork_pids(include_web: bool = True, include_extra: bool = True) -> list[int]:
|
||
"""枚举企微相关进程 PID（ctypes，无需 psutil）。
|
||
|
||
include_web=True 时含 WXWorkWeb.exe / WXWorkXNet.exe 等子进程；
|
||
include_extra=True 时含 WeChatAppEx.exe / WXDrive_x64.exe 等周边进程。
|
||
"""
|
||
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
||
|
||
class PROCESSENTRY32W(ctypes.Structure):
|
||
_fields_ = [("dwSize", wt.DWORD), ("cntUsage", wt.DWORD),
|
||
("th32ProcessID", wt.DWORD),
|
||
("th32DefaultHeapID", ctypes.POINTER(ctypes.c_ulong)),
|
||
("th32ModuleID", wt.DWORD), ("cntThreads", wt.DWORD),
|
||
("th32ParentProcessID", wt.DWORD), ("pcPriClassBase", ctypes.c_long),
|
||
("dwFlags", wt.DWORD), ("szExeFile", ctypes.c_wchar * 260)]
|
||
|
||
snap = kernel32.CreateToolhelp32Snapshot(2, 0) # TH32CS_SNAPPROCESS
|
||
if snap == -1:
|
||
return []
|
||
extra = ("wechatappex.exe", "wxdrive_x64.exe") if include_extra else ()
|
||
pids = []
|
||
pe = PROCESSENTRY32W()
|
||
pe.dwSize = ctypes.sizeof(PROCESSENTRY32W)
|
||
ok = kernel32.Process32FirstW(snap, ctypes.byref(pe))
|
||
while ok:
|
||
name = pe.szExeFile.lower()
|
||
if name == "wxwork.exe" or (include_web and name.startswith("wxwork")) or \
|
||
(include_extra and name in extra):
|
||
pids.append(int(pe.th32ProcessID))
|
||
ok = kernel32.Process32NextW(snap, ctypes.byref(pe))
|
||
kernel32.CloseHandle(snap)
|
||
return pids
|
||
|
||
|
||
class _ProcessMemory:
|
||
"""只读进程内存访问器。"""
|
||
|
||
def __init__(self, pid: int):
|
||
self.pid = pid
|
||
self._kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
|
||
self._handle = self._kernel32.OpenProcess(
|
||
PROCESS_QUERY_INFORMATION | PROCESS_VM_READ, False, pid)
|
||
if not self._handle:
|
||
raise PermissionError(f"OpenProcess({pid}) 失败，可能需要管理员权限")
|
||
|
||
def close(self):
|
||
if self._handle:
|
||
self._kernel32.CloseHandle(self._handle)
|
||
self._handle = None
|
||
|
||
def regions(self, rw_only: bool = False):
|
||
"""枚举可读已提交内存区域，返回 [(base_addr, size), ...]。
|
||
|
||
rw_only=True 时仅保留可写区（堆/栈），排除只读映像/资源区，
|
||
密钥为运行时生成通常只在堆/栈，可大幅缩小扫描范围。
|
||
"""
|
||
out = []
|
||
addr = 0
|
||
mbi = _MEMORY_BASIC_INFORMATION()
|
||
while True:
|
||
size = self._kernel32.VirtualQueryEx(self._handle, ctypes.c_void_p(addr),
|
||
ctypes.byref(mbi), ctypes.sizeof(mbi))
|
||
if size == 0:
|
||
break
|
||
base = int(mbi.BaseAddress or 0)
|
||
region_size = int(mbi.RegionSize or 0)
|
||
if region_size > 0 and mbi.State == MEM_COMMIT and \
|
||
(mbi.Protect & PAGE_GUARD) == 0 and \
|
||
(mbi.Protect & PAGE_NOACCESS) == 0 and \
|
||
(mbi.Protect & 0xFF) in READABLE_PROTECTS:
|
||
if rw_only and (mbi.Protect & 0xFF) not in (
|
||
PAGE_READWRITE, PAGE_EXECUTE_READWRITE, PAGE_WRITECOPY):
|
||
pass # 跳过只读区
|
||
else:
|
||
out.append((base, region_size))
|
||
addr = base + region_size
|
||
if addr <= 0:
|
||
break
|
||
return out
|
||
|
||
def read(self, base: int, size: int) -> bytes | None:
|
||
buf = ctypes.create_string_buffer(size)
|
||
nread = ctypes.c_size_t(0)
|
||
ok = self._kernel32.ReadProcessMemory(self._handle, ctypes.c_void_p(base),
|
||
buf, size, ctypes.byref(nread))
|
||
if not ok:
|
||
return None
|
||
return buf.raw[: nread.value]
|
||
|
||
|
||
def _candidates_from_chunk(chunk: bytes):
|
||
"""从一块内存中提取候选 16 字节密钥（多模式）。"""
|
||
seen = set()
|
||
# 模式 A: 32 hex 字符
|
||
for m in _HEX32_RE.finditer(chunk):
|
||
hexs = m.group().decode("ascii")
|
||
try:
|
||
key = bytes.fromhex(hexs)
|
||
except ValueError:
|
||
continue
|
||
if key not in seen:
|
||
seen.add(key)
|
||
yield ("hex", key)
|
||
# 模式 C: x'hex'
|
||
for m in _HEX32_X_RE.finditer(chunk):
|
||
hexs = m.group()[2:-1].decode("ascii")
|
||
try:
|
||
key = bytes.fromhex(hexs)
|
||
except ValueError:
|
||
continue
|
||
if key not in seen:
|
||
seen.add(key)
|
||
yield ("sqlcipher_hex", key)
|
||
# 模式 B: raw: + 16 字节
|
||
for m in _RAW16_RE.finditer(chunk):
|
||
key = m.group()[4:20]
|
||
if key not in seen:
|
||
seen.add(key)
|
||
yield ("raw", key)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 二进制滑窗扫描（模式 D）：企微内存中密钥为二进制 16 字节，需逐窗口验证
|
||
# ---------------------------------------------------------------------------
|
||
_IV_PAGE1 = None
|
||
_PAGE1_KEY_TAIL = struct.pack("<I", 1) + b"sAlT" # 页1 key 派生尾段
|
||
|
||
|
||
def _prep_verify(db_path: str):
|
||
"""预读取 db 页1 的密文块与明文头，返回 (cipher_block, header8) 或 None。
|
||
|
||
cipher_block 必须与 decrypt_page1_block 一致: 8..16 密文A + 24..32 密文B，
|
||
对应解密数据流 data[16:32]（16..24 是明文头，不是密文，不能直接当密文读）。
|
||
"""
|
||
from wxwork_crypto import generate_iv, read_page_header
|
||
global _IV_PAGE1
|
||
if _IV_PAGE1 is None:
|
||
_IV_PAGE1 = generate_iv(1)
|
||
info = read_page_header(db_path)
|
||
if info is None:
|
||
return None
|
||
_, header8 = info
|
||
with open(db_path, "rb") as f:
|
||
f.seek(8)
|
||
part_a = f.read(8) # 8..16: 加密的版本头
|
||
f.seek(24)
|
||
part_b = f.read(8) # 24..32: 加密的后续数据
|
||
if len(part_a) < 8 or len(part_b) < 8:
|
||
return None
|
||
return part_a + part_b, header8
|
||
|
||
|
||
def _quick_verify_multi(key: bytes, prep) -> str | None:
|
||
"""多算法变体验证，返回命中的变体名或 None。
|
||
|
||
变体覆盖 wxSQLite3 新旧实现的页面密钥派生差异：
|
||
- md5/sha256 摘要算法
|
||
- 是否带 "sAlT" 后缀
|
||
- 页号小端/大端
|
||
"""
|
||
import hashlib
|
||
from Crypto.Cipher import AES
|
||
cipher_block, header8 = prep
|
||
pg_le = b"\x01\x00\x00\x00"
|
||
pg_be = b"\x00\x00\x00\x01"
|
||
tail_salt = b"sAlT"
|
||
variants = []
|
||
if len(key) == 16:
|
||
variants = [
|
||
("md5_le_salt", hashlib.md5(key + pg_le + tail_salt).digest()),
|
||
("md5_le", hashlib.md5(key + pg_le).digest()),
|
||
("md5_be", hashlib.md5(key + pg_be).digest()),
|
||
]
|
||
elif len(key) == 32:
|
||
variants = [
|
||
("sha256_le_salt", hashlib.sha256(key + pg_le + tail_salt).digest()),
|
||
("sha256_le", hashlib.sha256(key + pg_le).digest()),
|
||
("sha256_be", hashlib.sha256(key + pg_be).digest()),
|
||
]
|
||
for name, pagekey in variants:
|
||
dec = AES.new(pagekey, AES.MODE_CBC, _IV_PAGE1).decrypt(cipher_block)
|
||
if dec[:8] == header8:
|
||
return name
|
||
return None
|
||
|
||
|
||
# codec 结构体特征: m_legacy(4B) + m_legacyPageSize(4B) + m_keyLength(4B) + m_key[16/32]
|
||
# 小端 int: m_keyLength=16 -> 10 00 00 00, m_keyLength=32 -> 20 00 00 00
|
||
_CODEC128_RE = re.compile(rb".{0,16}\x10\x00\x00\x00(.{16})", re.DOTALL)
|
||
_CODEC256_RE = re.compile(rb".{0,16}\x20\x00\x00\x00(.{32})", re.DOTALL)
|
||
|
||
|
||
def _candidates_codec(chunk: bytes):
|
||
"""从内存块中提取 codec 结构体内的密钥候选。"""
|
||
for m in _CODEC128_RE.finditer(chunk):
|
||
yield ("codec128", m.group(1))
|
||
for m in _CODEC256_RE.finditer(chunk):
|
||
yield ("codec256", m.group(1))
|
||
|
||
|
||
def _read_rw_regions(pm) -> list[tuple[int, int]]:
|
||
"""只保留 PAGE_READWRITE / PAGE_EXECUTE_READWRITE 私有已提交区（堆/栈），
|
||
密钥通常缓存在堆上，可大幅缩小扫描范围。"""
|
||
out = []
|
||
for base, size in pm.regions():
|
||
# regions() 已过滤可读，这里再过滤到可写区
|
||
out.append((base, size))
|
||
return out
|
||
|
||
|
||
def _fast_verify(key: bytes, prep) -> bool:
|
||
"""单变体快速验证：wxSQLite3 标准派生（md5/sha256 + 页号LE + 'sAlT'）。
|
||
|
||
命中后如需确认具体变体再调 _quick_verify_multi（3 次 AES），
|
||
快速路径只做 1 次 AES，扫描性能提升约 3 倍。
|
||
"""
|
||
import hashlib
|
||
from Crypto.Cipher import AES
|
||
cipher_block, header8 = prep
|
||
pg_le = b"\x01\x00\x00\x00"
|
||
if len(key) == 16:
|
||
pagekey = hashlib.md5(key + pg_le + b"sAlT").digest()
|
||
elif len(key) == 32:
|
||
pagekey = hashlib.sha256(key + pg_le + b"sAlT").digest()
|
||
else:
|
||
return False
|
||
dec = AES.new(pagekey, AES.MODE_CBC, _IV_PAGE1).decrypt(cipher_block)
|
||
return dec[:8] == header8
|
||
|
||
|
||
def scan_binary_single(pid: int, prep, aligned: bool = True, key_len: int = 16,
|
||
step: int | None = None, progress_cb=None) -> list[str]:
|
||
"""单进程二进制滑窗扫描，返回验证通过的 key hex 列表。
|
||
|
||
aligned=True 时窗口起始按 step 步进（默认 16 字节对齐）；
|
||
False 时全滑窗（step=1，兜底）。key_len 支持 16(aes128)/32(aes256)。
|
||
step=8 时覆盖 8 字节对齐分配（部分内存池 malloc 只保证 8 对齐）。
|
||
只扫可写区 + 字节多样性预过滤，避免对文本/零填充区做无谓 AES。
|
||
"""
|
||
import hashlib
|
||
from Crypto.Cipher import AES
|
||
if step is None:
|
||
step = 16 if aligned else 1
|
||
found = []
|
||
try:
|
||
pm = _ProcessMemory(pid)
|
||
except PermissionError as exc:
|
||
if progress_cb:
|
||
progress_cb(f"[PID {pid}] 打开失败: {exc}")
|
||
return []
|
||
min_unique = 8 if key_len == 16 else 12 # 随机密钥的字节多样性下界
|
||
try:
|
||
regions = pm.regions(rw_only=True)
|
||
total_regions = len(regions)
|
||
scanned_bytes = 0
|
||
for idx, (base, rsize) in enumerate(regions):
|
||
off = 0
|
||
while off < rsize:
|
||
size = min(_CHUNK, rsize - off)
|
||
data = pm.read(base + off, size)
|
||
if data:
|
||
n = len(data)
|
||
for start in range(0, n - key_len + 1, step):
|
||
w = data[start:start + key_len]
|
||
# 预过滤：字节多样性不足直接跳过
|
||
if len(set(w)) >= min_unique and _fast_verify(w, prep):
|
||
variant = _quick_verify_multi(w, prep) or "fast"
|
||
kh = w.hex()
|
||
if kh not in found:
|
||
found.append(kh)
|
||
if progress_cb:
|
||
progress_cb(f"[PID {pid}] 命中! variant={variant} key={kh}")
|
||
scanned_bytes += n
|
||
off += size
|
||
if progress_cb and idx % 50 == 0:
|
||
progress_cb(f"[PID {pid}] 区域 {idx + 1}/{total_regions} "
|
||
f"(已扫 {scanned_bytes / 1e6:.0f}MB)")
|
||
finally:
|
||
pm.close()
|
||
return found
|
||
|
||
|
||
def _binary_worker(pid, prep, aligned, key_len, step, q):
|
||
keys = scan_binary_single(pid, prep, aligned=aligned, key_len=key_len, step=step,
|
||
progress_cb=None)
|
||
q.put((pid, keys))
|
||
|
||
|
||
def scan_binary_all(db_paths: list[str], aligned: bool = True, key_len: int = 16,
|
||
step: int | None = None, progress_cb=None) -> dict[str, list[str]]:
|
||
"""多进程并行扫描所有 WXWork.exe，返回 {db_path: [key_hex,...]}。
|
||
|
||
注意：Windows 下 multiprocessing 用 spawn，需在 __main__ 保护内调用。
|
||
"""
|
||
import multiprocessing as mp
|
||
|
||
valid = []
|
||
for p in db_paths:
|
||
prep = _prep_verify(p)
|
||
if prep is not None:
|
||
valid.append((p, prep))
|
||
if not valid:
|
||
return {}
|
||
# 取第一个可用 db 做验证基准（其余 db 的密钥通常同源，命中后统一再验）
|
||
db_path, prep = valid[0]
|
||
pids = find_wxwork_pids()
|
||
if not pids:
|
||
return {}
|
||
|
||
ctx = mp.get_context("spawn")
|
||
q = ctx.Queue()
|
||
procs = []
|
||
for pid in pids:
|
||
p = ctx.Process(target=_binary_worker, args=(pid, prep, aligned, key_len, step, q))
|
||
p.start()
|
||
procs.append(p)
|
||
for p in procs:
|
||
p.join()
|
||
all_keys: set[str] = set()
|
||
while not q.empty():
|
||
_, keys = q.get_nowait()
|
||
all_keys.update(keys)
|
||
# 对全部 db 复核
|
||
result: dict[str, set] = {}
|
||
for db_path, _ in valid:
|
||
for kh in all_keys:
|
||
key = bytes.fromhex(kh)
|
||
if verify_key(key, db_path):
|
||
result.setdefault(db_path, set()).add(kh)
|
||
return {p: sorted(s) for p, s in result.items()}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# nkey 缓冲区特征搜索（模式 E）：sqlite3mcAES128 内部构造
|
||
# nkey = master_key || page_le32 || "sAlT" 连续内存，可直接正则捕获
|
||
# ---------------------------------------------------------------------------
|
||
_NKEY128_RE = re.compile(rb"(.{16}).{4}sAlT", re.DOTALL)
|
||
_NKEY256_RE = re.compile(rb"(.{32}).{4}sAlT", re.DOTALL)
|
||
|
||
|
||
def _candidates_nkey(chunk: bytes):
|
||
"""从一块内存中提取 nkey 缓冲区里的主密钥候选（16/32 字节）。"""
|
||
for m in _NKEY128_RE.finditer(chunk):
|
||
yield ("nkey128", m.group(1))
|
||
for m in _NKEY256_RE.finditer(chunk):
|
||
yield ("nkey256", m.group(1))
|
||
|
||
|
||
def scan_nkey_single(pid: int, prep, progress_cb=None) -> list[str]:
|
||
"""单进程特征扫描：nkey 缓冲区 (key||pg||sAlT) + codec 结构体 (m_keyLength||key)。"""
|
||
found = []
|
||
try:
|
||
pm = _ProcessMemory(pid)
|
||
except PermissionError as exc:
|
||
if progress_cb:
|
||
progress_cb(f"[PID {pid}] 打开失败: {exc}")
|
||
return []
|
||
try:
|
||
regions = pm.regions(rw_only=True) # 密钥只在堆/栈，跳过只读映像区
|
||
total = len(regions)
|
||
for idx, (base, rsize) in enumerate(regions):
|
||
off = 0
|
||
while off < rsize:
|
||
size = min(_CHUNK, rsize - off)
|
||
data = pm.read(base + off, size)
|
||
if data:
|
||
for mode, key in _candidates_nkey(data):
|
||
variant = _quick_verify_multi(key, prep)
|
||
if variant:
|
||
found.append(key.hex())
|
||
if progress_cb:
|
||
progress_cb(f"[PID {pid}] 命中! mode={mode} variant={variant} key={key.hex()}")
|
||
for mode, key in _candidates_codec(data):
|
||
variant = _quick_verify_multi(key, prep)
|
||
if variant:
|
||
found.append(key.hex())
|
||
if progress_cb:
|
||
progress_cb(f"[PID {pid}] 命中! mode={mode} variant={variant} key={key.hex()}")
|
||
off += size
|
||
if progress_cb and idx % 200 == 0:
|
||
progress_cb(f"[PID {pid}] 区域 {idx + 1}/{total}")
|
||
finally:
|
||
pm.close()
|
||
return found
|
||
|
||
|
||
def scan_nkey_all(db_paths: list[str], progress_cb=None) -> dict[str, list[str]]:
|
||
"""多进程 nkey 特征扫描所有 WXWork.exe。"""
|
||
import multiprocessing as mp
|
||
valid = [(p, _prep_verify(p)) for p in db_paths if _prep_verify(p) is not None]
|
||
if not valid:
|
||
return {}
|
||
_, prep = valid[0]
|
||
pids = find_wxwork_pids()
|
||
if not pids:
|
||
return {}
|
||
|
||
ctx = mp.get_context("spawn")
|
||
q = ctx.Queue()
|
||
procs = []
|
||
for pid in pids:
|
||
p = ctx.Process(target=_nkey_worker, args=(pid, prep, q))
|
||
p.start()
|
||
procs.append(p)
|
||
for p in procs:
|
||
p.join()
|
||
all_keys: set[str] = set()
|
||
while not q.empty():
|
||
_, keys = q.get_nowait()
|
||
all_keys.update(keys)
|
||
result: dict[str, set] = {}
|
||
for db_path, _ in valid:
|
||
for kh in all_keys:
|
||
if verify_key(bytes.fromhex(kh), db_path):
|
||
result.setdefault(db_path, set()).add(kh)
|
||
return {p: sorted(s) for p, s in result.items()}
|
||
|
||
|
||
def _nkey_worker(pid, prep, q):
|
||
keys = scan_nkey_single(pid, prep, progress_cb=None)
|
||
q.put((pid, keys))
|
||
|
||
|
||
def _scan_pid(pid: int, db_paths: list[str], progress_cb=None) -> dict[str, list[str]]:
|
||
"""扫描单个进程内存，返回 {db_path: [key_hex, ...]}（只含验证通过的密钥）。"""
|
||
found: dict[str, set] = {p: set() for p in db_paths}
|
||
try:
|
||
pm = _ProcessMemory(pid)
|
||
except PermissionError as exc:
|
||
if progress_cb:
|
||
progress_cb(f"[PID {pid}] 打开进程失败: {exc}")
|
||
return {}
|
||
try:
|
||
regions = pm.regions()
|
||
if progress_cb:
|
||
progress_cb(f"[PID {pid}] 可读区域 {len(regions)} 个")
|
||
for idx, (base, rsize) in enumerate(regions):
|
||
off = 0
|
||
while off < rsize:
|
||
size = min(_CHUNK, rsize - off)
|
||
data = pm.read(base + off, size)
|
||
if data:
|
||
for mode, key in _candidates_from_chunk(data):
|
||
for db_path in db_paths:
|
||
if db_path in found and key not in found[db_path] and \
|
||
verify_key(key, db_path):
|
||
found[db_path].add(key)
|
||
if progress_cb:
|
||
progress_cb(f"[PID {pid}] 命中! db={os.path.basename(db_path)} "
|
||
f"mode={mode} key={key.hex()}")
|
||
off += size
|
||
if progress_cb and idx % 200 == 0:
|
||
progress_cb(f"[PID {pid}] 区域 {idx + 1}/{len(regions)} ...")
|
||
finally:
|
||
pm.close()
|
||
return {p: sorted(s) for p, s in found.items() if s}
|
||
|
||
|
||
def extract_keys_from_running(db_paths: list[str], progress_cb=None) -> dict[str, list[str]]:
|
||
"""扫描所有 WXWork.exe 进程，返回 {db_path: [key_hex,...]}。
|
||
|
||
db_paths: 待验证的加密 db 文件绝对路径列表（只需其中任意一个可命中）。
|
||
"""
|
||
db_paths = [p for p in db_paths if read_page_header(p) is not None]
|
||
if not db_paths:
|
||
return {}
|
||
pids = find_wxwork_pids()
|
||
if not pids:
|
||
if progress_cb:
|
||
progress_cb("未发现 WXWork.exe 进程，请先启动并登录企业微信")
|
||
return {}
|
||
results: dict[str, set] = {p: set() for p in db_paths}
|
||
for pid in pids:
|
||
r = _scan_pid(pid, db_paths, progress_cb)
|
||
for db_path, keys in r.items():
|
||
results[db_path].update(keys)
|
||
return {p: sorted(s) for p, s in results.items() if s}
|
||
|
||
|
||
def discover_db_files(data_root: str | None = None) -> list[str]:
|
||
"""自动发现 Documents\\WXWork\\<account>\\Data\\ 下的加密 db 文件。"""
|
||
root = data_root or os.path.join(os.path.expanduser("~"), "Documents", "WXWork")
|
||
dbs = []
|
||
if not os.path.isdir(root):
|
||
return dbs
|
||
for name in sorted(os.listdir(root)):
|
||
data_dir = os.path.join(root, name, "Data")
|
||
if os.path.isdir(data_dir):
|
||
for fname in ("message.db", "user.db", "session.db"):
|
||
p = os.path.join(data_dir, fname)
|
||
if os.path.isfile(p) and read_page_header(p) is not None:
|
||
dbs.append(p)
|
||
return dbs
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import multiprocessing as mp
|
||
mp.freeze_support()
|
||
scan_mode = "text"
|
||
if len(sys.argv) > 1:
|
||
scan_mode = sys.argv[1].lstrip("-")
|
||
dbs = discover_db_files()
|
||
print(f"发现加密 db {len(dbs)} 个:")
|
||
for p in dbs:
|
||
print(" ", p)
|
||
if not dbs:
|
||
sys.exit(1)
|
||
if scan_mode in ("binary", "binary8"):
|
||
step = 8 if scan_mode == "binary8" else 16
|
||
print(f"\n[二进制滑窗扫描] 并行扫描所有 WXWork.exe "
|
||
f"(16+32 字节窗口, {step} 字节对齐) ...")
|
||
result: dict[str, set] = {}
|
||
for kl in (16, 32):
|
||
print(f" -- 窗口 {kl} 字节 ...")
|
||
r = scan_binary_all(dbs, aligned=True, key_len=kl, step=step, progress_cb=print)
|
||
for p, ks in r.items():
|
||
result.setdefault(p, set()).update(ks)
|
||
result = {p: sorted(s) for p, s in result.items()}
|
||
elif scan_mode == "nkey":
|
||
print("\n[nkey 特征扫描] 并行搜索 key||pg||sAlT 缓冲区 ...")
|
||
result = scan_nkey_all(dbs, progress_cb=print)
|
||
else:
|
||
result = extract_keys_from_running(dbs, progress_cb=print)
|
||
print("\n=== 提取结果 ===")
|
||
if not result:
|
||
print("未提取到密钥（可尝试: --nkey / --binary 全滑窗 / 管理员权限）")
|
||
else:
|
||
for db_path, keys in result.items():
|
||
for k in keys:
|
||
print(f"{db_path}\n -> {k}")
|