diff --git a/find_image_key.py b/find_image_key.py index f170306..eaa24e7 100644 --- a/find_image_key.py +++ b/find_image_key.py @@ -1,419 +1,492 @@ -"""从微信进程内存中提取图片 AES 密钥 (V2 .dat 格式) - -V2 .dat 文件结构: - [6B signature: 07 08 V2 08 07] [4B aes_size LE] [4B xor_size LE] [1B padding] - [aes_size bytes AES-ECB encrypted] [raw_data unencrypted] [xor_size bytes XOR encrypted] - -AES key: 16-byte ASCII string found in Weixin.exe process memory -XOR key: single byte, same as old format (derived from JPEG FF D9 ending) - -Usage: - 1. 打开微信, 进入聊天/朋友圈, 点击查看 2-3 张图片 - 2. 立即运行: python find_image_key.py -""" -import os -import sys -import re -import struct -import glob -import json -import time -import ctypes -from ctypes import wintypes -from Crypto.Cipher import AES -from Crypto.Util import Padding - -# Windows API constants -PROCESS_ALL_ACCESS = 0x1F0FFF -PROCESS_VM_READ = 0x0010 -PROCESS_QUERY_INFORMATION = 0x0400 -MEM_COMMIT = 0x1000 -PAGE_NOACCESS = 0x01 -PAGE_GUARD = 0x100 -PAGE_READWRITE = 0x04 -PAGE_WRITECOPY = 0x08 -PAGE_EXECUTE_READWRITE = 0x40 -PAGE_EXECUTE_WRITECOPY = 0x80 - -class MEMORY_BASIC_INFORMATION(ctypes.Structure): - _fields_ = [ - ("BaseAddress", ctypes.c_void_p), - ("AllocationBase", ctypes.c_void_p), - ("AllocationProtect", wintypes.DWORD), - ("RegionSize", ctypes.c_size_t), - ("State", wintypes.DWORD), - ("Protect", wintypes.DWORD), - ("Type", wintypes.DWORD), - ] - -kernel32 = ctypes.windll.kernel32 - -# 正则: 精确 32 字符 alphanum (前后是非 alphanum 或边界) -RE_KEY32 = re.compile(rb'(?= 2: - pids.append(int(parts[1])) - return pids - - -def find_v2_ciphertext(attach_dir): - """从多个 V2 .dat 文件中提取第一个 AES 密文块 (16 bytes)""" - v2_magic = b'\x07\x08V2\x08\x07' - - # Search _t.dat (thumbnails, likely JPEG) - pattern = os.path.join(attach_dir, "*", "*", "Img", "*_t.dat") - dat_files = sorted(glob.glob(pattern), key=os.path.getmtime, reverse=True) - - for f in dat_files[:100]: - try: - with open(f, 'rb') as fp: - header = fp.read(31) - if header[:6] == v2_magic and len(header) >= 31: - return header[15:31], os.path.basename(f) - except: - continue - return None, None - - -def find_xor_key(attach_dir): - """从缩略图文件末尾推导 XOR key (JPEG 结尾 FF D9)""" - v2_magic = b'\x07\x08V2\x08\x07' - pattern = os.path.join(attach_dir, "*", "*", "Img", "*_t.dat") - dat_files = sorted(glob.glob(pattern), key=os.path.getmtime, reverse=True) - - tail_counts = {} - for f in dat_files[:32]: - try: - sz = os.path.getsize(f) - with open(f, 'rb') as fp: - head = fp.read(6) - fp.seek(sz - 2) - tail = fp.read(2) - if head == v2_magic and len(tail) == 2: - key = (tail[0], tail[1]) - tail_counts[key] = tail_counts.get(key, 0) + 1 - except: - continue - - if not tail_counts: - return None - - most_common = max(tail_counts, key=tail_counts.get) - x, y = most_common - xor_key = x ^ 0xFF - check = y ^ 0xD9 - - if xor_key == check: - return xor_key - return xor_key # return best guess anyway - - -def try_key(key_bytes, ciphertext): - """Try decrypting ciphertext with key, return format name if successful""" - try: - cipher = AES.new(key_bytes, AES.MODE_ECB) - dec = cipher.decrypt(ciphertext) - if dec[:3] == b'\xFF\xD8\xFF': - return 'JPEG' - if dec[:4] == bytes([0x89, 0x50, 0x4E, 0x47]): - return 'PNG' - if dec[:4] == b'RIFF': - return 'WEBP' - if dec[:4] == b'wxgf': - return 'WXGF' - if dec[:3] == b'GIF': - return 'GIF' - except: - pass - return None - - -def is_rw_protect(protect): - """Check if memory region is readable/writable (where string keys live)""" - rw_flags = (PAGE_READWRITE | PAGE_WRITECOPY | - PAGE_EXECUTE_READWRITE | PAGE_EXECUTE_WRITECOPY) - return (protect & rw_flags) != 0 - - -def scan_memory_for_aes_key(pid, ciphertext): - """扫描微信进程内存寻找 AES key (regex 加速版)""" - access = PROCESS_VM_READ | PROCESS_QUERY_INFORMATION - h_process = kernel32.OpenProcess(access, False, pid) - if not h_process: - print(f" 无法打开进程 {pid} (尝试以管理员运行)", flush=True) - return None - - try: - # Enumerate memory regions - address = 0 - mbi = MEMORY_BASIC_INFORMATION() - rw_regions = [] - all_regions = [] - - while address < 0x7FFFFFFFFFFF: - result = kernel32.VirtualQueryEx( - h_process, ctypes.c_void_p(address), - ctypes.byref(mbi), ctypes.sizeof(mbi) - ) - if result == 0: - break - if (mbi.State == MEM_COMMIT and - mbi.Protect != PAGE_NOACCESS and - (mbi.Protect & PAGE_GUARD) == 0 and - mbi.RegionSize <= 50 * 1024 * 1024): - region = (mbi.BaseAddress, mbi.RegionSize, mbi.Protect) - all_regions.append(region) - if is_rw_protect(mbi.Protect): - rw_regions.append(region) - next_addr = address + mbi.RegionSize - if next_addr <= address: - break - address = next_addr - - rw_mb = sum(r[1] for r in rw_regions) / 1024 / 1024 - all_mb = sum(r[1] for r in all_regions) / 1024 / 1024 - print(f" RW 区域: {len(rw_regions)} ({rw_mb:.0f} MB), 总计: {len(all_regions)} ({all_mb:.0f} MB)", flush=True) - - # Phase 1: 只扫描 RW 区域 (key 字符串最可能在这里) - print(" === Phase 1: 扫描 RW 内存 ===", flush=True) - result = _scan_regions(h_process, rw_regions, ciphertext) - if result: - return result - - # Phase 2: 扫描所有可读区域 - print(" === Phase 2: 扫描所有内存 ===", flush=True) - # 排除已扫描的 RW 区域 - rw_set = set((r[0], r[1]) for r in rw_regions) - other_regions = [r for r in all_regions if (r[0], r[1]) not in rw_set] - result = _scan_regions(h_process, other_regions, ciphertext) - if result: - return result - - return None - - finally: - kernel32.CloseHandle(h_process) - - -def _scan_regions(h_process, regions, ciphertext): - """扫描指定内存区域列表，返回找到的 key 或 None""" - candidates_32 = 0 - candidates_16 = 0 - t0 = time.time() - - for idx, (base_addr, region_size, _protect) in enumerate(regions): - if idx % 100 == 0: - elapsed = time.time() - t0 - print(f" 扫描 {idx}/{len(regions)} ({elapsed:.1f}s)", end='\r', flush=True) - - buffer = ctypes.create_string_buffer(region_size) - bytes_read = ctypes.c_size_t(0) - ok = kernel32.ReadProcessMemory( - h_process, ctypes.c_void_p(base_addr), - buffer, region_size, ctypes.byref(bytes_read) - ) - if not ok or bytes_read.value < 32: - continue - - data = buffer.raw[:bytes_read.value] - - # 用正则找 32 字符 alphanum (C 级速度) - for m in RE_KEY32.finditer(data): - key_bytes = m.group() - candidates_32 += 1 - - # 前 16 字符作为 AES-128 key - fmt = try_key(key_bytes[:16], ciphertext) - if fmt: - key_str = key_bytes.decode('ascii') - print(f"\n*** 找到 AES key (32-char)! → {fmt} ***", flush=True) - print(f" 完整: {key_str}", flush=True) - print(f" AES key: {key_str[:16]}", flush=True) - return key_str[:16] - - # 也试完整 32 字节作 AES-256 - fmt = try_key(key_bytes, ciphertext) - if fmt: - key_str = key_bytes.decode('ascii') - print(f"\n*** 找到 AES key (32-byte)! → {fmt} ***", flush=True) - print(f" 完整: {key_str}", flush=True) - return key_str - - # 也找独立的 16 字符 alphanum - for m in RE_KEY16.finditer(data): - key_bytes = m.group() - candidates_16 += 1 - - fmt = try_key(key_bytes, ciphertext) - if fmt: - key_str = key_bytes.decode('ascii') - print(f"\n*** 找到 AES key (16-char)! → {fmt} ***", flush=True) - print(f" AES key: {key_str}", flush=True) - return key_str - - elapsed = time.time() - t0 - print(f"\n 测试: {candidates_32} x 32-char + {candidates_16} x 16-char ({elapsed:.1f}s)", flush=True) - return None - - -def verify_and_decrypt(attach_dir, aes_key_str, xor_key): - """完整解密一个 V2 文件作为验证""" - v2_magic = b'\x07\x08V2\x08\x07' - key = aes_key_str.encode('ascii')[:16] - - pattern = os.path.join(attach_dir, "*", "*", "Img", "*_t.dat") - dat_files = sorted(glob.glob(pattern), key=os.path.getmtime, reverse=True) - - for f in dat_files[:10]: - try: - with open(f, 'rb') as fp: - data = fp.read() - if data[:6] != v2_magic: - continue - - sig, aes_size, xor_size = struct.unpack_from('<6sLL', data) - - # AES 对齐: 向上取整到 16 的倍数 (PKCS7 填充) - aligned_aes_size = aes_size - aligned_aes_size -= ~(~aligned_aes_size % 16) - - offset = 15 - aes_data = data[offset:offset + aligned_aes_size] - cipher = AES.new(key, AES.MODE_ECB) - dec_aes = Padding.unpad(cipher.decrypt(aes_data), AES.block_size) - offset += aligned_aes_size - - # Raw portion - raw_data = data[offset:len(data) - xor_size] - offset += len(raw_data) - - # XOR portion - xor_data = data[offset:] - dec_xor = bytes(b ^ xor_key for b in xor_data) if xor_key is not None else xor_data - - result = dec_aes + raw_data + dec_xor - - fmt = "unknown" - ext = ".bin" - if result[:3] == b'\xFF\xD8\xFF': - fmt, ext = "JPEG", ".jpg" - elif result[:4] == bytes([0x89, 0x50, 0x4E, 0x47]): - fmt, ext = "PNG", ".png" - elif result[:4] == b'RIFF': - fmt, ext = "WEBP", ".webp" - elif result[:4] == b'wxgf': - fmt, ext = "WXGF", ".hevc" - - print(f" {os.path.basename(f)} -> {fmt} ({len(result):,}B)", flush=True) - - if fmt != "unknown": - out_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "decoded_images") - os.makedirs(out_dir, exist_ok=True) - out_path = os.path.join(out_dir, os.path.splitext(os.path.basename(f))[0] + ext) - with open(out_path, 'wb') as fp: - fp.write(result) - print(f" saved: {out_path}", flush=True) - return True - except Exception as e: - continue - return False - - -def main(): - from config import _config_file_path, load_config - config_path = _config_file_path() - # 加载完整配置用于逻辑，读取原始 JSON 用于保存 - config = load_config() - try: - with open(config_path, encoding="utf-8") as f: - config_raw = json.load(f) - except (FileNotFoundError, json.JSONDecodeError): - config_raw = {} - - db_dir = os.path.expanduser(os.path.expandvars(config['db_dir'])) - base_dir = os.path.dirname(db_dir) - attach_dir = os.path.join(base_dir, 'msg', 'attach') - - # 1. XOR key - print("=== XOR Key ===", flush=True) - xor_key = find_xor_key(attach_dir) - if xor_key is not None: - print(f"XOR key: 0x{xor_key:02x}", flush=True) - - # 2. V2 ciphertext - print("\n=== V2 ciphertext ===", flush=True) - ciphertext, ct_file = find_v2_ciphertext(attach_dir) - if ciphertext is None: - print("No V2 .dat files found") - return - print(f"File: {ct_file}", flush=True) - print(f"Cipher: {ciphertext.hex()}", flush=True) - - # 3. Check if already have key in config - if config.get('image_aes_key'): - print(f"\nExisting image_aes_key: {config['image_aes_key']}", flush=True) - fmt = try_key(config['image_aes_key'].encode('ascii')[:16], ciphertext) - if fmt: - print(f"Key valid! -> {fmt}", flush=True) - print("\n=== Verify decrypt ===", flush=True) - verify_and_decrypt(attach_dir, config['image_aes_key'], xor_key) - return - else: - print("Saved key invalid, re-scanning...", flush=True) - - # 4. Scan memory - print("\n=== Scanning WeChat process memory ===", flush=True) - pids = get_wechat_pids() - if not pids: - print("WeChat not running!") - return - print(f"PIDs: {pids}", flush=True) - print("Tip: View 2-3 images in WeChat first, then run this script immediately\n", flush=True) - - aes_key = None - for pid in pids: - print(f"Scanning PID {pid}...", flush=True) - aes_key = scan_memory_for_aes_key(pid, ciphertext) - if aes_key: - break - - if aes_key: - print(f"\n=== Result ===", flush=True) - print(f"AES key: {aes_key}", flush=True) - print(f"XOR key: 0x{xor_key:02x}" if xor_key is not None else "XOR key: unknown", flush=True) - - config['image_aes_key'] = aes_key - if xor_key is not None: - config['image_xor_key'] = xor_key - config_raw['image_aes_key'] = aes_key - if xor_key is not None: - config_raw['image_xor_key'] = xor_key - with open(config_path, 'w', encoding="utf-8") as f: - json.dump(config_raw, f, indent=2, ensure_ascii=False) - print(f"Saved to {config_path}", flush=True) - - print("\n=== Verify decrypt ===", flush=True) - verify_and_decrypt(attach_dir, aes_key, xor_key) - else: - print("\nAES key not found!", flush=True) - print("Steps:", flush=True) - print(" 1. Login WeChat and keep it running", flush=True) - print(" 2. Open Moments or a chat, view 2-3 images (tap to open full size)", flush=True) - print(" 3. Immediately re-run this script", flush=True) - - -if __name__ == '__main__': - main() +"""从微信进程内存中提取图片 AES 密钥 (V2 .dat 格式) — 跨平台版 + +平台支持: + - Linux: /proc/pid/mem 读取内存 + - Windows: Win32 API (ReadProcessMemory) + - macOS: 请使用 find_image_key_macos.py + +V2 .dat 文件结构: + [6B signature: 07 08 V2 08 07] [4B aes_size LE] [4B xor_size LE] [1B padding] + [aes_size bytes AES-ECB encrypted] [raw_data unencrypted] [xor_size bytes XOR encrypted] + +AES key: 16-byte ASCII string found in WeChat process memory +XOR key: single byte, same as old format (derived from JPEG FF D9 ending) + +Usage: + 1. 打开微信, 进入聊天/朋友圈, 点击查看 2-3 张图片 + 2. 立即运行: python find_image_key.py (Windows) + sudo python3 find_image_key.py (Linux) +""" +import os +import sys +import re +import struct +import glob +import json +import time + +# ── 平台判断: Linux 走 /proc/pid/mem, Windows 走 Win32 API ────────────────── +IS_LINUX = sys.platform.startswith('linux') +IS_WINDOWS = sys.platform == 'win32' + +from Crypto.Cipher import AES +from Crypto.Util import Padding + +# 正则: 精确 32 字符 alphanum (前后是非 alphanum 或边界) +RE_KEY32 = re.compile(rb'(?= 2: + pids.append(int(parts[1])) + return pids + + +def find_v2_ciphertext(attach_dir): + """从多个 V2 .dat 文件中提取第一个 AES 密文块 (16 bytes)""" + pattern = os.path.join(attach_dir, "*", "*", "Img", "*_t.dat") + dat_files = sorted(glob.glob(pattern), key=os.path.getmtime, reverse=True) + if not dat_files: + pattern = os.path.join(attach_dir, "**", "*.dat") + dat_files = sorted(glob.glob(pattern, recursive=True), key=os.path.getmtime, reverse=True) + + for f in dat_files[:100]: + try: + with open(f, 'rb') as fp: + header = fp.read(31) + if header[:6] == V2_MAGIC and len(header) >= 31: + return header[15:31], os.path.basename(f) + except: + continue + return None, None + + +def find_xor_key(attach_dir): + """从缩略图文件末尾推导 XOR key (JPEG 结尾 FF D9)""" + pattern = os.path.join(attach_dir, "*", "*", "Img", "*_t.dat") + dat_files = sorted(glob.glob(pattern), key=os.path.getmtime, reverse=True) + if not dat_files: + pattern = os.path.join(attach_dir, "**", "*_t.dat") + dat_files = sorted(glob.glob(pattern, recursive=True), key=os.path.getmtime, reverse=True) + + tail_counts = {} + for f in dat_files[:32]: + try: + sz = os.path.getsize(f) + with open(f, 'rb') as fp: + head = fp.read(6) + fp.seek(sz - 2) + tail = fp.read(2) + if head == V2_MAGIC and len(tail) == 2: + key = (tail[0], tail[1]) + tail_counts[key] = tail_counts.get(key, 0) + 1 + except: + continue + + if not tail_counts: + return None + most_common = max(tail_counts, key=tail_counts.get) + return most_common[0] ^ 0xFF + + +def try_key(key_bytes, ciphertext): + """Try decrypting ciphertext with key, return format name if successful""" + try: + cipher = AES.new(key_bytes[:16], AES.MODE_ECB) + dec = cipher.decrypt(ciphertext) + if dec[:3] == b'\xFF\xD8\xFF': + return 'JPEG' + if dec[:4] == bytes([0x89, 0x50, 0x4E, 0x47]): + return 'PNG' + if dec[:4] == b'RIFF': + return 'WEBP' + if dec[:4] == b'wxgf': + return 'WXGF' + if dec[:3] == b'GIF': + return 'GIF' + except: + pass + return None + + +# ── Linux 内存扫描 ──────────────────────────────────────────────────────────── +def scan_proc_mem(pid, ciphertext): + """Linux: 扫描 /proc/pid/mem 寻找 AES key""" + maps_path = f"/proc/{pid}/maps" + mem_path = f"/proc/{pid}/mem" + + try: + with open(maps_path, 'r') as f: + maps = f.readlines() + except PermissionError: + print(f" 权限不足, 请用 sudo 运行") + return None + + regions = [] + for line in maps: + parts = line.split() + addr_range = parts[0] + perms = parts[1] if len(parts) > 1 else '' + if 'r' not in perms: + continue + start_s, end_s = addr_range.split('-') + start = int(start_s, 16) + end = int(end_s, 16) + size = end - start + if size > 100 * 1024 * 1024: + continue + is_rw = 'w' in perms + regions.append((start, size, is_rw)) + + rw_regions = [r for r in regions if r[2]] + other_regions = [r for r in regions if not r[2]] + print(f" RW 区域: {len(rw_regions)}, 只读区域: {len(other_regions)}") + + try: + mem_fd = os.open(mem_path, os.O_RDONLY) + except PermissionError: + print(f" 无法读取 /proc/{pid}/mem, 请用 sudo 运行") + return None + + t0 = time.time() + candidates_32 = 0 + candidates_16 = 0 + + print(" === Phase 1: 扫描 RW 内存 ===") + for phase_regions in [rw_regions, other_regions]: + for idx, (start, size, _rw) in enumerate(phase_regions): + if idx % 200 == 0 and idx > 0: + print(f" 扫描 {idx}/{len(phase_regions)} ({time.time()-t0:.1f}s)", end='\r') + try: + os.lseek(mem_fd, start, os.SEEK_SET) + data = os.read(mem_fd, size) + except: + continue + if len(data) < 32: + continue + for m in RE_KEY32.finditer(data): + key_bytes = m.group() + candidates_32 += 1 + fmt = try_key(key_bytes[:16], ciphertext) + if fmt: + key_str = key_bytes.decode('ascii') + print(f"\n*** 找到 AES key (32-char)! -> {fmt} ***") + print(f" 完整: {key_str}") + print(f" AES key: {key_str[:16]}") + os.close(mem_fd) + return key_str[:16] + for m in RE_KEY16.finditer(data): + key_bytes = m.group() + candidates_16 += 1 + fmt = try_key(key_bytes, ciphertext) + if fmt: + key_str = key_bytes.decode('ascii') + print(f"\n*** 找到 AES key (16-char)! -> {fmt} ***") + print(f" AES key: {key_str}") + os.close(mem_fd) + return key_str + if phase_regions == rw_regions: + print(f"\n Phase 1 完成: {candidates_32} x 32char + {candidates_16} x 16char") + print(" === Phase 2: 扫描只读内存 ===") + else: + print(f"\n Phase 2 完成: {candidates_32} x 32char + {candidates_16} x 16char") + + os.close(mem_fd) + print(f" 总耗时: {time.time()-t0:.1f}s") + return None + + +# ── Windows 内存扫描 ────────────────────────────────────────────────────────── +def scan_memory_for_aes_key_win(pid, ciphertext): + """Windows: 扫描微信进程内存寻找 AES key""" + import ctypes + from ctypes import wintypes + + PROCESS_VM_READ = 0x0010 + PROCESS_QUERY_INFORMATION = 0x0400 + MEM_COMMIT = 0x1000 + PAGE_NOACCESS = 0x01 + PAGE_GUARD = 0x100 + PAGE_READWRITE = 0x04 + PAGE_WRITECOPY = 0x08 + PAGE_EXECUTE_READWRITE = 0x40 + PAGE_EXECUTE_WRITECOPY = 0x80 + + class MEMORY_BASIC_INFORMATION(ctypes.Structure): + _fields_ = [ + ("BaseAddress", ctypes.c_void_p), + ("AllocationBase", ctypes.c_void_p), + ("AllocationProtect", wintypes.DWORD), + ("RegionSize", ctypes.c_size_t), + ("State", wintypes.DWORD), + ("Protect", wintypes.DWORD), + ("Type", wintypes.DWORD), + ] + + kernel32 = ctypes.windll.kernel32 + access = PROCESS_VM_READ | PROCESS_QUERY_INFORMATION + h_process = kernel32.OpenProcess(access, False, pid) + if not h_process: + print(f" 无法打开进程 {pid} (尝试以管理员运行)", flush=True) + return None + + try: + address = 0 + mbi = MEMORY_BASIC_INFORMATION() + rw_regions = [] + all_regions = [] + + while address < 0x7FFFFFFFFFFF: + result = kernel32.VirtualQueryEx( + h_process, ctypes.c_void_p(address), + ctypes.byref(mbi), ctypes.sizeof(mbi) + ) + if result == 0: + break + if (mbi.State == MEM_COMMIT and + mbi.Protect != PAGE_NOACCESS and + (mbi.Protect & PAGE_GUARD) == 0 and + mbi.RegionSize <= 50 * 1024 * 1024): + region = (mbi.BaseAddress, mbi.RegionSize, mbi.Protect) + all_regions.append(region) + rw_flags = (PAGE_READWRITE | PAGE_WRITECOPY | + PAGE_EXECUTE_READWRITE | PAGE_EXECUTE_WRITECOPY) + if (mbi.Protect & rw_flags) != 0: + rw_regions.append(region) + next_addr = address + mbi.RegionSize + if next_addr <= address: + break + address = next_addr + + rw_mb = sum(r[1] for r in rw_regions) / 1024 / 1024 + all_mb = sum(r[1] for r in all_regions) / 1024 / 1024 + print(f" RW 区域: {len(rw_regions)} ({rw_mb:.0f} MB), 总计: {len(all_regions)} ({all_mb:.0f} MB)", flush=True) + + print(" === Phase 1: 扫描 RW 内存 ===", flush=True) + result = _scan_regions_win(h_process, kernel32, rw_regions, ciphertext) + if result: + return result + + print(" === Phase 2: 扫描所有内存 ===", flush=True) + rw_set = set((r[0], r[1]) for r in rw_regions) + other_regions = [r for r in all_regions if (r[0], r[1]) not in rw_set] + result = _scan_regions_win(h_process, kernel32, other_regions, ciphertext) + if result: + return result + return None + finally: + kernel32.CloseHandle(h_process) + + +def _scan_regions_win(h_process, kernel32, regions, ciphertext): + """Windows: 扫描指定内存区域列表""" + candidates_32 = 0 + candidates_16 = 0 + t0 = time.time() + + for idx, (base_addr, region_size, _protect) in enumerate(regions): + if idx % 100 == 0: + print(f" 扫描 {idx}/{len(regions)} ({time.time()-t0:.1f}s)", end='\r', flush=True) + buffer = ctypes.create_string_buffer(region_size) + bytes_read = ctypes.c_size_t(0) + ok = kernel32.ReadProcessMemory( + h_process, ctypes.c_void_p(base_addr), + buffer, region_size, ctypes.byref(bytes_read) + ) + if not ok or bytes_read.value < 32: + continue + data = buffer.raw[:bytes_read.value] + + for m in RE_KEY32.finditer(data): + key_bytes = m.group() + candidates_32 += 1 + fmt = try_key(key_bytes[:16], ciphertext) + if fmt: + key_str = key_bytes.decode('ascii') + print(f"\n*** 找到 AES key (32-char)! -> {fmt} ***", flush=True) + print(f" AES key: {key_str[:16]}", flush=True) + return key_str[:16] + + for m in RE_KEY16.finditer(data): + key_bytes = m.group() + candidates_16 += 1 + fmt = try_key(key_bytes, ciphertext) + if fmt: + key_str = key_bytes.decode('ascii') + print(f"\n*** 找到 AES key (16-char)! -> {fmt} ***", flush=True) + print(f" AES key: {key_str}", flush=True) + return key_str + + print(f"\n 测试: {candidates_32} x 32-char + {candidates_16} x 16-char ({time.time()-t0:.1f}s)", flush=True) + return None + + +def scan_memory_for_aes_key(pid, ciphertext): + """跨平台内存扫描入口""" + if IS_LINUX: + return scan_proc_mem(pid, ciphertext) + else: + return scan_memory_for_aes_key_win(pid, ciphertext) + + +def verify_and_decrypt(attach_dir, aes_key_str, xor_key): + """完整解密一个 V2 文件作为验证""" + key = aes_key_str.encode('ascii')[:16] + pattern = os.path.join(attach_dir, "*", "*", "Img", "*_t.dat") + dat_files = sorted(glob.glob(pattern), key=os.path.getmtime, reverse=True) + if not dat_files: + pattern = os.path.join(attach_dir, "**", "*.dat") + dat_files = sorted(glob.glob(pattern, recursive=True), key=os.path.getmtime, reverse=True) + + for f in dat_files[:10]: + try: + with open(f, 'rb') as fp: + data = fp.read() + if data[:6] != V2_MAGIC: + continue + aes_size, xor_size = struct.unpack_from(' {fmt} ({len(result):,}B)", flush=True) + + if fmt != "unknown": + out_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "decoded_images") + os.makedirs(out_dir, exist_ok=True) + out_path = os.path.join(out_dir, os.path.splitext(os.path.basename(f))[0] + ext) + with open(out_path, 'wb') as fp: + fp.write(result) + print(f" saved: {out_path}", flush=True) + return True + except: + continue + return False + + +def main(): + from config import _config_file_path, load_config + config_path = _config_file_path() + config = load_config() + try: + with open(config_path, encoding="utf-8") as f: + config_raw = json.load(f) + except (FileNotFoundError, json.JSONDecodeError): + config_raw = {} + + db_dir = os.path.expanduser(os.path.expandvars(config['db_dir'])) + base_dir = os.path.dirname(db_dir) + attach_dir = os.path.join(base_dir, 'msg', 'attach') + + print(f"attach_dir: {attach_dir}") + + # 1. XOR key + print("=== XOR Key ===", flush=True) + xor_key = find_xor_key(attach_dir) + if xor_key is not None: + print(f"XOR key: 0x{xor_key:02x}", flush=True) + + # 2. V2 ciphertext + print("\n=== V2 ciphertext ===", flush=True) + ciphertext, ct_file = find_v2_ciphertext(attach_dir) + if ciphertext is None: + print("No V2 .dat files found") + print("提示: 在微信里打开聊天, 点开 2-3 张图片, 然后立即重新运行此脚本") + return + print(f"File: {ct_file}", flush=True) + print(f"Cipher: {ciphertext.hex()}", flush=True) + + # 3. Check if already have key in config + if config.get('image_aes_key'): + print(f"\nExisting image_aes_key: {config['image_aes_key']}", flush=True) + fmt = try_key(config['image_aes_key'].encode('ascii')[:16], ciphertext) + if fmt: + print(f"Key valid! -> {fmt}", flush=True) + print("\n=== Verify decrypt ===", flush=True) + verify_and_decrypt(attach_dir, config['image_aes_key'], xor_key) + return + else: + print("Saved key invalid, re-scanning...", flush=True) + + # 4. Scan memory + print("\n=== Scanning WeChat process memory ===", flush=True) + pids = find_wechat_pids() + if not pids: + print("WeChat not running!") + return + print(f"PIDs: {pids}", flush=True) + print("Tip: View 2-3 images in WeChat first, then run this script immediately\n", flush=True) + + aes_key = None + for pid in pids: + print(f"Scanning PID {pid}...", flush=True) + aes_key = scan_memory_for_aes_key(pid, ciphertext) + if aes_key: + break + + if aes_key: + print(f"\n=== Result ===", flush=True) + print(f"AES key: {aes_key}", flush=True) + print(f"XOR key: 0x{xor_key:02x}" if xor_key is not None else "XOR key: unknown", flush=True) + + config['image_aes_key'] = aes_key + if xor_key is not None: + config['image_xor_key'] = xor_key + config_raw['image_aes_key'] = aes_key + if xor_key is not None: + config_raw['image_xor_key'] = xor_key + with open(config_path, 'w', encoding="utf-8") as f: + json.dump(config_raw, f, indent=2, ensure_ascii=False) + print(f"Saved to {config_path}", flush=True) + + print("\n=== Verify decrypt ===", flush=True) + verify_and_decrypt(attach_dir, aes_key, xor_key) + else: + print("\nAES key not found!", flush=True) + print("Steps:", flush=True) + print(" 1. Login WeChat and keep it running", flush=True) + print(" 2. Open Moments or a chat, view 2-3 images (tap to open full size)", flush=True) + print(" 3. Immediately re-run this script", flush=True) + + +if __name__ == '__main__': + main() diff --git a/find_image_key_linux.py b/find_image_key_linux.py new file mode 100644 index 0000000..e5376a5 --- /dev/null +++ b/find_image_key_linux.py @@ -0,0 +1,342 @@ +#!/usr/bin/env python3 +"""Linux 版图片 AES 密钥提取（参考 find_image_key.py 的 Windows 版改写） + +原理: + 1. 从 V2 .dat 文件提取 AES 密文块 (16 字节) + 2. 读 /proc/pid/mem 搜索微信进程内存中的 16/32 字符字母数字串 + 3. 用候选字符串尝试 AES-ECB 解密, 解密出 JPEG/PNG 头即为正确 key + +用法: + sudo python3 find_image_key_linux.py +""" +import os +import re +import sys +import json +import glob +import struct +import ctypes.util + +# pycryptodome +from Crypto.Cipher import AES +from Crypto.Util import Padding + +V2_MAGIC = b'\x07\x08V2\x08\x07' +RE_KEY32 = re.compile(rb'(?= 31: + return header[15:31], os.path.basename(f) + except: + continue + return None, None + + +def find_xor_key(attach_dir): + """从缩略图末尾推导 XOR key (JPEG 结尾 FF D9)""" + pattern = os.path.join(attach_dir, "*", "*", "Img", "*_t.dat") + dat_files = sorted(glob.glob(pattern), key=os.path.getmtime, reverse=True) + if not dat_files: + pattern = os.path.join(attach_dir, "**", "*_t.dat") + dat_files = sorted(glob.glob(pattern, recursive=True), key=os.path.getmtime, reverse=True) + + tail_counts = {} + for f in dat_files[:32]: + try: + sz = os.path.getsize(f) + with open(f, 'rb') as fp: + head = fp.read(6) + fp.seek(sz - 2) + tail = fp.read(2) + if head == V2_MAGIC and len(tail) == 2: + key = (tail[0], tail[1]) + tail_counts[key] = tail_counts.get(key, 0) + 1 + except: + continue + + if not tail_counts: + return None + most_common = max(tail_counts, key=tail_counts.get) + return most_common[0] ^ 0xFF + + +def try_key(key_bytes, ciphertext): + """尝试解密, 返回格式名或 None""" + try: + cipher = AES.new(key_bytes[:16], AES.MODE_ECB) + dec = cipher.decrypt(ciphertext) + if dec[:3] == b'\xFF\xD8\xFF': + return 'JPEG' + if dec[:4] == bytes([0x89, 0x50, 0x4E, 0x47]): + return 'PNG' + if dec[:4] == b'RIFF': + return 'WEBP' + if dec[:4] == b'wxgf': + return 'WXGF' + if dec[:3] == b'GIF': + return 'GIF' + except: + pass + return None + + +def scan_proc_mem(pid, ciphertext): + """扫描 /proc/pid/mem 寻找 AES key""" + maps_path = f"/proc/{pid}/maps" + mem_path = f"/proc/{pid}/mem" + + try: + with open(maps_path, 'r') as f: + maps = f.readlines() + except PermissionError: + print(f" 权限不足, 请用 sudo 运行") + return None + + regions = [] + for line in maps: + parts = line.split() + addr_range = parts[0] + perms = parts[1] if len(parts) > 1 else '' + if 'r' not in perms: + continue + start_s, end_s = addr_range.split('-') + start = int(start_s, 16) + end = int(end_s, 16) + size = end - start + if size > 100 * 1024 * 1024: # 跳过超大区域 + continue + # 只扫描可写区域 (key 字符串最可能在这里) + is_rw = 'w' in perms + regions.append((start, size, is_rw)) + + rw_regions = [r for r in regions if r[2]] + other_regions = [r for r in regions if not r[2]] + print(f" RW 区域: {len(rw_regions)}, 只读区域: {len(other_regions)}") + + try: + mem_fd = os.open(mem_path, os.O_RDONLY) + except PermissionError: + print(f" 无法读取 /proc/{pid}/mem, 请用 sudo 运行") + return None + + import time + t0 = time.time() + candidates_32 = 0 + candidates_16 = 0 + + # Phase 1: RW 区域 + print(" === Phase 1: 扫描 RW 内存 ===") + for phase_regions in [rw_regions, other_regions]: + for idx, (start, size, _rw) in enumerate(phase_regions): + if idx % 200 == 0 and idx > 0: + print(f" 扫描 {idx}/{len(phase_regions)} ({time.time()-t0:.1f}s)", end='\r') + + try: + os.lseek(mem_fd, start, os.SEEK_SET) + data = os.read(mem_fd, size) + except: + continue + + if len(data) < 32: + continue + + # 搜索 32 字符字母数字串 + for m in RE_KEY32.finditer(data): + key_bytes = m.group() + candidates_32 += 1 + fmt = try_key(key_bytes[:16], ciphertext) + if fmt: + key_str = key_bytes.decode('ascii') + print(f"\n*** 找到 AES key (32-char)! -> {fmt} ***") + print(f" 完整: {key_str}") + print(f" AES key: {key_str[:16]}") + os.close(mem_fd) + return key_str[:16] + + # 搜索 16 字符字母数字串 + for m in RE_KEY16.finditer(data): + key_bytes = m.group() + candidates_16 += 1 + fmt = try_key(key_bytes, ciphertext) + if fmt: + key_str = key_bytes.decode('ascii') + print(f"\n*** 找到 AES key (16-char)! -> {fmt} ***") + print(f" AES key: {key_str}") + os.close(mem_fd) + return key_str + + if phase_regions == rw_regions: + print(f"\n Phase 1 完成: {candidates_32} x 32char + {candidates_16} x 16char") + print(" === Phase 2: 扫描只读内存 ===") + else: + print(f"\n Phase 2 完成: {candidates_32} x 32char + {candidates_16} x 16char") + + os.close(mem_fd) + print(f" 总耗时: {time.time()-t0:.1f}s") + return None + + +def verify_decrypt(attach_dir, aes_key_str, xor_key): + """验证解密一个 V2 文件""" + key = aes_key_str.encode('ascii')[:16] + pattern = os.path.join(attach_dir, "*", "*", "Img", "*_t.dat") + dat_files = sorted(glob.glob(pattern), key=os.path.getmtime, reverse=True) + if not dat_files: + pattern = os.path.join(attach_dir, "**", "*.dat") + dat_files = sorted(glob.glob(pattern, recursive=True), key=os.path.getmtime, reverse=True) + + for f in dat_files[:10]: + try: + with open(f, 'rb') as fp: + data = fp.read() + if data[:6] != V2_MAGIC: + continue + aes_size, xor_size = struct.unpack_from(' {fmt} ({len(result):,}B)") + print(f" saved: {out_path}") + return True + except: + continue + return False + + +def main(): + # 加载配置 + config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json") + try: + with open(config_path, encoding="utf-8") as f: + config = json.load(f) + except: + print("无法读取 config.json, 请先运行 find_all_keys_linux.py") + return + + db_dir = os.path.expanduser(os.path.expandvars(config.get('db_dir', ''))) + base_dir = os.path.dirname(db_dir) + attach_dir = os.path.join(base_dir, 'msg', 'attach') + + print(f"attach_dir: {attach_dir}") + + # 1. XOR key + print("\n=== XOR Key ===") + xor_key = find_xor_key(attach_dir) + if xor_key is not None: + print(f"XOR key: 0x{xor_key:02x}") + + # 2. V2 密文 + print("\n=== V2 ciphertext ===") + ciphertext, ct_file = find_v2_ciphertext(attach_dir) + if ciphertext is None: + print("未找到 V2 .dat 文件") + print("提示: 在微信里打开聊天, 点开 2-3 张图片, 然后立即重新运行此脚本") + return + print(f"File: {ct_file}") + print(f"Cipher: {ciphertext.hex()}") + + # 3. 检查已有 key + existing_key = config.get('image_aes_key') + if existing_key: + print(f"\n已有 image_aes_key: {existing_key}") + fmt = try_key(existing_key.encode('ascii')[:16], ciphertext) + if fmt: + print(f"Key 有效! -> {fmt}") + print("\n=== 验证解密 ===") + verify_decrypt(attach_dir, existing_key, xor_key) + return + else: + print("已保存的 key 无效, 重新扫描...") + + # 4. 扫描内存 + print("\n=== 扫描微信进程内存 ===") + pids = find_wechat_pid() + if not pids: + print("微信未运行! 请先启动微信并登录") + return + print(f"PIDs: {pids}") + print("提示: 先在微信里点开 2-3 张图片, 然后立即运行此脚本\n") + + aes_key = None + for pid in pids: + print(f"扫描 PID {pid}...") + aes_key = scan_proc_mem(pid, ciphertext) + if aes_key: + break + + if aes_key: + print(f"\n=== 结果 ===") + print(f"AES key: {aes_key}") + if xor_key is not None: + print(f"XOR key: 0x{xor_key:02x}") + + config['image_aes_key'] = aes_key + if xor_key is not None: + config['image_xor_key'] = xor_key + with open(config_path, 'w', encoding="utf-8") as f: + json.dump(config, f, indent=2, ensure_ascii=False) + print(f"已保存到 {config_path}") + + print("\n=== 验证解密 ===") + verify_decrypt(attach_dir, aes_key, xor_key) + else: + print("\n未找到 AES key!") + print("步骤:") + print(" 1. 启动微信并登录") + print(" 2. 打开聊天或朋友圈, 点开 2-3 张图片 (点开大图)") + print(" 3. 立即重新运行: sudo python3 find_image_key_linux.py") + + +if __name__ == '__main__': + main()