115!_游戏黑客：从内存修改到脚本注入的游戏安全实战指南

import ctypes
from ctypes import wintypes

# 定义常量
PROCESS_VM_READ = 0x0010
PROCESS_VM_WRITE = 0x0020
PROCESS_VM_OPERATION = 0x0008
PROCESS_QUERY_INFORMATION = 0x0400

# 加载kernel32.dll
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)

# 定义函数签名
kernel32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
kernel32.OpenProcess.restype = wintypes.HANDLE

kernel32.ReadProcessMemory.argtypes = [
    wintypes.HANDLE,  # hProcess
    wintypes.LPCVOID, # lpBaseAddress
    wintypes.LPVOID,  # lpBuffer
    ctypes.c_size_t,  # nSize
    ctypes.POINTER(ctypes.c_size_t)  # lpNumberOfBytesRead
]
kernel32.ReadProcessMemory.restype = wintypes.BOOL

kernel32.WriteProcessMemory.argtypes = [
    wintypes.HANDLE,  # hProcess
    wintypes.LPVOID,  # lpBaseAddress
    wintypes.LPCVOID, # lpBuffer
    ctypes.c_size_t,  # nSize
    ctypes.POINTER(ctypes.c_size_t)  # lpNumberOfBytesWritten
]
kernel32.WriteProcessMemory.restype = wintypes.BOOL

# 示例：读取进程内存
def read_process_memory(process_handle, address, size):
    buffer = ctypes.create_string_buffer(size)
    bytes_read = ctypes.c_size_t(0)
    
    if not kernel32.ReadProcessMemory(
        process_handle, 
        ctypes.c_void_p(address), 
        buffer, 
        size, 
        ctypes.byref(bytes_read)
    ):
        raise ctypes.WinError(ctypes.get_last_error())
    
    return buffer.raw[:bytes_read.value]

# 示例：写入进程内存
def write_process_memory(process_handle, address, data):
    size = len(data)
    buffer = ctypes.create_string_buffer(data)
    bytes_written = ctypes.c_size_t(0)
    
    if not kernel32.WriteProcessMemory(
        process_handle, 
        ctypes.c_void_p(address), 
        buffer, 
        size, 
        ctypes.byref(bytes_written)
    ):
        raise ctypes.WinError(ctypes.get_last_error())
    
    return bytes_written.value

# 使用示例
def example():
    # 获取进程ID（需要通过其他方式获取，如任务管理器）
    pid = 1234  # 示例PID
    
    # 打开进程
    process_handle = kernel32.OpenProcess(
        PROCESS_VM_READ | PROCESS_VM_WRITE | PROCESS_VM_OPERATION | PROCESS_QUERY_INFORMATION,
        False,
        pid
    )
    
    if not process_handle:
        raise ctypes.WinError(ctypes.get_last_error())
    
    try:
        # 读取内存示例
        address = 0x12345678  # 示例内存地址
        data = read_process_memory(process_handle, address, 4)  # 读取4字节
        print(f"读取的数据: {data.hex()}")
        
        # 写入内存示例
        new_data = b'\x01\x02\x03\x04'  # 要写入的数据
        bytes_written = write_process_memory(process_handle, address, new_data)
        print(f"写入的字节数: {bytes_written}")
    finally:
        # 关闭进程句柄
        kernel32.CloseHandle(process_handle)

import ctypes
import struct
from collections import defaultdict

def find_value_in_memory(process_handle, value, value_type='int'):
    """在进程内存中搜索指定的值"""
    # 将值转换为字节串
    if value_type == 'int':
        pattern = struct.pack('<i', value)  # 小端序
    elif value_type == 'float':
        pattern = struct.pack('<f', value)  # 小端序
    elif value_type == 'string':
        pattern = value.encode('utf-8')
    else:
        raise ValueError(f"不支持的值类型: {value_type}")
    
    pattern_length = len(pattern)
    found_addresses = []
    
    # 定义内存基本信息结构
    class MEMORY_BASIC_INFORMATION(ctypes.Structure):
        _fields_ = [
            ("BaseAddress", ctypes.c_void_p),
            ("AllocationBase", ctypes.c_void_p),
            ("AllocationProtect", ctypes.c_ulong),
            ("RegionSize", ctypes.c_size_t),
            ("State", ctypes.c_ulong),
            ("Protect", ctypes.c_ulong),
            ("Type", ctypes.c_ulong),
        ]
    
    # 定义VirtualQueryEx函数
    kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
    kernel32.VirtualQueryEx.argtypes = [
        ctypes.c_void_p,  # hProcess
        ctypes.c_void_p,  # lpAddress
        ctypes.POINTER(MEMORY_BASIC_INFORMATION),  # lpBuffer
        ctypes.c_size_t  # dwLength
    ]
    kernel32.VirtualQueryEx.restype = ctypes.c_size_t
    
    # 定义ReadProcessMemory函数
    kernel32.ReadProcessMemory.argtypes = [
        ctypes.c_void_p,  # hProcess
        ctypes.c_void_p,  # lpBaseAddress
        ctypes.c_void_p,  # lpBuffer
        ctypes.c_size_t,  # nSize
        ctypes.POINTER(ctypes.c_size_t)  # lpNumberOfBytesRead
    ]
    kernel32.ReadProcessMemory.restype = ctypes.c_int
    
    # 开始扫描内存
    address = 0x00000000
    while address < 0x80000000:  # 限制扫描范围，避免扫描系统内存
        # 查询内存区域信息
        mbi = MEMORY_BASIC_INFORMATION()
        if kernel32.VirtualQueryEx(process_handle, ctypes.c_void_p(address), ctypes.byref(mbi), ctypes.sizeof(mbi)) == 0:
            break
        
        # 检查内存区域是否可读
        if mbi.State == 0x1000 and mbi.Protect & 0x0002:
            # 分配缓冲区
            buffer = ctypes.create_string_buffer(mbi.RegionSize)
            bytes_read = ctypes.c_size_t(0)
            
            # 读取内存
            if kernel32.ReadProcessMemory(
                process_handle,
                mbi.BaseAddress,
                buffer,
                mbi.RegionSize,
                ctypes.byref(bytes_read)
            ):
                # 在缓冲区中搜索模式
                buffer_data = buffer.raw[:bytes_read.value]
                offset = 0
                while True:
                    # 查找模式
                    offset = buffer_data.find(pattern, offset)
                    if offset == -1:
                        break
                    
                    # 计算完整地址
                    found_address = ctypes.cast(mbi.BaseAddress, ctypes.c_ulonglong).value + offset
                    found_addresses.append(found_address)
                    
                    offset += 1
        
        # 移动到下一个内存区域
        address = ctypes.cast(mbi.BaseAddress, ctypes.c_ulonglong).value + mbi.RegionSize
    
    return found_addresses

# 使用示例
def search_example():
    # 这里需要先获取进程句柄，参考前面的示例
    # process_handle = ...
    
    # 搜索整数100
    # addresses = find_value_in_memory(process_handle, 100, 'int')
    # print(f"找到的地址: {addresses}")
    pass

import ctypes
import struct
import time
import psutil

def get_process_id(process_name):
    """通过进程名获取进程ID"""
    for process in psutil.process_iter(['pid', 'name']):
        if process.info['name'] == process_name:
            return process.info['pid']
    return None

def get_module_base_address(process_handle, module_name):
    """获取进程中指定模块的基地址"""
    # 定义MODULEENTRY32结构
    class MODULEENTRY32(ctypes.Structure):
        _fields_ = [
            ("dwSize", ctypes.c_ulong),
            ("th32ModuleID", ctypes.c_ulong),
            ("th32ProcessID", ctypes.c_ulong),
            ("GlblcntUsage", ctypes.c_ulong),
            ("ProccntUsage", ctypes.c_ulong),
            ("modBaseAddr", ctypes.c_void_p),
            ("modBaseSize", ctypes.c_ulong),
            ("hModule", ctypes.c_void_p),
            ("szModule", ctypes.c_char * 256),
            ("szExePath", ctypes.c_char * 260)
        ]
    
    # 加载Toolhelp32.dll
    kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
    kernel32.CreateToolhelp32Snapshot.argtypes = [ctypes.c_ulong, ctypes.c_ulong]
    kernel32.CreateToolhelp32Snapshot.restype = ctypes.c_void_p
    
    kernel32.Module32First.argtypes = [ctypes.c_void_p, ctypes.POINTER(MODULEENTRY32)]
    kernel32.Module32First.restype = ctypes.c_int
    
    kernel32.Module32Next.argtypes = [ctypes.c_void_p, ctypes.POINTER(MODULEENTRY32)]
    kernel32.Module32Next.restype = ctypes.c_int
    
    # 创建快照
    snapshot = kernel32.CreateToolhelp32Snapshot(0x00000008, ctypes.c_ulong(ctypes.windll.kernel32.GetProcessId(process_handle)))
    if snapshot == 0xFFFFFFFF:
        return None
    
    try:
        # 初始化MODULEENTRY32结构
        me32 = MODULEENTRY32()
        me32.dwSize = ctypes.sizeof(MODULEENTRY32)
        
        # 获取第一个模块
        if kernel32.Module32First(snapshot, ctypes.byref(me32)):
            # 遍历所有模块
            while True:
                # 比较模块名
                if me32.szModule.decode('utf-8').lower() == module_name.lower():
                    return ctypes.cast(me32.modBaseAddr, ctypes.c_ulonglong).value
                
                # 获取下一个模块
                if not kernel32.Module32Next(snapshot, ctypes.byref(me32)):
                    break
    finally:
        # 关闭快照
        kernel32.CloseHandle(snapshot)
    
    return None

def read_integer(process_handle, address):
    """从进程内存中读取整数"""
    buffer = ctypes.c_int()
    bytes_read = ctypes.c_size_t(0)
    
    kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
    result = kernel32.ReadProcessMemory(
        process_handle,
        ctypes.c_void_p(address),
        ctypes.byref(buffer),
        ctypes.sizeof(buffer),
        ctypes.byref(bytes_read)
    )
    
    if result and bytes_read.value == ctypes.sizeof(buffer):
        return buffer.value
    return None

def write_integer(process_handle, address, value):
    """向进程内存中写入整数"""
    data = ctypes.c_int(value)
    bytes_written = ctypes.c_size_t(0)
    
    kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
    result = kernel32.WriteProcessMemory(
        process_handle,
        ctypes.c_void_p(address),
        ctypes.byref(data),
        ctypes.sizeof(data),
        ctypes.byref(bytes_written)
    )
    
    return result and bytes_written.value == ctypes.sizeof(data)

def simple_game_hack(process_name, module_name):
    """简单的游戏修改器"""
    # 获取进程ID
    pid = get_process_id(process_name)
    if not pid:
        print(f"未找到进程: {process_name}")
        return
    
    print(f"找到进程 {process_name}, PID: {pid}")
    
    # 打开进程
    PROCESS_ALL_ACCESS = 0x1F0FFF
    kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
    process_handle = kernel32.OpenProcess(PROCESS_ALL_ACCESS, False, pid)
    
    if not process_handle:
        print(f"无法打开进程, 错误代码: {ctypes.get_last_error()}")
        return
    
    try:
        # 获取模块基地址
        base_address = get_module_base_address(process_handle, module_name)
        if not base_address:
            print(f"未找到模块: {module_name}")
            return
        
        print(f"模块 {module_name} 的基地址: 0x{base_address:X}")
        
        # 假设我们已经知道了分数和生命值相对于基地址的偏移量
        # 注意：在实际情况下，需要通过逆向工程或内存分析来确定这些偏移量
        score_offset = 0x00456780  # 示例偏移量
        health_offset = 0x00456784  # 示例偏移量
        
        # 计算实际地址
        score_address = base_address + score_offset
        health_address = base_address + health_offset
        
        print(f"分数地址: 0x{score_address:X}")
        print(f"生命值地址: 0x{health_address:X}")
        
        # 读取当前值
        current_score = read_integer(process_handle, score_address)
        current_health = read_integer(process_handle, health_address)
        
        print(f"当前分数: {current_score}")
        print(f"当前生命值: {current_health}")
        
        # 修改值
        new_score = 9999
        new_health = 999
        
        if write_integer(process_handle, score_address, new_score):
            print(f"分数已修改为: {new_score}")
        else:
            print("修改分数失败")
        
        if write_integer(process_handle, health_address, new_health):
            print(f"生命值已修改为: {new_health}")
        else:
            print("修改生命值失败")
        
        # 验证修改
        updated_score = read_integer(process_handle, score_address)
        updated_health = read_integer(process_handle, health_address)
        
        print(f"修改后分数: {updated_score}")
        print(f"修改后生命值: {updated_health}")
        
    finally:
        # 关闭进程句柄
        kernel32.CloseHandle(process_handle)

# 使用示例
# simple_game_hack("game.exe", "game.dll")

def follow_pointer_chain(process_handle, base_address, offsets):
    """跟随指针链，获取最终地址"""
    current_address = base_address
    
    kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
    
    # 遍历所有偏移量，除了最后一个
    for i in range(len(offsets) - 1):
        # 读取当前地址指向的值（指针）
        pointer_value = ctypes.c_ulonglong()
        bytes_read = ctypes.c_size_t(0)
        
        result = kernel32.ReadProcessMemory(
            process_handle,
            ctypes.c_void_p(current_address + offsets[i]),
            ctypes.byref(pointer_value),
            ctypes.sizeof(pointer_value),
            ctypes.byref(bytes_read)
        )
        
        if not result or bytes_read.value != ctypes.sizeof(pointer_value):
            print(f"读取指针失败，偏移量: 0x{offsets[i]:X}")
            return None
        
        # 更新当前地址
        current_address = pointer_value.value
    
    # 返回最终地址（基址 + 所有偏移量）
    return current_address + offsets[-1]

def read_value_with_pointers(process_handle, base_address, offsets, value_type='int'):
    """使用指针链读取值"""
    final_address = follow_pointer_chain(process_handle, base_address, offsets)
    if final_address is None:
        return None
    
    # 读取最终值
    if value_type == 'int':
        return read_integer(process_handle, final_address)
    elif value_type == 'float':
        # 需要实现读取浮点数的函数
        buffer = ctypes.c_float()
        bytes_read = ctypes.c_size_t(0)
        
        kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
        result = kernel32.ReadProcessMemory(
            process_handle,
            ctypes.c_void_p(final_address),
            ctypes.byref(buffer),
            ctypes.sizeof(buffer),
            ctypes.byref(bytes_read)
        )
        
        if result and bytes_read.value == ctypes.sizeof(buffer):
            return buffer.value
        return None
    # 可以扩展支持其他类型
    
    return None

def write_value_with_pointers(process_handle, base_address, offsets, value, value_type='int'):
    """使用指针链写入值"""
    final_address = follow_pointer_chain(process_handle, base_address, offsets)
    if final_address is None:
        return False
    
    # 写入最终值
    if value_type == 'int':
        return write_integer(process_handle, final_address, value)
    elif value_type == 'float':
        # 需要实现写入浮点数的函数
        data = ctypes.c_float(value)
        bytes_written = ctypes.c_size_t(0)
        
        kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
        result = kernel32.WriteProcessMemory(
            process_handle,
            ctypes.c_void_p(final_address),
            ctypes.byref(data),
            ctypes.sizeof(data),
            ctypes.byref(bytes_written)
        )
        
        return result and bytes_written.value == ctypes.sizeof(data)
    # 可以扩展支持其他类型
    
    return False

# 使用示例
def pointer_example():
    # 这里需要先获取进程句柄和基址，参考前面的示例
    # process_handle = ...
    # base_address = ...
    
    # 示例指针链：[0x1234, 0x56, 0x78]
    # offsets = [0x1234, 0x56, 0x78]
    
    # 读取值
    # value = read_value_with_pointers(process_handle, base_address, offsets, 'int')
    # print(f"通过指针链读取的值: {value}")
    
    # 修改值
    # success = write_value_with_pointers(process_handle, base_address, offsets, 9999, 'int')
    # print(f"修改结果: {success}")
    pass

def inject_dll(process_handle, dll_path):
    """向目标进程注入DLL"""
    kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
    
    # 1. 在目标进程中分配内存，用于存储DLL路径
    dll_path_bytes = dll_path.encode('utf-8')
    dll_path_length = len(dll_path_bytes) + 1  # +1 用于null终止符
    
    allocated_memory = kernel32.VirtualAllocEx(
        process_handle,
        None,
        dll_path_length,
        0x1000 | 0x2000,  # MEM_COMMIT | MEM_RESERVE
        0x04  # PAGE_READWRITE
    )
    
    if not allocated_memory:
        print(f"分配内存失败，错误代码: {ctypes.get_last_error()}")
        return False
    
    # 2. 将DLL路径写入分配的内存
    bytes_written = ctypes.c_size_t(0)
    result = kernel32.WriteProcessMemory(
        process_handle,
        allocated_memory,
        dll_path_bytes,
        dll_path_length,
        ctypes.byref(bytes_written)
    )
    
    if not result or bytes_written.value != dll_path_length:
        print(f"写入DLL路径失败，错误代码: {ctypes.get_last_error()}")
        kernel32.VirtualFreeEx(process_handle, allocated_memory, 0, 0x8000)  # MEM_RELEASE
        return False
    
    # 3. 获取LoadLibraryA函数的地址
    load_library_addr = kernel32.GetProcAddress(
        kernel32.GetModuleHandleA(b'kernel32.dll'),
        b'LoadLibraryA'
    )
    
    if not load_library_addr:
        print(f"获取LoadLibraryA地址失败，错误代码: {ctypes.get_last_error()}")
        kernel32.VirtualFreeEx(process_handle, allocated_memory, 0, 0x8000)
        return False
    
    # 4. 在目标进程中创建远程线程，执行LoadLibraryA加载DLL
    thread_id = ctypes.c_ulong()
    thread_handle = kernel32.CreateRemoteThread(
        process_handle,
        None,
        0,
        load_library_addr,
        allocated_memory,
        0,
        ctypes.byref(thread_id)
    )
    
    if not thread_handle:
        print(f"创建远程线程失败，错误代码: {ctypes.get_last_error()}")
        kernel32.VirtualFreeEx(process_handle, allocated_memory, 0, 0x8000)
        return False
    
    # 5. 等待线程执行完成
    kernel32.WaitForSingleObject(thread_handle, 0xFFFFFFFF)  # INFINITE
    
    # 6. 获取线程退出代码（即LoadLibraryA的返回值，DLL的模块句柄）
    exit_code = ctypes.c_ulong()
    kernel32.GetExitCodeThread(thread_handle, ctypes.byref(exit_code))
    
    # 7. 清理资源
    kernel32.CloseHandle(thread_handle)
    kernel32.VirtualFreeEx(process_handle, allocated_memory, 0, 0x8000)
    
    if exit_code.value == 0:
        print("DLL加载失败")
        return False
    
    print(f"DLL注入成功，DLL模块句柄: 0x{exit_code.value:X}")
    return True

# 使用示例
def dll_injection_example():
    # 这里需要先获取进程句柄，参考前面的示例
    # process_handle = ...
    
    # dll_path = "C:\path\to\your\hack.dll"
    # success = inject_dll(process_handle, dll_path)
    # print(f"DLL注入结果: {success}")
    pass

def inject_shellcode(process_handle, shellcode):
    """向目标进程注入shellcode"""
    kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
    
    # 1. 在目标进程中分配可执行内存
    shellcode_length = len(shellcode)
    
    allocated_memory = kernel32.VirtualAllocEx(
        process_handle,
        None,
        shellcode_length,
        0x1000 | 0x2000,  # MEM_COMMIT | MEM_RESERVE
        0x40  # PAGE_EXECUTE_READWRITE
    )
    
    if not allocated_memory:
        print(f"分配内存失败，错误代码: {ctypes.get_last_error()}")
        return False
    
    # 2. 将shellcode写入分配的内存
    bytes_written = ctypes.c_size_t(0)
    result = kernel32.WriteProcessMemory(
        process_handle,
        allocated_memory,
        shellcode,
        shellcode_length,
        ctypes.byref(bytes_written)
    )
    
    if not result or bytes_written.value != shellcode_length:
        print(f"写入shellcode失败，错误代码: {ctypes.get_last_error()}")
        kernel32.VirtualFreeEx(process_handle, allocated_memory, 0, 0x8000)  # MEM_RELEASE
        return False
    
    # 3. 在目标进程中创建远程线程，执行shellcode
    thread_id = ctypes.c_ulong()
    thread_handle = kernel32.CreateRemoteThread(
        process_handle,
        None,
        0,
        allocated_memory,
        None,
        0,
        ctypes.byref(thread_id)
    )
    
    if not thread_handle:
        print(f"创建远程线程失败，错误代码: {ctypes.get_last_error()}")
        kernel32.VirtualFreeEx(process_handle, allocated_memory, 0, 0x8000)
        return False
    
    print(f"Shellcode注入成功，线程ID: {thread_id.value}")
    print(f"Shellcode地址: 0x{allocated_memory:X}")
    
    # 4. 可选：等待线程执行完成
    # kernel32.WaitForSingleObject(thread_handle, 0xFFFFFFFF)  # INFINITE
    
    # 5. 可选：清理资源
    # kernel32.CloseHandle(thread_handle)
    # kernel32.VirtualFreeEx(process_handle, allocated_memory, 0, 0x8000)
    
    return True

# 使用示例
def shellcode_injection_example():
    # 这里需要先获取进程句柄，参考前面的示例
    # process_handle = ...
    
    # 注意：以下是一个简单的示例shellcode，在实际使用前需要根据目标系统定制
    # 这个shellcode只是一个示例，不会做任何实际操作
    # shellcode = b'\x90\x90\x90\x90\xc3'  # NOP指令序列 + RET
    
    # success = inject_shellcode(process_handle, shellcode)
    # print(f"Shellcode注入结果: {success}")
    pass

// game_hack.cpp
#include <windows.h>
#include <stdio.h>

// DLL入口点
BOOL APIENTRY DllMain(HMODULE hModule, DWORD ul_reason_for_call, LPVOID lpReserved) {
    switch (ul_reason_for_call) {
    case DLL_PROCESS_ATTACH:
        // 创建一个新线程来执行我们的代码，避免阻塞DLL加载过程
        CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)HackThread, NULL, 0, NULL);
        break;
    case DLL_THREAD_ATTACH:
    case DLL_THREAD_DETACH:
    case DLL_PROCESS_DETACH:
        break;
    }
    return TRUE;
}

// 我们的hack线程
DWORD WINAPI HackThread(LPVOID lpParam) {
    // 等待游戏窗口出现
    Sleep(1000);
    
    // 查找游戏窗口
    HWND hGameWindow = FindWindow(NULL, L"Game Window Title");
    if (hGameWindow) {
        // 显示一个消息框，表示注入成功
        MessageBox(NULL, L"DLL已成功注入到游戏进程！", L"游戏黑客", MB_OK);
        
        // 在这里添加你的游戏修改代码
        // 例如：修改内存、钩取函数等
        
        // 示例：无限生命
        // DWORD healthAddress = 0x00456780; // 示例地址
        // *(int*)healthAddress = 999; // 设置生命值为999
    }
    
    return 0;
}

import socket
import threading
import time
import struct

def hexdump(src, length=16):
    """打印十六进制转储，用于调试"""
    result = []
    digits = 2
    
    for i in range(0, len(src), length):
        s = src[i:i+length]
        hexa = ' '.join([f"{x:0{digits}x}" for x in s])
        text = ''.join([chr(x) if 0x20 <= x < 0x7f else '.' for x in s])
        result.append(f"{i:04x}: {hexa:<{length*(digits+1)-1}} |{text}|")
    
    return '\n'.join(result)

class GameProxy:
    def __init__(self, local_host, local_port, remote_host, remote_port):
        self.local_host = local_host
        self.local_port = local_port
        self.remote_host = remote_host
        self.remote_port = remote_port
        self.server_socket = None
        self.threads = []
        self.running = False
        self.client_data_callback = None
        self.server_data_callback = None
    
    def start(self):
        """启动代理服务器"""
        try:
            # 创建服务器套接字
            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            self.server_socket.bind((self.local_host, self.local_port))
            self.server_socket.listen(5)
            
            self.running = True
            print(f"游戏代理启动: {self.local_host}:{self.local_port} -> {self.remote_host}:{self.remote_port}")
            
            # 接受客户端连接
            while self.running:
                try:
                    client_socket, addr = self.server_socket.accept()
                    print(f"客户端连接: {addr[0]}:{addr[1]}")
                    
                    # 创建处理线程
                    thread = threading.Thread(target=self.handle_client, args=(client_socket,))
                    thread.daemon = True
                    thread.start()
                    self.threads.append(thread)
                except socket.error as e:
                    if not self.running:  # 如果是因为停止代理而中断，则不报错
                        break
                    print(f"接受连接时出错: {e}")
        except Exception as e:
            print(f"启动代理时出错: {e}")
            self.stop()
    
    def stop(self):
        """停止代理服务器"""
        self.running = False
        
        if self.server_socket:
            try:
                self.server_socket.close()
            except:
                pass
        
        # 等待所有线程结束
        for thread in self.threads:
            if thread.is_alive():
                thread.join(1)
        
        print("游戏代理已停止")
    
    def handle_client(self, client_socket):
        """处理客户端连接"""
        remote_socket = None
        
        try:
            # 连接到远程服务器
            remote_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            remote_socket.connect((self.remote_host, self.remote_port))
            print(f"已连接到远程服务器: {self.remote_host}:{self.remote_port}")
            
            # 创建两个线程分别处理双向通信
            client_thread = threading.Thread(
                target=self.forward_data,
                args=(client_socket, remote_socket, "客户端", self.client_data_callback)
            )
            server_thread = threading.Thread(
                target=self.forward_data,
                args=(remote_socket, client_socket, "服务器", self.server_data_callback)
            )
            
            client_thread.daemon = True
            server_thread.daemon = True
            
            client_thread.start()
            server_thread.start()
            
            # 等待任一线程结束
            client_thread.join()
            server_thread.join()
        except Exception as e:
            print(f"处理连接时出错: {e}")
        finally:
            # 关闭套接字
            try:
                client_socket.close()
            except:
                pass
            
            try:
                if remote_socket:
                    remote_socket.close()
            except:
                pass
            
            print("连接已关闭")
    
    def forward_data(self, source, destination, direction, callback):
        """转发数据，并在转发前调用回调函数"""
        while self.running:
            try:
                # 从源读取数据
                data = source.recv(4096)
                if not data:
                    break
                
                print(f"\n[{time.strftime('%H:%M:%S')}] 从{direction}接收 {len(data)} 字节:")
                print(hexdump(data))
                
                # 调用回调函数处理数据
                if callback:
                    data = callback(data)
                
                # 发送数据到目标
                destination.sendall(data)
                print(f"[{time.strftime('%H:%M:%S')}] 发送到{direction} {len(data)} 字节")
            except Exception as e:
                print(f"转发数据时出错: {e}")
                break
    
    def set_client_data_callback(self, callback):
        """设置客户端数据处理回调函数"""
        self.client_data_callback = callback
    
    def set_server_data_callback(self, callback):
        """设置服务器数据处理回调函数"""
        self.server_data_callback = callback

