441 lines
17 KiB
Python
441 lines
17 KiB
Python
# -*- coding: utf-8 -*-#
|
||
# -------------------------------------------------------------------------------
|
||
# Name: getwxinfo.py
|
||
# Description:
|
||
# Author: xaoyaoo
|
||
# Date: 2023/08/21
|
||
# -------------------------------------------------------------------------------
|
||
import ctypes
|
||
import json
|
||
import os
|
||
import re
|
||
import winreg
|
||
from typing import List, Union
|
||
from .utils import verify_key, get_exe_bit, wx_core_error
|
||
from .utils import get_process_list, get_memory_maps, get_process_exe_path, get_file_version_info
|
||
from .utils import search_memory
|
||
from .utils import wx_core_loger, CORE_DB_TYPE
|
||
import ctypes.wintypes as wintypes
|
||
|
||
# 定义常量
|
||
PROCESS_QUERY_INFORMATION = 0x0400
|
||
PROCESS_VM_READ = 0x0010
|
||
|
||
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
|
||
OpenProcess = kernel32.OpenProcess
|
||
OpenProcess.restype = wintypes.HANDLE
|
||
OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
|
||
|
||
CloseHandle = kernel32.CloseHandle
|
||
CloseHandle.restype = wintypes.BOOL
|
||
CloseHandle.argtypes = [wintypes.HANDLE]
|
||
|
||
ReadProcessMemory = kernel32.ReadProcessMemory
|
||
void_p = ctypes.c_void_p
|
||
|
||
|
||
# 读取内存中的字符串(key部分)
|
||
@wx_core_error
|
||
def get_key_by_offs(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_string = bytes(key).hex()
|
||
return key_string
|
||
|
||
|
||
# 读取内存中的字符串(非key部分)
|
||
@wx_core_error
|
||
def get_info_string(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
|
||
|
||
|
||
# 读取内存中的字符串(昵称部分name)
|
||
@wx_core_error
|
||
def get_info_name(h_process, address, address_len=8, n_size=64):
|
||
array = ctypes.create_string_buffer(n_size)
|
||
if ReadProcessMemory(h_process, void_p(address), array, n_size, 0) == 0: return None
|
||
address1 = int.from_bytes(array[:address_len], byteorder='little') # 逆序转换为int地址（key地址）
|
||
info_name = get_info_string(h_process, address1, n_size)
|
||
if info_name != None:
|
||
return info_name
|
||
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
|
||
|
||
|
||
# 读取内存中的wxid
|
||
@wx_core_error
|
||
def get_info_wxid(h_process):
|
||
find_num = 100
|
||
addrs = search_memory(h_process, br'\\Msg\\FTSContact', max_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
|
||
|
||
|
||
# 读取内存中的wx_path基于wxid（慢）
|
||
@wx_core_error
|
||
def get_wx_dir_by_wxid(h_process, wxid=""):
|
||
find_num = 10
|
||
addrs = search_memory(h_process, wxid.encode() + br'\\Msg\\FTSContact', max_num=find_num)
|
||
wxid_dir = []
|
||
for addr in addrs:
|
||
win_addr_len = 260
|
||
array = ctypes.create_string_buffer(win_addr_len)
|
||
if ReadProcessMemory(h_process, void_p(addr - win_addr_len + 50), array, win_addr_len, 0) == 0: return None
|
||
array = bytes(array).split(b"\\Msg")[0]
|
||
array = array.split(b"\00")[-1]
|
||
wxid_dir.append(array.decode('utf-8', errors='ignore'))
|
||
wxid_dir = max(wxid_dir, key=wxid_dir.count) if wxid_dir else None
|
||
return wxid_dir
|
||
|
||
|
||
@wx_core_error
|
||
def get_wx_dir_by_reg(wxid="all"):
|
||
"""
|
||
# 读取 wx_dir (微信文件路径) （快）
|
||
:param wxid: 微信id
|
||
:return: 返回wx_dir,if wxid="all" return wx_dir else return wx_dir/wxid
|
||
"""
|
||
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")
|
||
|
||
wx_dir = os.path.join(w_dir, "WeChat Files")
|
||
|
||
if wxid and wxid != "all":
|
||
wxid_dir = os.path.join(wx_dir, wxid)
|
||
return wxid_dir if os.path.exists(wxid_dir) else None
|
||
return wx_dir if os.path.exists(wx_dir) else None
|
||
|
||
|
||
def get_wx_dir(wxid: str = "", Handle=None):
|
||
"""
|
||
综合运用多种方法获取wx_path
|
||
优先调用 get_wx_dir_by_reg (该方法速度快)
|
||
次要调用 get_wx_dir_by_wxid （该方法通过搜索内存进行，速度较慢）
|
||
"""
|
||
if wxid:
|
||
wx_dir = get_wx_dir_by_reg(wxid) if wxid else None
|
||
if wxid is not None and wx_dir is None and Handle: # 通过wxid获取wx_path,如果wx_path为空则通过wxid获取wx_path
|
||
wx_dir = get_wx_dir_by_wxid(Handle, wxid=wxid)
|
||
else:
|
||
wx_dir = get_wx_dir_by_reg()
|
||
return wx_dir
|
||
|
||
|
||
@wx_core_error
|
||
def get_key_by_mem_search(pid, db_path, addr_len):
|
||
"""
|
||
获取key （慢）
|
||
:param pid: 进程id
|
||
:param db_path: 微信数据库路径
|
||
:param addr_len: 地址长度
|
||
:return: 返回key
|
||
"""
|
||
|
||
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
|
||
|
||
phone_type1 = "iphone\x00"
|
||
phone_type2 = "android\x00"
|
||
phone_type3 = "ipad\x00"
|
||
|
||
MicroMsg_path = os.path.join(db_path, "MSG", "MicroMsg.db")
|
||
|
||
start_adress = 0x7FFFFFFFFFFFFFFF
|
||
end_adress = 0
|
||
|
||
memory_maps = get_memory_maps(pid)
|
||
for module in memory_maps:
|
||
if module.FileName and 'WeChatWin.dll' in module.FileName:
|
||
s = module.BaseAddress
|
||
e = module.BaseAddress + module.RegionSize
|
||
start_adress = s if s < start_adress else start_adress
|
||
end_adress = e if e > end_adress else end_adress
|
||
|
||
hProcess = OpenProcess(PROCESS_QUERY_INFORMATION | PROCESS_VM_READ, False, pid)
|
||
type1_addrs = search_memory(hProcess, phone_type1.encode(), max_num=2, start_address=start_adress,
|
||
end_address=end_adress)
|
||
type2_addrs = search_memory(hProcess, phone_type2.encode(), max_num=2, start_address=start_adress,
|
||
end_address=end_adress)
|
||
type3_addrs = search_memory(hProcess, phone_type3.encode(), max_num=2, start_address=start_adress,
|
||
end_address=end_adress)
|
||
|
||
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(hProcess, j, addr_len)
|
||
if key_bytes == None:
|
||
continue
|
||
if verify_key(key_bytes, MicroMsg_path):
|
||
return key_bytes.hex()
|
||
CloseHandle(hProcess)
|
||
return None
|
||
|
||
|
||
@wx_core_error
|
||
def get_wx_key(key: str = "", wx_dir: str = "", pid=0, addrLen=8):
|
||
"""
|
||
获取key （慢）
|
||
:param key: 微信key
|
||
:param wx_dir: 微信文件路径
|
||
:param pid: 进程id
|
||
:param addrLen: 地址长度
|
||
:return: 返回key
|
||
"""
|
||
isKey = verify_key(
|
||
bytes.fromhex(key),
|
||
os.path.join(wx_dir, "MSG", "MicroMsg.db")) if key is not None and wx_dir is not None else False
|
||
if wx_dir is not None and not isKey:
|
||
key = get_key_by_mem_search(pid, wx_dir, addrLen)
|
||
return key
|
||
|
||
|
||
@wx_core_error
|
||
def get_info_details(pid, WX_OFFS: dict = None):
|
||
path = get_process_exe_path(pid)
|
||
rd = {'pid': pid, 'version': get_file_version_info(path),
|
||
"account": None, "mobile": None, "nickname": None, "mail": None,
|
||
"wxid": None, "key": None, "wx_dir": None}
|
||
try:
|
||
bias_list = WX_OFFS.get(rd['version'], None)
|
||
|
||
Handle = OpenProcess(PROCESS_QUERY_INFORMATION | PROCESS_VM_READ, False, pid)
|
||
|
||
addrLen = get_exe_bit(path) // 8
|
||
if not isinstance(bias_list, list) or len(bias_list) <= 4:
|
||
wx_core_loger.warning(f"[-] WeChat Current Version Is Not Supported(not get account,mobile,nickname,mail)")
|
||
else:
|
||
wechat_base_address = 0
|
||
memory_maps = get_memory_maps(pid)
|
||
for module in memory_maps:
|
||
if module.FileName and 'WeChatWin.dll' in module.FileName:
|
||
wechat_base_address = module.BaseAddress
|
||
rd['version'] = get_file_version_info(module.FileName) if os.path.exists(module.FileName) else rd[
|
||
'version']
|
||
bias_list = WX_OFFS.get(rd['version'], None)
|
||
break
|
||
if wechat_base_address != 0:
|
||
name_baseaddr = wechat_base_address + bias_list[0]
|
||
account_baseaddr = wechat_base_address + bias_list[1]
|
||
mobile_baseaddr = wechat_base_address + bias_list[2]
|
||
mail_baseaddr = wechat_base_address + bias_list[3]
|
||
key_baseaddr = wechat_base_address + bias_list[4]
|
||
|
||
rd['account'] = get_info_string(Handle, account_baseaddr, 32) if bias_list[1] != 0 else None
|
||
rd['mobile'] = get_info_string(Handle, mobile_baseaddr, 64) if bias_list[2] != 0 else None
|
||
rd['nickname'] = get_info_name(Handle, name_baseaddr, addrLen, 64) if bias_list[0] != 0 else None
|
||
rd['mail'] = get_info_string(Handle, mail_baseaddr, 64) if bias_list[3] != 0 else None
|
||
rd['key'] = get_key_by_offs(Handle, key_baseaddr, addrLen) if bias_list[4] != 0 else None
|
||
else:
|
||
wx_core_loger.warning(f"[-] WeChat WeChatWin.dll Not Found")
|
||
|
||
rd['wxid'] = get_info_wxid(Handle)
|
||
rd['wx_dir'] = get_wx_dir(rd['wxid'], Handle)
|
||
rd['key'] = get_wx_key(rd['key'], rd['wx_dir'], rd['pid'], addrLen)
|
||
|
||
CloseHandle(Handle)
|
||
except Exception as e:
|
||
wx_core_loger.error(f"[-] WeChat Get Info Error:{e}", exc_info=True)
|
||
return rd
|
||
|
||
|
||
# 读取微信信息(account,mobile,nickname,mail,wxid,key)
|
||
@wx_core_error
|
||
def get_wx_info(WX_OFFS: dict = None, is_print: bool = False, save_path: str = None):
|
||
"""
|
||
读取微信信息(account,mobile,nickname,mail,wxid,key)
|
||
:param WX_OFFS: 版本偏移量
|
||
:param is_print: 是否打印结果
|
||
:param save_path: 保存路径
|
||
:return: 返回微信信息 [{"pid": pid, "version": version, "account": account,
|
||
"mobile": mobile, "nickname": nickname, "mail": mail, "wxid": wxid,
|
||
"key": key, "wx_dir": wx_dir}, ...]
|
||
"""
|
||
if WX_OFFS is None:
|
||
WX_OFFS = {}
|
||
|
||
wechat_pids = []
|
||
result = []
|
||
|
||
processes = get_process_list()
|
||
for pid, name in processes:
|
||
if name == "WeChat.exe":
|
||
wechat_pids.append(pid)
|
||
|
||
if len(wechat_pids) <= 0:
|
||
wx_core_loger.error("[-] WeChat No Run")
|
||
return result
|
||
|
||
for pid in wechat_pids:
|
||
rd = get_info_details(pid, WX_OFFS)
|
||
result.append(rd)
|
||
|
||
if is_print:
|
||
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 if v else 'None'}")
|
||
print(end="-" * 32 + "\n" if i != len(result) - 1 else "")
|
||
print("=" * 32)
|
||
|
||
if save_path:
|
||
try:
|
||
infos = json.load(open(save_path, "r", encoding="utf-8")) if os.path.exists(save_path) else []
|
||
except:
|
||
infos = []
|
||
with open(save_path, "w", encoding="utf-8") as f:
|
||
infos += result
|
||
json.dump(infos, f, ensure_ascii=False, indent=4)
|
||
return result
|
||
|
||
|
||
@wx_core_error
|
||
def get_wx_db(msg_dir: str = None,
|
||
db_types: Union[List[str], str] = None,
|
||
wxids: Union[List[str], str] = None) -> List[dict]:
|
||
r"""
|
||
获取微信数据库路径
|
||
:param msg_dir: 微信数据库目录 eg: C:\Users\user\Documents\WeChat Files （非wxid目录）
|
||
:param db_types: 需要获取的数据库类型,如果为空,则获取所有数据库
|
||
:param wxids: 微信id列表,如果为空,则获取所有wxid下的数据库
|
||
:return: [{"wxid": wxid, "db_type": db_type, "db_path": db_path, "wxid_dir": wxid_dir}, ...]
|
||
"""
|
||
result = []
|
||
|
||
if not msg_dir or not os.path.exists(msg_dir):
|
||
wx_core_loger.warning(f"[-] 微信文件目录不存在: {msg_dir}, 将使用默认路径")
|
||
msg_dir = get_wx_dir_by_reg(wxid="all")
|
||
|
||
if not os.path.exists(msg_dir):
|
||
wx_core_loger.error(f"[-] 目录不存在: {msg_dir}", exc_info=True)
|
||
return result
|
||
|
||
wxids = wxids.split(";") if isinstance(wxids, str) else wxids
|
||
if not isinstance(wxids, list) or len(wxids) <= 0:
|
||
wxids = None
|
||
db_types = db_types.split(";") if isinstance(db_types, str) and db_types else db_types
|
||
if not isinstance(db_types, list) or len(db_types) <= 0:
|
||
db_types = None
|
||
|
||
wxid_dirs = {} # wx用户目录
|
||
if wxids or "All Users" in os.listdir(msg_dir) or "Applet" in os.listdir(msg_dir) or "WMPF" in os.listdir(msg_dir):
|
||
for sub_dir in os.listdir(msg_dir):
|
||
if os.path.isdir(os.path.join(msg_dir, sub_dir)) and sub_dir not in ["All Users", "Applet", "WMPF"]:
|
||
wxid_dirs[os.path.basename(sub_dir)] = os.path.join(msg_dir, sub_dir)
|
||
else:
|
||
wxid_dirs[os.path.basename(msg_dir)] = msg_dir
|
||
for wxid, wxid_dir in wxid_dirs.items():
|
||
if wxids and wxid not in wxids: # 如果指定wxid,则过滤掉其他wxid
|
||
continue
|
||
for root, dirs, files in os.walk(wxid_dir):
|
||
for file_name in files:
|
||
if not file_name.endswith(".db"):
|
||
continue
|
||
db_type = re.sub(r"\d*\.db$", "", file_name)
|
||
if db_types and db_type not in db_types: # 如果指定db_type,则过滤掉其他db_type
|
||
continue
|
||
db_path = os.path.join(root, file_name)
|
||
result.append({"wxid": wxid, "db_type": db_type, "db_path": db_path, "wxid_dir": wxid_dir})
|
||
return result
|
||
|
||
|
||
@wx_core_error
|
||
def get_core_db(wx_path: str, db_types: list = None) -> [dict]:
|
||
"""
|
||
获取聊天消息核心数据库路径
|
||
:param wx_path: 微信文件夹路径 eg：C:\*****\WeChat Files\wxid*******
|
||
:param db_types: 数据库类型 eg: CORE_DB_TYPE，中选择一个或多个
|
||
:return: 返回数据库路径 eg: [{"wxid": wxid, "db_type": db_type, "db_path": db_path, "wxid_dir": wxid_dir}, ...]
|
||
"""
|
||
if not os.path.exists(wx_path):
|
||
return False, f"[-] 目录不存在: {wx_path}"
|
||
|
||
if not db_types:
|
||
db_types = CORE_DB_TYPE
|
||
db_types = [dt for dt in db_types if dt in CORE_DB_TYPE]
|
||
msg_dir = os.path.dirname(wx_path)
|
||
my_wxid = os.path.basename(wx_path)
|
||
wxdbpaths = get_wx_db(msg_dir=msg_dir, db_types=db_types, wxids=my_wxid)
|
||
|
||
if len(wxdbpaths) == 0:
|
||
wx_core_loger.error(f"[-] get_core_db 未获取到数据库路径")
|
||
return False, "未获取到数据库路径"
|
||
return True, wxdbpaths
|
||
|
||
|
||
if __name__ == '__main__':
|
||
from pywxdump import WX_OFFS
|
||
|
||
get_wx_info(WX_OFFS, is_print=True)
|