From 3f727f734bd4336dcb49bf0143ad4a4496ef3e4a Mon Sep 17 00:00:00 2001 From: xaoyaoo Date: Mon, 25 Dec 2023 16:12:39 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96wx=5Finfo=E5=AF=BC=E5=85=A5?= =?UTF-8?q?=EF=BC=8C=E5=88=86=E7=A6=BB=E5=B7=A5=E5=85=B7=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 3 +- pywxdump/{utils => analyzer}/utils.py | 0 pywxdump/wx_info/get_bias_addr.py | 109 +---------- pywxdump/wx_info/get_wx_info.py | 88 +-------- pywxdump/wx_info/simplify_wx_info.py | 253 -------------------------- pywxdump/wx_info/utils.py | 113 ++++++++++++ 6 files changed, 122 insertions(+), 444 deletions(-) rename pywxdump/{utils => analyzer}/utils.py (100%) delete mode 100644 pywxdump/wx_info/simplify_wx_info.py create mode 100644 pywxdump/wx_info/utils.py diff --git a/README.md b/README.md index 0e92257..b11ebe4 100644 --- a/README.md +++ b/README.md @@ -15,12 +15,11 @@ * 欢迎大家提供更多的想法，或者提供代码，一起完善这个项目。 - ### 如有问题，请先查看：[FAQ](https://github.com/xaoyaoo/PyWxDump/tree/master/doc/FAQ.md) 是否有答案，或者关注公众号回复: `FAQ`。 ### 如果是小白，请关注公众号：`逍遥之芯` (右边二维码) ，回复：`PyWxDump` 获取图文教程。 -qq交流群：577704006（左边二维码） or 点击链接加入群聊[pywxdump功能交流](https://s.xaoyo.top/gOLUDl)。 +qq交流群：577704006（左边二维码） or 点击链接加入群聊[pywxdump功能交流](https://s.xaoyo.top/gOLUDl)。 （因为qq群将满，所以进群需要密码，密码请查看[UserGuide.md](https://github.com/xaoyaoo/PyWxDump/tree/master/doc/UserGuide.md)） diff --git a/pywxdump/utils/utils.py b/pywxdump/analyzer/utils.py similarity index 100% rename from pywxdump/utils/utils.py rename to pywxdump/analyzer/utils.py diff --git a/pywxdump/wx_info/get_bias_addr.py b/pywxdump/wx_info/get_bias_addr.py index d62beab..9fe62af 100644 --- a/pywxdump/wx_info/get_bias_addr.py +++ b/pywxdump/wx_info/get_bias_addr.py @@ -5,92 +5,18 @@ # Author: xaoyaoo # Date: 2023/08/22 # ------------------------------------------------------------------------------- -import argparse import ctypes -import hashlib import json -import multiprocessing import os import re import sys - import psutil -from win32com.client import Dispatch -from pymem import Pymem import pymem -import hmac + +from .utils import get_exe_version, get_exe_bit, verify_key ReadProcessMemory = ctypes.windll.kernel32.ReadProcessMemory void_p = ctypes.c_void_p -KEY_SIZE = 32 -DEFAULT_PAGESIZE = 4096 -DEFAULT_ITER = 64000 - - -def validate_key(key, salt, first, mac_salt): - byteKey = hashlib.pbkdf2_hmac("sha1", key, salt, DEFAULT_ITER, KEY_SIZE) - mac_key = hashlib.pbkdf2_hmac("sha1", byteKey, mac_salt, 2, KEY_SIZE) - hash_mac = hmac.new(mac_key, first[:-32], hashlib.sha1) - hash_mac.update(b'\x01\x00\x00\x00') - - if hash_mac.digest() == first[-32:-12]: - return True - else: - return False - - -def get_exe_bit(file_path): - """ - 获取 PE 文件的位数: 32 位或 64 位 - :param file_path: PE 文件路径(可执行文件) - :return: 如果遇到错误则返回 64 - """ - try: - with open(file_path, 'rb') as f: - dos_header = f.read(2) - if dos_header != b'MZ': - print('get exe bit error: Invalid PE file') - return 64 - # Seek to the offset of the PE signature - f.seek(60) - pe_offset_bytes = f.read(4) - pe_offset = int.from_bytes(pe_offset_bytes, byteorder='little') - - # Seek to the Machine field in the PE header - f.seek(pe_offset + 4) - machine_bytes = f.read(2) - machine = int.from_bytes(machine_bytes, byteorder='little') - - if machine == 0x14c: - return 32 - elif machine == 0x8664: - return 64 - else: - print('get exe bit error: Unknown architecture: %s' % hex(machine)) - return 64 - except IOError: - print('get exe bit error: File not found or cannot be opened') - return 64 - - -def get_exe_version(file_path): - """ - 获取 PE 文件的版本号 - :param file_path: PE 文件路径(可执行文件) - :return: 如果遇到错误则返回 - """ - file_version = Dispatch("Scripting.FileSystemObject").GetFileVersion(file_path) - return file_version - - -def find_all(c: bytes, string: bytes, base_addr=0): - """ - 查找字符串中所有子串的位置 - :param c: 子串 b'123' - :param string: 字符串 b'123456789123' - :return: - """ - return [base_addr + m.start() for m in re.finditer(re.escape(c), string)] class BiasAddr: @@ -116,7 +42,7 @@ class BiasAddr: def get_process_handle(self): try: - self.pm = Pymem(self.process_name) + self.pm = pymem.Pymem(self.process_name) self.pm.check_wow64() self.is_WoW64 = self.pm.is_WoW64 self.process_handle = self.pm.process_handle @@ -189,25 +115,6 @@ class BiasAddr: key_bytes = bytes(key) return key_bytes - def verify_key(key, wx_db_path): - KEY_SIZE = 32 - DEFAULT_PAGESIZE = 4096 - DEFAULT_ITER = 64000 - with open(wx_db_path, "rb") as file: - blist = file.read(5000) - salt = blist[:16] - byteKey = hashlib.pbkdf2_hmac("sha1", key, salt, DEFAULT_ITER, KEY_SIZE) - first = blist[16:DEFAULT_PAGESIZE] - - mac_salt = bytes([(salt[i] ^ 58) for i in range(16)]) - mac_key = hashlib.pbkdf2_hmac("sha1", byteKey, mac_salt, 2, KEY_SIZE) - hash_mac = hmac.new(mac_key, first[:-32], hashlib.sha1) - hash_mac.update(b'\x01\x00\x00\x00') - - if hash_mac.digest() != first[-32:-12]: - return False - return True - phone_type1 = "iphone\x00" phone_type2 = "android\x00" phone_type3 = "ipad\x00" @@ -264,15 +171,7 @@ class BiasAddr: return rdata -def get_info_without_key(h_process, address, n_size=64): - array = ctypes.create_string_buffer(n_size) - if ReadProcessMemory(h_process, void_p(address), array, n_size, 0) == 0: return "None" - array = bytes(array).split(b"\x00")[0] if b"\x00" in array else bytes(array) - text = array.decode('utf-8', errors='ignore') - return text.strip() if text.strip() != "" else "None" - - if __name__ == '__main__': - account, mobile, name, key, db_path = "test", "test", "test",None, r"test" + account, mobile, name, key, db_path = "test", "test", "test", None, r"test" bias_addr = BiasAddr(account, mobile, name, key, db_path) bias_addr.run(logging_path=True) diff --git a/pywxdump/wx_info/get_wx_info.py b/pywxdump/wx_info/get_wx_info.py index 1ecb89f..390a589 100644 --- a/pywxdump/wx_info/get_wx_info.py +++ b/pywxdump/wx_info/get_wx_info.py @@ -5,57 +5,19 @@ # Author: xaoyaoo # Date: 2023/08/21 # ------------------------------------------------------------------------------- -import hmac -import hashlib import ctypes import os import re import winreg -import pymem -from win32com.client import Dispatch import psutil -import sys +import pymem from typing import List, Union +from .utils import pattern_scan_all,verify_key,get_exe_version,get_exe_bit ReadProcessMemory = ctypes.windll.kernel32.ReadProcessMemory void_p = ctypes.c_void_p -# 获取exe文件的位数 -def get_exe_bit(file_path): - """ - 获取 PE 文件的位数: 32 位或 64 位 - :param file_path: PE 文件路径(可执行文件) - :return: 如果遇到错误则返回 64 - """ - try: - with open(file_path, 'rb') as f: - dos_header = f.read(2) - if dos_header != b'MZ': - print('get exe bit error: Invalid PE file') - return 64 - # Seek to the offset of the PE signature - f.seek(60) - pe_offset_bytes = f.read(4) - pe_offset = int.from_bytes(pe_offset_bytes, byteorder='little') - - # Seek to the Machine field in the PE header - f.seek(pe_offset + 4) - machine_bytes = f.read(2) - machine = int.from_bytes(machine_bytes, byteorder='little') - - if machine == 0x14c: - return 32 - elif machine == 0x8664: - return 64 - else: - print('get exe bit error: Unknown architecture: %s' % hex(machine)) - return 64 - except IOError: - print('get exe bit error: File not found or cannot be opened') - return 64 - - # 读取内存中的字符串(非key部分) def get_info_without_key(h_process, address, n_size=64): array = ctypes.create_string_buffer(n_size) @@ -65,30 +27,6 @@ def get_info_without_key(h_process, address, n_size=64): return text.strip() if text.strip() != "" else "None" -def pattern_scan_all(handle, pattern, *, return_multiple=False, find_num=100): - next_region = 0 - found = [] - user_space_limit = 0x7FFFFFFF0000 if sys.maxsize > 2 ** 32 else 0x7fff0000 - while next_region < user_space_limit: - try: - next_region, page_found = pymem.pattern.scan_pattern_page( - handle, - next_region, - pattern, - return_multiple=return_multiple - ) - except Exception as e: - print(e) - break - if not return_multiple and page_found: - return page_found - if page_found: - found += page_found - if len(found) > find_num: - break - return found - - def get_info_wxid(h_process): find_num = 100 addrs = pattern_scan_all(h_process, br'\\Msg\\FTSContact', return_multiple=True, find_num=find_num) @@ -167,25 +105,6 @@ def get_key(pid, db_path, addr_len): key_bytes = bytes(key) return key_bytes - def verify_key(key, wx_db_path): - KEY_SIZE = 32 - DEFAULT_PAGESIZE = 4096 - DEFAULT_ITER = 64000 - with open(wx_db_path, "rb") as file: - blist = file.read(5000) - salt = blist[:16] - byteKey = hashlib.pbkdf2_hmac("sha1", key, salt, DEFAULT_ITER, KEY_SIZE) - first = blist[16:DEFAULT_PAGESIZE] - - mac_salt = bytes([(salt[i] ^ 58) for i in range(16)]) - mac_key = hashlib.pbkdf2_hmac("sha1", byteKey, mac_salt, 2, KEY_SIZE) - hash_mac = hmac.new(mac_key, first[:-32], hashlib.sha1) - hash_mac.update(b'\x01\x00\x00\x00') - - if hash_mac.digest() != first[-32:-12]: - return False - return True - phone_type1 = "iphone\x00" phone_type2 = "android\x00" phone_type3 = "ipad\x00" @@ -237,7 +156,7 @@ def read_info(version_list, is_logging=False): tmp_rd = {} tmp_rd['pid'] = process.pid - tmp_rd['version'] = Dispatch("Scripting.FileSystemObject").GetFileVersion(process.exe()) + tmp_rd['version'] = get_exe_version(process.exe()) Handle = ctypes.windll.kernel32.OpenProcess(0x1F0FFF, False, process.pid) @@ -355,4 +274,5 @@ def get_wechat_db(require_list: Union[List[str], str] = "all", msg_dir: str = No if __name__ == '__main__': from pywxdump import VERSION_LIST + read_info(VERSION_LIST, is_logging=True) diff --git a/pywxdump/wx_info/simplify_wx_info.py b/pywxdump/wx_info/simplify_wx_info.py deleted file mode 100644 index 9fc309b..0000000 --- a/pywxdump/wx_info/simplify_wx_info.py +++ /dev/null @@ -1,253 +0,0 @@ -# -*- coding: utf-8 -*-# -# ------------------------------------------------------------------------------- -# Name: simplify_wx_info.py -# Description: -# Author: xaoyaoo -# Date: 2023/12/07 -# ------------------------------------------------------------------------------- -import hmac -import hashlib -import ctypes -import os -import winreg -import pymem -import psutil -import sys - -ReadProcessMemory = ctypes.windll.kernel32.ReadProcessMemory -void_p = ctypes.c_void_p - - -# 获取exe文件的位数 -def get_exe_bit(file_path): - """ - 获取 PE 文件的位数: 32 位或 64 位 - :param file_path: PE 文件路径(可执行文件) - :return: 如果遇到错误则返回 64 - """ - try: - with open(file_path, 'rb') as f: - dos_header = f.read(2) - if dos_header != b'MZ': - print('get exe bit error: Invalid PE file') - return 64 - # Seek to the offset of the PE signature - f.seek(60) - pe_offset_bytes = f.read(4) - pe_offset = int.from_bytes(pe_offset_bytes, byteorder='little') - - # Seek to the Machine field in the PE header - f.seek(pe_offset + 4) - machine_bytes = f.read(2) - machine = int.from_bytes(machine_bytes, byteorder='little') - - if machine == 0x14c: - return 32 - elif machine == 0x8664: - return 64 - else: - print('get exe bit error: Unknown architecture: %s' % hex(machine)) - return 64 - except IOError: - print('get exe bit error: File not found or cannot be opened') - return 64 - - -def pattern_scan_all(handle, pattern, *, return_multiple=False, find_num=100): - next_region = 0 - found = [] - user_space_limit = 0x7FFFFFFF0000 if sys.maxsize > 2 ** 32 else 0x7fff0000 - while next_region < user_space_limit: - try: - next_region, page_found = pymem.pattern.scan_pattern_page( - handle, - next_region, - pattern, - return_multiple=return_multiple - ) - except Exception as e: - print(e) - break - if not return_multiple and page_found: - return page_found - if page_found: - found += page_found - if len(found) > find_num: - break - return found - - -def get_info_wxid(h_process): - find_num = 100 - addrs = pattern_scan_all(h_process, br'\\Msg\\FTSContact', return_multiple=True, find_num=find_num) - wxids = [] - for addr in addrs: - array = ctypes.create_string_buffer(80) - if ReadProcessMemory(h_process, void_p(addr - 30), array, 80, 0) == 0: return "None" - array = bytes(array) # .split(b"\\")[0] - array = array.split(b"\\Msg")[0] - array = array.split(b"\\")[-1] - wxids.append(array.decode('utf-8', errors='ignore')) - wxid = max(wxids, key=wxids.count) if wxids else "None" - return wxid - - -def get_info_filePath(wxid="all"): - if not wxid: - return "None" - w_dir = "MyDocument:" - is_w_dir = False - - try: - key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, r"Software\Tencent\WeChat", 0, winreg.KEY_READ) - value, _ = winreg.QueryValueEx(key, "FileSavePath") - winreg.CloseKey(key) - w_dir = value - is_w_dir = True - except Exception as e: - w_dir = "MyDocument:" - - if not is_w_dir: - try: - user_profile = os.environ.get("USERPROFILE") - path_3ebffe94 = os.path.join(user_profile, "AppData", "Roaming", "Tencent", "WeChat", "All Users", "config", - "3ebffe94.ini") - with open(path_3ebffe94, "r", encoding="utf-8") as f: - w_dir = f.read() - is_w_dir = True - except Exception as e: - w_dir = "MyDocument:" - - if w_dir == "MyDocument:": - try: - # 打开注册表路径 - key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, - r"Software\Microsoft\Windows\CurrentVersion\Explorer\User Shell Folders") - documents_path = winreg.QueryValueEx(key, "Personal")[0] # 读取文档实际目录路径 - winreg.CloseKey(key) # 关闭注册表 - documents_paths = os.path.split(documents_path) - if "%" in documents_paths[0]: - w_dir = os.environ.get(documents_paths[0].replace("%", "")) - w_dir = os.path.join(w_dir, os.path.join(*documents_paths[1:])) - # print(1, w_dir) - else: - w_dir = documents_path - except Exception as e: - profile = os.environ.get("USERPROFILE") - w_dir = os.path.join(profile, "Documents") - - msg_dir = os.path.join(w_dir, "WeChat Files") - - if wxid == "all" and os.path.exists(msg_dir): - return msg_dir - - filePath = os.path.join(msg_dir, wxid) - return filePath if os.path.exists(filePath) else "None" - - -def get_key(pid, db_path, addr_len): - def read_key_bytes(h_process, address, address_len=8): - array = ctypes.create_string_buffer(address_len) - if ReadProcessMemory(h_process, void_p(address), array, address_len, 0) == 0: return "None" - address = int.from_bytes(array, byteorder='little') # 逆序转换为int地址（key地址） - key = ctypes.create_string_buffer(32) - if ReadProcessMemory(h_process, void_p(address), key, 32, 0) == 0: return "None" - key_bytes = bytes(key) - return key_bytes - - def verify_key(key, wx_db_path): - KEY_SIZE = 32 - DEFAULT_PAGESIZE = 4096 - DEFAULT_ITER = 64000 - with open(wx_db_path, "rb") as file: - blist = file.read(5000) - salt = blist[:16] - byteKey = hashlib.pbkdf2_hmac("sha1", key, salt, DEFAULT_ITER, KEY_SIZE) - first = blist[16:DEFAULT_PAGESIZE] - - mac_salt = bytes([(salt[i] ^ 58) for i in range(16)]) - mac_key = hashlib.pbkdf2_hmac("sha1", byteKey, mac_salt, 2, KEY_SIZE) - hash_mac = hmac.new(mac_key, first[:-32], hashlib.sha1) - hash_mac.update(b'\x01\x00\x00\x00') - - if hash_mac.digest() != first[-32:-12]: - return False - return True - - phone_type1 = "iphone\x00" - phone_type2 = "android\x00" - phone_type3 = "ipad\x00" - - pm = pymem.Pymem(pid) - module_name = "WeChatWin.dll" - - MicroMsg_path = os.path.join(db_path, "MSG", "MicroMsg.db") - - type1_addrs = pm.pattern_scan_module(phone_type1.encode(), module_name, return_multiple=True) - type2_addrs = pm.pattern_scan_module(phone_type2.encode(), module_name, return_multiple=True) - type3_addrs = pm.pattern_scan_module(phone_type3.encode(), module_name, return_multiple=True) - - # print(type1_addrs, type2_addrs, type3_addrs) - - type_addrs = [] - if len(type1_addrs) >= 2: type_addrs += type1_addrs - if len(type2_addrs) >= 2: type_addrs += type2_addrs - if len(type3_addrs) >= 2: type_addrs += type3_addrs - if len(type_addrs) == 0: return "None" - - type_addrs.sort() # 从小到大排序 - - for i in type_addrs[::-1]: - for j in range(i, i - 2000, -addr_len): - key_bytes = read_key_bytes(pm.process_handle, j, addr_len) - if key_bytes == "None": - continue - if verify_key(key_bytes, MicroMsg_path): - return key_bytes.hex() - return "None" - - -# 读取微信信息(account,mobile,name,mail,wxid,key) -def read_info(is_logging=False): - wechat_process = [] - result = [] - for process in psutil.process_iter(['name', 'exe', 'pid', 'cmdline']): - if process.name() == 'WeChat.exe': - wechat_process.append(process) - - if len(wechat_process) == 0: - error = "[-] WeChat No Run" - if is_logging: print(error) - return error - - for process in wechat_process: - tmp_rd = {} - - tmp_rd['pid'] = process.pid - # tmp_rd['version'] = Dispatch("Scripting.FileSystemObject").GetFileVersion(process.exe()) - - Handle = ctypes.windll.kernel32.OpenProcess(0x1F0FFF, False, process.pid) - - addrLen = get_exe_bit(process.exe()) // 8 - - tmp_rd['wxid'] = get_info_wxid(Handle) - tmp_rd['filePath'] = get_info_filePath(tmp_rd['wxid']) if tmp_rd['wxid'] != "None" else "None" - tmp_rd['key'] = get_key(tmp_rd['pid'], tmp_rd['filePath'], addrLen) if tmp_rd['filePath'] != "None" else "None" - result.append(tmp_rd) - - - if is_logging: - print("=" * 32) - if isinstance(result, str): # 输出报错 - print(result) - else: # 输出结果 - for i, rlt in enumerate(result): - for k, v in rlt.items(): - print(f"[+] {k:>8}: {v}") - print(end="-" * 32 + "\n" if i != len(result) - 1 else "") - print("=" * 32) - return result - - -if __name__ == '__main__': - read_info(is_logging=True) diff --git a/pywxdump/wx_info/utils.py b/pywxdump/wx_info/utils.py new file mode 100644 index 0000000..57b50ee --- /dev/null +++ b/pywxdump/wx_info/utils.py @@ -0,0 +1,113 @@ +# -*- coding: utf-8 -*-# +# ------------------------------------------------------------------------------- +# Name: utils.py +# Description: +# Author: xaoyaoo +# Date: 2023/12/25 +# ------------------------------------------------------------------------------- +import re +import sys + +import pymem +from win32com.client import Dispatch +import hmac +import hashlib + + +def verify_key(key, wx_db_path): + KEY_SIZE = 32 + DEFAULT_PAGESIZE = 4096 + DEFAULT_ITER = 64000 + with open(wx_db_path, "rb") as file: + blist = file.read(5000) + salt = blist[:16] + byteKey = hashlib.pbkdf2_hmac("sha1", key, salt, DEFAULT_ITER, KEY_SIZE) + first = blist[16:DEFAULT_PAGESIZE] + + mac_salt = bytes([(salt[i] ^ 58) for i in range(16)]) + mac_key = hashlib.pbkdf2_hmac("sha1", byteKey, mac_salt, 2, KEY_SIZE) + hash_mac = hmac.new(mac_key, first[:-32], hashlib.sha1) + hash_mac.update(b'\x01\x00\x00\x00') + + if hash_mac.digest() != first[-32:-12]: + return False + return True + + +def get_exe_version(file_path): + """ + 获取 PE 文件的版本号 + :param file_path: PE 文件路径(可执行文件) + :return: 如果遇到错误则返回 + """ + file_version = Dispatch("Scripting.FileSystemObject").GetFileVersion(file_path) + return file_version + + +def find_all(c: bytes, string: bytes, base_addr=0): + """ + 查找字符串中所有子串的位置 + :param c: 子串 b'123' + :param string: 字符串 b'123456789123' + :return: + """ + return [base_addr + m.start() for m in re.finditer(re.escape(c), string)] + + +# 获取exe文件的位数 +def get_exe_bit(file_path): + """ + 获取 PE 文件的位数: 32 位或 64 位 + :param file_path: PE 文件路径(可执行文件) + :return: 如果遇到错误则返回 64 + """ + try: + with open(file_path, 'rb') as f: + dos_header = f.read(2) + if dos_header != b'MZ': + print('get exe bit error: Invalid PE file') + return 64 + # Seek to the offset of the PE signature + f.seek(60) + pe_offset_bytes = f.read(4) + pe_offset = int.from_bytes(pe_offset_bytes, byteorder='little') + + # Seek to the Machine field in the PE header + f.seek(pe_offset + 4) + machine_bytes = f.read(2) + machine = int.from_bytes(machine_bytes, byteorder='little') + + if machine == 0x14c: + return 32 + elif machine == 0x8664: + return 64 + else: + print('get exe bit error: Unknown architecture: %s' % hex(machine)) + return 64 + except IOError: + print('get exe bit error: File not found or cannot be opened') + return 64 + + +def pattern_scan_all(handle, pattern, *, return_multiple=False, find_num=100): + next_region = 0 + found = [] + user_space_limit = 0x7FFFFFFF0000 if sys.maxsize > 2 ** 32 else 0x7fff0000 + while next_region < user_space_limit: + try: + next_region, page_found = pymem.pattern.scan_pattern_page( + handle, + next_region, + pattern, + return_multiple=return_multiple + ) + except Exception as e: + print(e) + break + if not return_multiple and page_found: + return page_found + if page_found: + found += page_found + if len(found) > find_num: + break + return found