312 lines
8.4 KiB
Python
312 lines
8.4 KiB
Python
"""
|
||
NoveLoader - DLL 加载器和函数封装
|
||
|
||
基于个微大客户版 Loader.dll 的 Python 封装
|
||
"""
|
||
|
||
import ctypes
|
||
import os
|
||
from ctypes import WinDLL, create_string_buffer, WINFUNCTYPE
|
||
from typing import Callable
|
||
from loguru import logger
|
||
|
||
|
||
def c_string(data: str) -> ctypes.c_char_p:
|
||
"""将 Python 字符串转换为 C 字符串"""
|
||
return ctypes.c_char_p(data.encode('utf-8'))
|
||
|
||
|
||
def create_shared_memory():
|
||
"""创建共享内存（用于DLL通信）"""
|
||
try:
|
||
kernel32 = ctypes.WinDLL('kernel32')
|
||
|
||
# 创建文件映射
|
||
file_handle = kernel32.CreateFileMappingA(
|
||
-1,
|
||
None,
|
||
4, # PAGE_READWRITE
|
||
0,
|
||
33,
|
||
"windows_shell_global__".encode('utf-8')
|
||
)
|
||
|
||
if not file_handle:
|
||
logger.warning("创建共享内存失败")
|
||
return None, None
|
||
|
||
# 映射到内存
|
||
data_address = kernel32.MapViewOfFile(
|
||
file_handle,
|
||
983071, # FILE_MAP_ALL_ACCESS
|
||
0,
|
||
0,
|
||
0
|
||
)
|
||
|
||
if not data_address:
|
||
logger.warning("映射共享内存失败")
|
||
kernel32.CloseHandle(file_handle)
|
||
return None, None
|
||
|
||
# 写入Key数据
|
||
key = "3101b223dca7715b0154924f0eeeee20".encode('utf-8')
|
||
kernel32.RtlMoveMemory(data_address, key, len(key))
|
||
|
||
logger.success("共享内存创建成功")
|
||
return file_handle, data_address
|
||
|
||
except Exception as e:
|
||
logger.warning(f"创建共享内存异常: {e}")
|
||
return None, None
|
||
|
||
|
||
class NoveLoader:
|
||
"""
|
||
Loader.dll 封装类
|
||
|
||
通过内存偏移调用 DLL 中的未导出函数
|
||
"""
|
||
|
||
# 加载器模块基址
|
||
loader_module_base: int = 0
|
||
|
||
# 函数偏移地址（基于 Loader.dll）
|
||
_InitWeChatSocket: int = 0xB080
|
||
_GetUserWeChatVersion: int = 0xCB80
|
||
_InjectWeChat: int = 0xCC10
|
||
_SendWeChatData: int = 0xAF90
|
||
_DestroyWeChat: int = 0xC540
|
||
_UseUtf8: int = 0xC680
|
||
_InjectWeChat2: int = 0x14D7
|
||
_InjectWeChatPid: int = 0xB750
|
||
_InjectWeChatMultiOpen: int = 0x33B2
|
||
|
||
def __init__(self, loader_path: str):
|
||
"""
|
||
初始化 Loader
|
||
|
||
Args:
|
||
loader_path: Loader.dll 的路径
|
||
"""
|
||
loader_path = os.path.realpath(loader_path)
|
||
if not os.path.exists(loader_path):
|
||
logger.error(f'Loader.dll 不存在: {loader_path}')
|
||
raise FileNotFoundError(f'Loader.dll 不存在: {loader_path}')
|
||
|
||
logger.info(f"加载 Loader.dll: {loader_path}")
|
||
loader_module = WinDLL(loader_path)
|
||
self.loader_module_base = loader_module._handle
|
||
|
||
# 使用 UTF-8 编码
|
||
self.UseUtf8()
|
||
logger.success("Loader.dll 加载成功")
|
||
|
||
# 初始化回调
|
||
from WechatHook.callbacks import (
|
||
wechat_connect_callback,
|
||
wechat_recv_callback,
|
||
wechat_close_callback
|
||
)
|
||
self.InitWeChatSocket(
|
||
wechat_connect_callback,
|
||
wechat_recv_callback,
|
||
wechat_close_callback
|
||
)
|
||
|
||
def __get_non_exported_func(self, offset: int, arg_types, return_type):
|
||
"""
|
||
通过内存偏移获取未导出的函数
|
||
|
||
Args:
|
||
offset: 函数相对于模块基址的偏移
|
||
arg_types: 参数类型列表
|
||
return_type: 返回值类型
|
||
|
||
Returns:
|
||
可调用的函数对象
|
||
"""
|
||
func_addr = self.loader_module_base + offset
|
||
if arg_types:
|
||
func_type = ctypes.WINFUNCTYPE(return_type, *arg_types)
|
||
else:
|
||
func_type = ctypes.WINFUNCTYPE(return_type)
|
||
return func_type(func_addr)
|
||
|
||
def InitWeChatSocket(
|
||
self,
|
||
connect_callback: Callable,
|
||
recv_callback: Callable,
|
||
close_callback: Callable
|
||
) -> bool:
|
||
"""
|
||
初始化微信 Socket 回调
|
||
|
||
Args:
|
||
connect_callback: 连接回调函数
|
||
recv_callback: 接收消息回调函数
|
||
close_callback: 断开连接回调函数
|
||
|
||
Returns:
|
||
是否初始化成功
|
||
"""
|
||
func = self.__get_non_exported_func(
|
||
self._InitWeChatSocket,
|
||
[ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p],
|
||
ctypes.c_bool
|
||
)
|
||
result = func(connect_callback, recv_callback, close_callback)
|
||
logger.info(f"InitWeChatSocket: {result}")
|
||
return result
|
||
|
||
def GetUserWeChatVersion(self) -> str:
|
||
"""
|
||
获取用户电脑上安装的微信版本
|
||
|
||
Returns:
|
||
微信版本号，如 "3.9.10.19"
|
||
"""
|
||
func = self.__get_non_exported_func(
|
||
self._GetUserWeChatVersion,
|
||
[ctypes.c_char_p],
|
||
ctypes.c_bool
|
||
)
|
||
out = create_string_buffer(30)
|
||
if func(out):
|
||
version = out.value.decode('utf-8')
|
||
logger.info(f"微信版本: {version}")
|
||
return version
|
||
else:
|
||
logger.warning("获取微信版本失败")
|
||
return ''
|
||
|
||
def InjectWeChat(self, dll_path: str) -> int:
|
||
"""
|
||
注入微信（智能多开）
|
||
|
||
Args:
|
||
dll_path: Helper.dll 的路径
|
||
|
||
Returns:
|
||
客户端 ID（进程 ID），失败返回 0
|
||
"""
|
||
dll_path = os.path.realpath(dll_path)
|
||
if not os.path.exists(dll_path):
|
||
logger.error(f'Helper.dll 不存在: {dll_path}')
|
||
return 0
|
||
|
||
func = self.__get_non_exported_func(
|
||
self._InjectWeChat,
|
||
[ctypes.c_char_p],
|
||
ctypes.c_uint32
|
||
)
|
||
client_id = func(c_string(dll_path))
|
||
if client_id:
|
||
logger.success(f"注入微信成功，客户端 ID: {client_id}")
|
||
else:
|
||
logger.error("注入微信失败")
|
||
return client_id
|
||
|
||
def SendWeChatData(self, client_id: int, message: str) -> bool:
|
||
"""
|
||
向微信发送数据
|
||
|
||
Args:
|
||
client_id: 客户端 ID
|
||
message: JSON 格式的消息
|
||
|
||
Returns:
|
||
是否发送成功
|
||
"""
|
||
func = self.__get_non_exported_func(
|
||
self._SendWeChatData,
|
||
[ctypes.c_uint32, ctypes.c_char_p],
|
||
ctypes.c_bool
|
||
)
|
||
result = func(client_id, c_string(message))
|
||
return result
|
||
|
||
def DestroyWeChat(self) -> bool:
|
||
"""
|
||
销毁微信连接
|
||
|
||
Returns:
|
||
是否成功
|
||
"""
|
||
func = self.__get_non_exported_func(
|
||
self._DestroyWeChat,
|
||
None,
|
||
ctypes.c_bool
|
||
)
|
||
result = func()
|
||
logger.info(f"DestroyWeChat: {result}")
|
||
return result
|
||
|
||
def UseUtf8(self) -> bool:
|
||
"""
|
||
设置使用 UTF-8 编码
|
||
|
||
Returns:
|
||
是否成功
|
||
"""
|
||
func = self.__get_non_exported_func(
|
||
self._UseUtf8,
|
||
None,
|
||
ctypes.c_bool
|
||
)
|
||
return func()
|
||
|
||
def InjectWeChat2(self, dll_path: str, exe_path: str) -> int:
|
||
"""
|
||
注入微信（指定微信路径）
|
||
|
||
Args:
|
||
dll_path: Helper.dll 的路径
|
||
exe_path: WeChat.exe 的路径
|
||
|
||
Returns:
|
||
客户端 ID，失败返回 0
|
||
"""
|
||
func = self.__get_non_exported_func(
|
||
self._InjectWeChat2,
|
||
[ctypes.c_char_p, ctypes.c_char_p],
|
||
ctypes.c_uint32
|
||
)
|
||
return func(c_string(dll_path), c_string(exe_path))
|
||
|
||
def InjectWeChatPid(self, pid: int, dll_path: str) -> int:
|
||
"""
|
||
注入指定的微信进程
|
||
|
||
Args:
|
||
pid: 微信进程 ID
|
||
dll_path: Helper.dll 的路径
|
||
|
||
Returns:
|
||
客户端 ID，失败返回 0
|
||
"""
|
||
func = self.__get_non_exported_func(
|
||
self._InjectWeChatPid,
|
||
[ctypes.c_uint32, ctypes.c_char_p],
|
||
ctypes.c_uint32
|
||
)
|
||
return func(pid, c_string(dll_path))
|
||
|
||
def InjectWeChatMultiOpen(self, dll_path: str, exe_path: str = "") -> int:
|
||
"""
|
||
多开一个新的微信进程并注入
|
||
|
||
Args:
|
||
dll_path: Helper.dll 的路径
|
||
exe_path: WeChat.exe 的路径（可选）
|
||
|
||
Returns:
|
||
客户端 ID，失败返回 0
|
||
"""
|
||
func = self.__get_non_exported_func(
|
||
self._InjectWeChatMultiOpen,
|
||
[ctypes.c_char_p, ctypes.c_char_p],
|
||
ctypes.c_uint32
|
||
)
|
||
return func(c_string(dll_path), c_string(exe_path) if exe_path else c_string(""))
|