# 使用示例
def game_proxy_example():
    # 代理配置
    local_host = '127.0.0.1'
    local_port = 12345
    remote_host = 'game-server.example.com'
    remote_port = 12345
    
    # 创建代理实例
    proxy = GameProxy(local_host, local_port, remote_host, remote_port)
    
    # 定义数据处理回调函数
    def process_client_data(data):
        """处理从客户端到服务器的数据"""
        # 这里可以修改数据
        # 例如：修改游戏分数
        # data = modify_score(data)
        return data
    
    def process_server_data(data):
        """处理从服务器到客户端的数据"""
        # 这里可以修改数据或提取信息
        # 例如：提取其他玩家的位置信息
        # extract_player_positions(data)
        return data
    
    # 设置回调函数
    proxy.set_client_data_callback(process_client_data)
    proxy.set_server_data_callback(process_server_data)
    
    try:
        # 启动代理
        proxy.start()
    except KeyboardInterrupt:
        # 按下Ctrl+C时停止代理
        print("\n正在停止代理...")
        proxy.stop()

# 运行示例
# game_proxy_example()

import socket
import struct
import time
import random

class GameClient:
    def __init__(self, server_host, server_port):
        self.server_host = server_host
        self.server_port = server_port
        self.socket = None
        self.connected = False
    
    def connect(self):
        """连接到游戏服务器"""
        try:
            self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.socket.connect((self.server_host, self.server_port))
            self.connected = True
            print(f"已连接到游戏服务器: {self.server_host}:{self.server_port}")
            return True
        except Exception as e:
            print(f"连接服务器失败: {e}")
            return False
    
    def disconnect(self):
        """断开与游戏服务器的连接"""
        if self.socket:
            try:
                self.socket.close()
            except:
                pass
        self.connected = False
        print("已断开与服务器的连接")
    
    def send_packet(self, packet):
        """发送数据包到服务器"""
        if not self.connected:
            print("未连接到服务器")
            return False
        
        try:
            # 假设数据包有一个4字节的长度前缀
            packet_length = len(packet)
            header = struct.pack('<I', packet_length)
            self.socket.sendall(header + packet)
            print(f"发送数据包: 长度={packet_length} 数据={packet.hex()}")
            return True
        except Exception as e:
            print(f"发送数据包失败: {e}")
            self.disconnect()
            return False
    
    def receive_packet(self):
        """从服务器接收数据包"""
        if not self.connected:
            print("未连接到服务器")
            return None
        
        try:
            # 接收长度前缀
            header = self._receive_exactly(4)
            if not header:
                return None
            
            packet_length = struct.unpack('<I', header)[0]
            
            # 接收数据包内容
            packet = self._receive_exactly(packet_length)
            if not packet:
                return None
            
            print(f"接收数据包: 长度={packet_length} 数据={packet.hex()}")
            return packet
        except Exception as e:
            print(f"接收数据包失败: {e}")
            self.disconnect()
            return None
    
    def _receive_exactly(self, length):
        """精确接收指定长度的数据"""
        data = b''
        while len(data) < length:
            chunk = self.socket.recv(length - len(data))
            if not chunk:
                return None
            data += chunk
        return data
    
    def login(self, username, password):
        """发送登录请求"""
        # 这里需要根据游戏协议实现登录数据包的构造
        # 以下是一个示例
        username_bytes = username.encode('utf-8')
        password_bytes = password.encode('utf-8')
        
        # 假设登录数据包格式：类型(1字节) + 用户名长度(1字节) + 用户名 + 密码长度(1字节) + 密码
        packet_type = 0x01
        username_len = len(username_bytes)
        password_len = len(password_bytes)
        
        packet = struct.pack('BB', packet_type, username_len) + username_bytes
        packet += struct.pack('B', password_len) + password_bytes
        
        return self.send_packet(packet)
    
    def send_movement(self, x, y):
        """发送移动请求"""
        # 这里需要根据游戏协议实现移动数据包的构造
        # 以下是一个示例
        packet_type = 0x02  # 假设移动请求类型为0x02
        
        # 假设坐标使用浮点数表示
        packet = struct.pack('Bff', packet_type, x, y)
        
        return self.send_packet(packet)

# 使用示例
def game_client_example():
    client = GameClient('game-server.example.com', 12345)
    
    if client.connect():
        try:
            # 发送登录请求
            client.login('player1', 'password123')
            
            # 接收登录响应
            response = client.receive_packet()
            if response:
                # 解析响应...
                print("登录成功")
            
            # 模拟游戏操作
            while True:
                # 发送随机移动
                x = random.uniform(-100, 100)
                y = random.uniform(-100, 100)
                client.send_movement(x, y)
                
                # 接收服务器响应
                response = client.receive_packet()
                
                # 等待一段时间
                time.sleep(1)
                
        except KeyboardInterrupt:
            print("\n退出客户端")
        finally:
            client.disconnect()

# 运行示例
# game_client_example()

# IDA Python脚本示例
import idaapi
import idautils
import idc

def find_game_functions():
    """查找可能的游戏关键函数"""
    # 搜索与游戏相关的字符串
    game_strings = []
    for s in idautils.Strings():
        if "score" in str(s).lower() or "health" in str(s).lower() or "player" in str(s).lower():
            game_strings.append(str(s))
    
    print(f"找到 {len(game_strings)} 个游戏相关字符串")
    for s in game_strings:
        print(f"- {s}")
    
    # 查找引用这些字符串的函数
    game_functions = set()
    for s in idautils.Strings():
        if "score" in str(s).lower() or "health" in str(s).lower() or "player" in str(s).lower():
            for xref in XrefsTo(s.ea, 0):
                func = idaapi.get_func(xref.frm)
                if func:
                    game_functions.add(func.start_ea)
    
    print(f"\n找到 {len(game_functions)} 个可能的游戏关键函数")
    for func_ea in game_functions:
        func_name = idc.get_func_name(func_ea)
        print(f"- 0x{func_ea:X}: {func_name}")
    
    # 分析函数的交叉引用
    for func_ea in game_functions:
        func_name = idc.get_func_name(func_ea)
        print(f"\n函数 {func_name} (0x{func_ea:X}) 的调用者:")
        for xref in XrefsTo(func_ea, 0):
            caller_name = idc.get_func_name(xref.frm)
            print(f"- 0x{xref.frm:X}: {caller_name}")

# 运行脚本
# find_game_functions()

# 需要安装frida: pip install frida frida-tools
import frida
import sys

def on_message(message, data):
    """处理从frida脚本收到的消息"""
    if message['type'] == 'send':
        print(f"[*] {message['payload']}")
    elif message['type'] == 'error':
        print(f"[!] {message['stack']}")

def hook_unity_game(process_name):
    """使用frida钩取Unity游戏的关键函数"""
    # frida JavaScript代码
    js_code = """
    // 查找Unity游戏的Player类并钩取其方法
    Java.perform(function() {
        // 假设游戏使用com.example.game包
        var Player = Java.use("com.example.game.Player");
        
        // 钩取setScore方法
        Player.setScore.implementation = function(score) {
            console.log("[+] 调用setScore: " + score);
            // 可以修改参数
            this.setScore(9999);
        };
        
        // 钩取getHealth方法
        Player.getHealth.implementation = function() {
            console.log("[+] 调用getHealth");
            // 可以修改返回值
            return 999;
        };
    });
    """
    
    try:
        # 附加到进程
        session = frida.attach(process_name)
        script = session.create_script(js_code)
        script.on('message', on_message)
        script.load()
        print(f"已附加到进程 {process_name}")
        
        # 保持程序运行
        sys.stdin.read()
        
        # 分离
        session.detach()
    except Exception as e:
        print(f"出错: {e}")

# 使用示例
# hook_unity_game("com.example.game")

