diff --git a/README.md b/README.md index f8dc93f..9fec6cc 100644 --- a/README.md +++ b/README.md @@ -1,44 +1,56 @@ +

+ 微信数据库解密工具 +

+
- 微信数据库解密工具 +

WeChatDataAnalysis - 微信数据库解密与分析工具

+

一个专门用于微信4.x版本数据库解密的工具

+ Version + Stars + Forks + License + Python + FastAPI + Vue.js + SQLite
-# 微信数据库解密工具 - -一个专门用于微信4.x版本数据库解密的工具 - ## 界面预览 -### 检测页面 + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
首页检测页面
首页微信检测页面
解密页面图片密钥页面
数据库解密页面图片密钥页面
图片解密页面解密成功页面
图片解密页面解密成功页面
聊天记录页面
聊天记录页面
-
- 微信检测页面 -
- -自动检测微信安装路径和数据库文件位置，支持多账户识别。 - -### 解密页面 - -
- 数据库解密页面 -
- -输入解密密钥，选择数据库文件进行批量解密操作。 - -### 解密成功页面 - -
- 解密成功页面 -
- -解密完成后显示统计信息，可直接跳转查看聊天记录。 - -### 聊天记录页面 - -
- 聊天记录页面 -
- -> **注意**: 聊天记录页面目前仅完成了基础展示功能，包括消息列表、文本/图片/语音等基本消息类型的显示。更多功能（如搜索、导出、高级筛选等）尚在开发中，当前界面不代表最终成品。 +> **Note**: 聊天记录页面目前仅完成了基础展示功能，更多功能（搜索、导出、高级筛选等）尚在开发中。 ## 功能特性 @@ -164,7 +176,7 @@ uv run analyze_wechat_databases.py #### 1. 获取图片解密密钥 ```bash -# GET请求获取密钥（需要微信正在运行以提取AES密钥） +# GET请求获取密钥（需要微信正在运行；部分版本需以管理员身份运行后端才能提取AES密钥） curl http://localhost:8000/api/media/keys # 强制重新提取密钥 @@ -243,4 +255,4 @@ curl http://localhost:8000/api/media/resource/{md5} --- -**免责声明**: 本工具仅供学习研究使用，使用者需自行承担使用风险。开发者不对因使用本工具造成的任何损失负责。 \ No newline at end of file +**免责声明**: 本工具仅供学习研究使用，使用者需自行承担使用风险。开发者不对因使用本工具造成的任何损失负责。 diff --git a/frontend/pages/decrypt.vue b/frontend/pages/decrypt.vue index 54970bb..1a370db 100644 --- a/frontend/pages/decrypt.vue +++ b/frontend/pages/decrypt.vue @@ -133,15 +133,27 @@
XOR 密钥 - +
AES 密钥 - - {{ mediaKeys.aes_key ? mediaKeys.aes_key.substring(0, 8) + '...' : '未获取' }} - +
@@ -151,6 +163,13 @@ {{ mediaKeys.message }} + +
+ + + + {{ copyMessage }} +
@@ -300,7 +319,7 @@

可能的失败原因：