def analyze_memory_dump(dump_file_path):
    """分析内存转储文件"""
    print(f"分析内存转储文件: {dump_file_path}")
    
    # 读取内存转储文件
    try:
        with open(dump_file_path, 'rb') as f:
            dump_data = f.read()
        
        print(f"内存转储大小: {len(dump_data)} 字节")
        
        # 1. 搜索可能的flag
        print("\n=== 搜索可能的flag ===")
        # 常见的flag格式
        flag_patterns = [
            b'flag{', b'FLAG{', b'ctf{', b'CTF{',
            b'key{', b'KEY{', b'secret{', b'SECRET{'
        ]
        
        for pattern in flag_patterns:
            pos = 0
            while True:
                pos = dump_data.find(pattern, pos)
                if pos == -1:
                    break
                
                # 尝试提取完整的flag
                # 假设flag以}结尾
                end_pos = dump_data.find(b'}', pos)
                if end_pos != -1:
                    # 提取flag，并尝试解码为字符串
                    try:
                        flag = dump_data[pos:end_pos+1].decode('utf-8')
                        print(f"找到可能的flag: {flag} (偏移量: 0x{pos:X})")
                    except:
                        pass
                
                pos += len(pattern)
        
        # 2. 搜索可能的密码或凭证
        print("\n=== 搜索可能的凭证 ===")
        credential_patterns = [
            b'password=', b'password ', b'pwd=', b'pwd ',
            b'user=', b'username=', b'login=', b'session=',
            b'token=', b'key=', b'secret='
        ]
        
        for pattern in credential_patterns:
            pos = 0
            while True:
                pos = dump_data.find(pattern, pos)
                if pos == -1:
                    break
                
                # 提取周围的上下文
                context_start = max(0, pos - 20)
                context_end = min(len(dump_data), pos + 100)
                context = dump_data[context_start:context_end]
                
                try:
                    # 尝试解码为字符串，过滤掉不可打印字符
                    filtered_context = ''.join([chr(c) if 32 <= c < 127 else '.' for c in context])
                    print(f"找到可能的凭证: {filtered_context} (偏移量: 0x{pos:X})")
                except:
                    pass
                
                pos += len(pattern)
        
        # 3. 搜索可能的游戏数据
        print("\n=== 搜索可能的游戏数据 ===")
        game_data_patterns = [
            b'score', b'health', b'lives', b'ammo',
            b'gold', b'coins', b'exp', b'level'
        ]
        
        for pattern in game_data_patterns:
            pos = 0
            while True:
                pos = dump_data.find(pattern, pos)
                if pos == -1:
                    break
                
                # 提取周围的上下文
                context_start = max(0, pos - 20)
                context_end = min(len(dump_data), pos + 100)
                context = dump_data[context_start:context_end]
                
                try:
                    filtered_context = ''.join([chr(c) if 32 <= c < 127 else '.' for c in context])
                    print(f"找到可能的游戏数据: {filtered_context} (偏移量: 0x{pos:X})")
                except:
                    pass
                
                pos += len(pattern)
        
        # 4. 分析数字数据
        print("\n=== 分析数字数据 ===")
        # 搜索大的整数，可能是分数、金币等
        for i in range(0, len(dump_data) - 4):
            # 尝试解析为32位整数
            try:
                value = int.from_bytes(dump_data[i:i+4], byteorder='little')
                # 寻找大的正数，可能是游戏中的分数等
                if 100000 < value < 1000000000:
                    print(f"找到大整数: {value} (偏移量: 0x{i:X})")
            except:
                pass
                
    except Exception as e:
        print(f"分析内存转储时出错: {e}")

# 使用示例
# analyze_memory_dump("game_memory_dump.bin")

import pyautogui
import time
import keyboard
import threading

def auto_clicker(interval=0.1, hotkey='ctrl+alt+q'):
    """自动点击器"""
    running = False
    
    def toggle_running():
        nonlocal running
        running = not running
        status = "开始" if running else "停止"
        print(f"自动点击器已{status}")
    
    # 注册热键
    keyboard.add_hotkey(hotkey, toggle_running)
    print(f"自动点击器已启动，按 {hotkey} 开始/停止")
    
    try:
        while True:
            if running:
                pyautogui.click()
            time.sleep(interval)
    except KeyboardInterrupt:
        print("自动点击器已退出")

def auto_farm(resource_image, click_delay=1.0, scan_interval=2.0, hotkey='ctrl+alt+f'):
    """自动收集资源"""
    running = False
    
    def toggle_running():
        nonlocal running
        running = not running
        status = "开始" if running else "停止"
        print(f"自动收集资源已{status}")
    
    # 注册热键
    keyboard.add_hotkey(hotkey, toggle_running)
    print(f"自动收集资源已启动，按 {hotkey} 开始/停止")
    print(f"正在寻找资源图片: {resource_image}")
    
    try:
        while True:
            if running:
                try:
                    # 寻找资源图片
                    position = pyautogui.locateCenterOnScreen(resource_image, confidence=0.8)
                    if position:
                        # 移动到资源位置并点击
                        pyautogui.moveTo(position)
                        pyautogui.click()
                        print(f"找到并点击资源: {position}")
                        time.sleep(click_delay)
                except Exception as e:
                    print(f"查找资源时出错: {e}")
            time.sleep(scan_interval)
    except KeyboardInterrupt:
        print("自动收集资源已退出")

def game_bot_main():
    """游戏机器人主函数"""
    print("===== 游戏自动化工具 =====")
    print("1. 自动点击器")
    print("2. 自动收集资源")
    
    choice = input("请选择功能 (1/2): ")
    
    if choice == '1':
        try:
            interval = float(input("请输入点击间隔 (秒): "))
            clicker_thread = threading.Thread(target=auto_clicker, args=(interval,))
            clicker_thread.daemon = True
            clicker_thread.start()
            print("自动点击器线程已启动")
            
            # 保持主线程运行
            keyboard.wait('esc')
        except ValueError:
            print("无效的输入")
    elif choice == '2':
        resource_image = input("请输入资源图片路径: ")
        try:
            click_delay = float(input("请输入点击延迟 (秒): "))
            scan_interval = float(input("请输入扫描间隔 (秒): "))
            
            farm_thread = threading.Thread(target=auto_farm, args=(resource_image, click_delay, scan_interval))
            farm_thread.daemon = True
            farm_thread.start()
            print("自动收集资源线程已启动")
            
            # 保持主线程运行
            keyboard.wait('esc')
        except ValueError:
            print("无效的输入")
    else:
        print("无效的选择")

# 使用示例
# 需要安装相关库: pip install pyautogui keyboard
# game_bot_main()