@@ -411,6 +430,8 @@ const mediaKeys = reactive({ message: '' }) const mediaLoading = ref(false) +const copyMessage = ref('') +let copyMessageTimer = null // 图片解密相关 const mediaDecryptResult = ref(null) @@ -527,6 +548,53 @@ const fetchMediaKeys = async (forceExtract = false) => { } } +const _copyToClipboard = async (text) => { + if (!process.client || typeof window === 'undefined') return false + if (!text) return false + + try { + if (navigator?.clipboard?.writeText) { + await navigator.clipboard.writeText(text) + return true + } + } catch (e) { + // Ignore and fallback below + } + + try { + const textarea = document.createElement('textarea') + textarea.value = text + textarea.setAttribute('readonly', '') + textarea.style.position = 'fixed' + textarea.style.opacity = '0' + textarea.style.left = '-9999px' + textarea.style.top = '0' + document.body.appendChild(textarea) + textarea.select() + textarea.setSelectionRange(0, textarea.value.length) + const ok = document.execCommand('copy') + document.body.removeChild(textarea) + return ok + } catch (e) { + return false + } +} + +const _setCopyMessage = (message) => { + copyMessage.value = message + if (copyMessageTimer) clearTimeout(copyMessageTimer) + copyMessageTimer = setTimeout(() => { + copyMessage.value = '' + copyMessageTimer = null + }, 2000) +} + +const copyKey = async (label, value) => { + if (!value) return + const ok = await _copyToClipboard(value) + _setCopyMessage(ok ? `${label}已复制` : `${label}复制失败，请手动复制`) +} + // 批量解密所有图片（使用SSE实时进度） const decryptAllImages = async () => { mediaDecrypting.value = true @@ -633,4 +701,4 @@ onMounted(() => { } } }) - \ No newline at end of file + diff --git a/frontend/public/home.png b/frontend/public/home.png new file mode 100644 index 0000000..82bbaef Binary files /dev/null and b/frontend/public/home.png differ diff --git a/frontend/public/imageAES.png b/frontend/public/imageAES.png new file mode 100644 index 0000000..e3ecb64 Binary files /dev/null and b/frontend/public/imageAES.png differ diff --git a/frontend/public/imageSucces.png b/frontend/public/imageSucces.png new file mode 100644 index 0000000..4f7e799 Binary files /dev/null and b/frontend/public/imageSucces.png differ diff --git a/frontend/public/message.png b/frontend/public/message.png index 5203e1c..a693d5d 100644 Binary files a/frontend/public/message.png and b/frontend/public/message.png differ diff --git a/src/wechat_decrypt_tool/media_helpers.py b/src/wechat_decrypt_tool/media_helpers.py index b87ed23..13c885c 100644 --- a/src/wechat_decrypt_tool/media_helpers.py +++ b/src/wechat_decrypt_tool/media_helpers.py @@ -13,6 +13,7 @@ from concurrent.futures import ThreadPoolExecutor from functools import lru_cache from pathlib import Path from typing import Any, Optional +from ctypes import wintypes from fastapi import HTTPException @@ -999,6 +1000,14 @@ def _verify_wechat_aes_key(ciphertext: bytes, key16: bytes) -> bool: return True if plain.startswith(b"\x89PNG\r\n\x1a\n"): return True + if plain.startswith(b"GIF87a") or plain.startswith(b"GIF89a"): + return True + if plain.startswith(b"wxgf"): + return True + if len(plain) >= 12 and plain.startswith(b"RIFF") and plain[8:12] == b"WEBP": + return True + if len(plain) >= 8 and plain[4:8] == b"ftyp": + return True return False except Exception: return False @@ -1016,28 +1025,112 @@ class _MEMORY_BASIC_INFORMATION(ctypes.Structure): ] -def _find_weixin_pid() -> Optional[int]: +def _find_weixin_pids() -> list[int]: if psutil is None: - return None - for p in psutil.process_iter(["name"]): + return [] + + preferred = ["weixin.exe", "wechat.exe", "wechatappex.exe", "wechatapp.exe"] + preferred_set = set(preferred) + pids_by_name: dict[str, list[int]] = {n: [] for n in preferred} + extra: list[int] = [] + + for p in psutil.process_iter(["pid", "name"]): try: name = (p.info.get("name") or "").lower() - if name in {"weixin.exe", "wechat.exe"}: - return int(p.pid) + pid = int(p.info.get("pid") or 0) except Exception: continue - return None + if pid <= 0: + continue + + if name in preferred_set: + pids_by_name[name].append(pid) + continue + + if name.startswith("wechat") or name.startswith("weixin"): + extra.append(pid) + + ordered: list[int] = [] + for n in preferred: + ordered.extend(pids_by_name.get(n, [])) + ordered.extend(extra) + + seen: set[int] = set() + out: list[int] = [] + for pid in ordered: + if pid in seen: + continue + seen.add(pid) + out.append(pid) + return out + + +def _try_enable_windows_debug_privilege() -> None: + if os.name != "nt": + return + + try: + advapi32 = ctypes.windll.advapi32 + kernel32 = ctypes.windll.kernel32 + + TOKEN_ADJUST_PRIVILEGES = 0x0020 + TOKEN_QUERY = 0x0008 + SE_PRIVILEGE_ENABLED = 0x0002 + + class _LUID(ctypes.Structure): + _fields_ = [("LowPart", wintypes.DWORD), ("HighPart", wintypes.LONG)] + + class _LUID_AND_ATTRIBUTES(ctypes.Structure): + _fields_ = [("Luid", _LUID), ("Attributes", wintypes.DWORD)] + + class _TOKEN_PRIVILEGES(ctypes.Structure): + _fields_ = [("PrivilegeCount", wintypes.DWORD), ("Privileges", _LUID_AND_ATTRIBUTES * 1)] + + token = wintypes.HANDLE() + if not advapi32.OpenProcessToken( + kernel32.GetCurrentProcess(), + TOKEN_ADJUST_PRIVILEGES | TOKEN_QUERY, + ctypes.byref(token), + ): + return + + try: + luid = _LUID() + if not advapi32.LookupPrivilegeValueW(None, "SeDebugPrivilege", ctypes.byref(luid)): + return + + tp = _TOKEN_PRIVILEGES() + tp.PrivilegeCount = 1 + tp.Privileges[0].Luid = luid + tp.Privileges[0].Attributes = SE_PRIVILEGE_ENABLED + advapi32.AdjustTokenPrivileges(token, False, ctypes.byref(tp), 0, None, None) + finally: + kernel32.CloseHandle(token) + except Exception: + return def _extract_wechat_aes_key_from_process(ciphertext: bytes) -> Optional[bytes]: - pid = _find_weixin_pid() - if not pid: + _try_enable_windows_debug_privilege() + pids = _find_weixin_pids() + if not pids: return None PROCESS_VM_READ = 0x0010 PROCESS_QUERY_INFORMATION = 0x0400 MEM_COMMIT = 0x1000 MEM_PRIVATE = 0x20000 + MEM_MAPPED = 0x40000 + MEM_IMAGE = 0x1000000 + + PAGE_NOACCESS = 0x01 + PAGE_READONLY = 0x02 + PAGE_READWRITE = 0x04 + PAGE_WRITECOPY = 0x08 + PAGE_EXECUTE_READ = 0x20 + PAGE_EXECUTE_READWRITE = 0x40 + PAGE_EXECUTE_WRITECOPY = 0x80 + PAGE_GUARD = 0x100 kernel32 = ctypes.windll.kernel32 @@ -1063,68 +1156,98 @@ def _extract_wechat_aes_key_from_process(ciphertext: bytes) -> Optional[bytes]: CloseHandle.argtypes = [ctypes.c_void_p] CloseHandle.restype = ctypes.c_bool - handle = OpenProcess(PROCESS_QUERY_INFORMATION | PROCESS_VM_READ, False, pid) - if not handle: - return None + readable_mask = ( + PAGE_READONLY + | PAGE_READWRITE + | PAGE_WRITECOPY + | PAGE_EXECUTE_READ + | PAGE_EXECUTE_READWRITE + | PAGE_EXECUTE_WRITECOPY + ) - stop = threading.Event() - result: list[Optional[bytes]] = [None] - pattern = re.compile(rb"[^a-z0-9]([a-z0-9]{32})[^a-z0-9]", flags=re.IGNORECASE) + def is_readable(protect: int) -> bool: + if protect & PAGE_GUARD: + return False + if protect & PAGE_NOACCESS: + return False + return bool(protect & readable_mask) - def read_mem(addr: int, size: int) -> Optional[bytes]: - buf = ctypes.create_string_buffer(size) - read = ctypes.c_size_t(0) - ok = ReadProcessMemory(handle, ctypes.c_void_p(addr), buf, size, ctypes.byref(read)) - if not ok or read.value <= 0: + pattern = re.compile(rb"(?i)(? Optional[bytes]: + handle = OpenProcess(PROCESS_QUERY_INFORMATION | PROCESS_VM_READ, False, pid) + if not handle: return None - return buf.raw[: read.value] - def scan_region(base: int, region_size: int) -> Optional[bytes]: - chunk = 4 * 1024 * 1024 - offset = 0 - tail = b"" - while offset < region_size and not stop.is_set(): - to_read = min(chunk, region_size - offset) - b = read_mem(base + offset, int(to_read)) - if not b: + stop = threading.Event() + result: list[Optional[bytes]] = [None] + + def read_mem(addr: int, size: int) -> Optional[bytes]: + buf = ctypes.create_string_buffer(size) + read = ctypes.c_size_t(0) + ok = ReadProcessMemory(handle, ctypes.c_void_p(addr), buf, size, ctypes.byref(read)) + if not ok or read.value <= 0: return None - data = tail + b - for m in pattern.finditer(data): - cand32 = m.group(1) - cand16 = cand32[:16] - if _verify_wechat_aes_key(ciphertext, cand16): - return cand16 - tail = data[-64:] if len(data) > 64 else data - offset += to_read - return None + return buf.raw[: read.value] - regions: list[tuple[int, int]] = [] - mbi = _MEMORY_BASIC_INFORMATION() - addr = 0 - try: - while VirtualQueryEx(handle, ctypes.c_void_p(addr), ctypes.byref(mbi), ctypes.sizeof(mbi)): - try: - if int(mbi.State) == MEM_COMMIT and int(mbi.Type) == MEM_PRIVATE: - base = int(mbi.BaseAddress) - size = int(mbi.RegionSize) - if size > 0: - regions.append((base, size)) - addr = int(mbi.BaseAddress) + int(mbi.RegionSize) - except Exception: - addr += 0x1000 - if addr <= 0: - break + def scan_region(base: int, region_size: int) -> Optional[bytes]: + chunk = 4 * 1024 * 1024 + offset = 0 + tail = b"" + while offset < region_size and not stop.is_set(): + to_read = min(chunk, region_size - offset) + b = read_mem(base + offset, int(to_read)) + if not b: + return None + data = tail + b + for m in pattern.finditer(data): + cand = m.group(1) + if len(cand) == 16: + candidates = [cand] + else: + candidates = [cand[:16], cand[16:]] + for cand16 in candidates: + if _verify_wechat_aes_key(ciphertext, cand16): + return cand16 + tail = data[-64:] if len(data) > 64 else data + offset += to_read + return None - with ThreadPoolExecutor(max_workers=min(32, max(1, len(regions)))) as ex: - for found in ex.map(lambda r: scan_region(r[0], r[1]), regions): - if found: - result[0] = found - stop.set() + regions: list[tuple[int, int]] = [] + mbi = _MEMORY_BASIC_INFORMATION() + addr = 0 + try: + while VirtualQueryEx(handle, ctypes.c_void_p(addr), ctypes.byref(mbi), ctypes.sizeof(mbi)): + try: + if int(mbi.State) == MEM_COMMIT and int(mbi.Type) in {MEM_PRIVATE, MEM_MAPPED, MEM_IMAGE}: + protect = int(mbi.Protect) + if is_readable(protect): + base = int(mbi.BaseAddress) + size = int(mbi.RegionSize) + if size > 0: + regions.append((base, size)) + addr = int(mbi.BaseAddress) + int(mbi.RegionSize) + except Exception: + addr += 0x1000 + if addr <= 0: break - finally: - CloseHandle(handle) - return result[0] + with ThreadPoolExecutor(max_workers=min(32, max(1, len(regions)))) as ex: + for found in ex.map(lambda r: scan_region(r[0], r[1]), regions): + if found: + result[0] = found + stop.set() + break + finally: + CloseHandle(handle) + + return result[0] + + for pid in pids: + found = scan_pid(pid) + if found: + return found + return None def _save_media_keys(account_dir: Path, xor_key: int, aes_key16: bytes) -> None: diff --git a/src/wechat_decrypt_tool/media_key_finder.py b/src/wechat_decrypt_tool/media_key_finder.py index 1d09de6..d32cae2 100644 --- a/src/wechat_decrypt_tool/media_key_finder.py +++ b/src/wechat_decrypt_tool/media_key_finder.py @@ -73,7 +73,19 @@ def _verify(encrypted: bytes, key: bytes) -> bool: aes_key = key[:16] cipher = AES.new(aes_key, AES.MODE_ECB) text = cipher.decrypt(encrypted) - return bool(text.startswith(b"\xff\xd8\xff")) + if text.startswith(b"\xff\xd8\xff"): + return True + if text.startswith(b"\x89PNG\r\n\x1a\n"): + return True + if text.startswith(b"GIF87a") or text.startswith(b"GIF89a"): + return True + if text.startswith(b"wxgf"): + return True + if len(text) >= 12 and text.startswith(b"RIFF") and text[8:12] == b"WEBP": + return True + if len(text) >= 8 and text[4:8] == b"ftyp": + return True + return False def _search_memory_chunk(process_handle, base_address: int, region_size: int, encrypted: bytes, rules): @@ -101,7 +113,7 @@ def _get_aes_key(encrypted: bytes, pid: int) -> Any: rules_key = r""" rule AesKey { strings: - $pattern = /[^a-z0-9][a-z0-9]{32}[^a-z0-9]/ + $pattern = /[^0-9a-z]([0-9a-z]{16}|[0-9a-z]{32})[^0-9a-z]/ nocase condition: $pattern } diff --git a/src/wechat_decrypt_tool/routers/media.py b/src/wechat_decrypt_tool/routers/media.py index 40dc7e7..d124727 100644 --- a/src/wechat_decrypt_tool/routers/media.py +++ b/src/wechat_decrypt_tool/routers/media.py @@ -98,7 +98,7 @@ async def get_media_keys(account: Optional[str] = None, force_extract: bool = Fa # 保存密钥到缓存 _save_media_keys(account_dir, xor_key, aes_key16) else: - aes_message = "无法从微信进程提取AES密钥（微信是否正在运行？）" + aes_message = "无法从微信进程提取AES密钥（请确认微信正在运行，并尝试以管理员身份运行后端；部分新版微信可能暂不兼容）" else: aes_message = "未找到V2加密模板文件" else: