如何在Python中直接执行内存中的.exe文件字节数据

Alright, let's tackle this problem head-on. You want to run a PE-format .exe (like your PyInstaller-generated file) directly from memory—no disk writes allowed, since it's stored in a MemoryFS object. This relies on Windows API calls to manually load and execute the PE binary in memory, which is exactly what we'll cover below.

Prerequisites #

First, install the required library to parse PE files:

pip install pefile

We'll use ctypes to call Windows kernel32 APIs directly, so no extra pywin32 installation is needed (though it works too).

Option 1: Execute PE in the Current Python Process #

This approach loads the PE binary into your existing Python process's memory space and runs it via a new thread. Note: This may conflict with your Python runtime if the PE has conflicting dependencies or architecture mismatches.

Code Implementation #

import ctypes
from ctypes import wintypes
import pefile

# Load kernel32.dll for Windows API calls
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)

# Define Windows API signatures
VirtualAlloc = kernel32.VirtualAlloc
VirtualAlloc.argtypes = (wintypes.LPVOID, wintypes.SIZE_T, wintypes.DWORD, wintypes.DWORD)
VirtualAlloc.restype = wintypes.LPVOID

MEM_COMMIT = 0x1000
MEM_RESERVE = 0x2000
PAGE_EXECUTE_READWRITE = 0x40

CreateThread = kernel32.CreateThread
CreateThread.argtypes = (wintypes.LPVOID, wintypes.SIZE_T, wintypes.LPVOID, wintypes.LPVOID, wintypes.DWORD, wintypes.LPDWORD)
CreateThread.restype = wintypes.HANDLE

WaitForSingleObject = kernel32.WaitForSingleObject
WaitForSingleObject.argtypes = (wintypes.HANDLE, wintypes.DWORD)
WaitForSingleObject.restype = wintypes.DWORD

INFINITE = 0xFFFFFFFF

def execute_pe_bytes(pe_data):
    # Parse the PE file structure
    pe = pefile.PE(data=pe_data)
    
    # Allocate executable memory for the PE image
    image_base = VirtualAlloc(
        None, 
        pe.OPTIONAL_HEADER.SizeOfImage, 
        MEM_COMMIT | MEM_RESERVE, 
        PAGE_EXECUTE_READWRITE
    )
    if not image_base:
        raise ctypes.WinError(ctypes.get_last_error())
    
    # Copy PE headers to allocated memory
    ctypes.memmove(image_base, pe_data, pe.DOS_HEADER.e_lfanew + pe.NT_HEADERS.SIZE_OF_HEADERS)
    
    # Copy each PE section to its target virtual address
    for section in pe.sections:
        dest_addr = ctypes.c_void_p(ctypes.c_uint64(image_base) + section.VirtualAddress)
        raw_data = pe_data[section.PointerToRawData : section.PointerToRawData + section.SizeOfRawData]
        ctypes.memmove(dest_addr, raw_data, section.SizeOfRawData)
    
    # Calculate the PE's entry point address
    entry_point = ctypes.c_void_p(ctypes.c_uint64(image_base) + pe.OPTIONAL_HEADER.AddressOfEntryPoint)
    
    # Create a thread to execute the entry point
    thread_handle = CreateThread(None, 0, entry_point, None, 0, None)
    if not thread_handle:
        raise ctypes.WinError(ctypes.get_last_error())
    
    # Wait for the thread to finish execution (optional)
    WaitForSingleObject(thread_handle, INFINITE)
    
    # Clean up handles (optional)
    kernel32.CloseHandle(thread_handle)
    # kernel32.VirtualFree(image_base, 0, 0x8000)  # Uncomment to free memory after execution

# Your existing MemoryFS code
from fs.memoryfs import MemoryFS
from fs.osfs import OSFS
from fs.copy import copy_file

external_path = 'some/external/location/that/should/not/run/binary/files/Installer.exe'
external_fs = OSFS('//external/server')
in_memory_fs = MemoryFS()
copy_file(external_fs, external_path, in_memory_fs, 'Installer.exe')
data = in_memory_fs.readbytes('Installer.exe')

# Execute the in-memory PE
execute_pe_bytes(data)

Option 2: Execute PE in a New Isolated Process #

This approach is more stable because it runs the PE in a separate process (we use cmd.exe as a lightweight "carrier" process, then replace its memory with your PE). This avoids conflicts with your Python runtime.

Code Implementation #

import ctypes
from ctypes import wintypes
import pefile

kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)

# Define Windows API structures
class STARTUPINFO(ctypes.Structure):
    _fields_ = [
        ("cb", wintypes.DWORD),
        ("lpReserved", wintypes.LPWSTR),
        ("lpDesktop", wintypes.LPWSTR),
        ("lpTitle", wintypes.LPWSTR),
        ("dwX", wintypes.DWORD),
        ("dwY", wintypes.DWORD),
        ("dwXSize", wintypes.DWORD),
        ("dwYSize", wintypes.DWORD),
        ("dwXCountChars", wintypes.DWORD),
        ("dwYCountChars", wintypes.DWORD),
        ("dwFillAttribute", wintypes.DWORD),
        ("dwFlags", wintypes.DWORD),
        ("wShowWindow", wintypes.WORD),
        ("cbReserved2", wintypes.WORD),
        ("lpReserved2", wintypes.LPBYTE),
        ("hStdInput", wintypes.HANDLE),
        ("hStdOutput", wintypes.HANDLE),
        ("hStdError", wintypes.HANDLE),
    ]

class PROCESS_INFORMATION(ctypes.Structure):
    _fields_ = [
        ("hProcess", wintypes.HANDLE),
        ("hThread", wintypes.HANDLE),
        ("dwProcessId", wintypes.DWORD),
        ("dwThreadId", wintypes.DWORD),
    ]

# Define API function signatures
CreateProcessW = kernel32.CreateProcessW
CreateProcessW.argtypes = [
    wintypes.LPCWSTR, wintypes.LPWSTR, wintypes.LPCVOID, wintypes.LPCVOID,
    wintypes.BOOL, wintypes.DWORD, wintypes.LPCVOID, wintypes.LPCWSTR,
    ctypes.POINTER(STARTUPINFO), ctypes.POINTER(PROCESS_INFORMATION)
]
CreateProcessW.restype = wintypes.BOOL

VirtualAllocEx = kernel32.VirtualAllocEx
VirtualAllocEx.argtypes = [
    wintypes.HANDLE, wintypes.LPVOID, wintypes.SIZE_T,
    wintypes.DWORD, wintypes.DWORD
]
VirtualAllocEx.restype = wintypes.LPVOID

WriteProcessMemory = kernel32.WriteProcessMemory
WriteProcessMemory.argtypes = [
    wintypes.HANDLE, wintypes.LPVOID, wintypes.LPCVOID,
    wintypes.SIZE_T, wintypes.LPVOID
]
WriteProcessMemory.restype = wintypes.BOOL

GetThreadContext = kernel32.GetThreadContext
SetThreadContext = kernel32.SetThreadContext
ResumeThread = kernel32.ResumeThread
WaitForSingleObject = kernel32.WaitForSingleObject

MEM_COMMIT = 0x1000
MEM_RESERVE = 0x2000
PAGE_EXECUTE_READWRITE = 0x40
CREATE_SUSPENDED = 0x00000004
CONTEXT_INTEGER = 0x00000002
INFINITE = 0xFFFFFFFF

def execute_pe_in_new_process(pe_data):
    pe = pefile.PE(data=pe_data)
    si = STARTUPINFO()
    si.cb = ctypes.sizeof(STARTUPINFO)
    pi = PROCESS_INFORMATION()
    
    # Get path to cmd.exe (matches your Python's architecture)
    system32_path = ctypes.windll.shell32.SHGetFolderPathW(None, 37, None, 0)
    cmd_path = ctypes.c_wchar_p(system32_path).value + r"\cmd.exe"
    
    # Create a suspended cmd.exe process as our carrier
    if not CreateProcessW(
        cmd_path, None, None, None, False, CREATE_SUSPENDED,
        None, None, ctypes.byref(si), ctypes.byref(pi)
    ):
        raise ctypes.WinError(ctypes.get_last_error())
    
    try:
        # Allocate memory in the carrier process for our PE
        remote_memory = VirtualAllocEx(
            pi.hProcess, None, pe.OPTIONAL_HEADER.SizeOfImage,
            MEM_COMMIT | MEM_RESERVE, PAGE_EXECUTE_READWRITE
        )
        if not remote_memory:
            raise ctypes.WinError(ctypes.get_last_error())
        
        # Write PE headers to remote memory
        written = wintypes.SIZE_T()
        if not WriteProcessMemory(
            pi.hProcess, remote_memory, pe_data,
            pe.DOS_HEADER.e_lfanew + pe.NT_HEADERS.SIZE_OF_HEADERS,
            ctypes.byref(written)
        ):
            raise ctypes.WinError(ctypes.get_last_error())
        
        # Write each PE section to its target address
        for section in pe.sections:
            dest_addr = ctypes.c_void_p(ctypes.c_uint64(remote_memory) + section.VirtualAddress)
            raw_data = pe_data[section.PointerToRawData : section.PointerToRawData + section.SizeOfRawData]
            if not WriteProcessMemory(pi.hProcess, dest_addr, raw_data, section.SizeOfRawData, ctypes.byref(written)):
                raise ctypes.WinError(ctypes.get_last_error())
        
        # Define x64 context structure (adjust for x86 if needed)
        class CONTEXT_X64(ctypes.Structure):
            _fields_ = [
                ("ContextFlags", wintypes.DWORD),
                ("Rip", wintypes.ULONGLONG),
                # Omit other fields for brevity; full structure needed for production
            ]
        
        ctx = CONTEXT_X64()
        ctx.ContextFlags = CONTEXT_INTEGER
        if not GetThreadContext(pi.hThread, ctypes.byref(ctx)):
            raise ctypes.WinError(ctypes.get_last_error())
        
        # Set the process's instruction pointer to the PE's entry point
        ctx.Rip = ctypes.c_uint64(remote_memory) + pe.OPTIONAL_HEADER.AddressOfEntryPoint
        if not SetThreadContext(pi.hThread, ctypes.byref(ctx)):
            raise ctypes.WinError(ctypes.get_last_error())
        
        # Resume the suspended thread to run our PE
        if ResumeThread(pi.hThread) == -1:
            raise ctypes.WinError(ctypes.get_last_error())
        
        # Wait for the process to finish (optional)
        WaitForSingleObject(pi.hProcess, INFINITE)
    finally:
        # Clean up process/thread handles
        kernel32.CloseHandle(pi.hThread)
        kernel32.CloseHandle(pi.hProcess)

# Your existing MemoryFS code
from fs.memoryfs import MemoryFS
from fs.osfs import OSFS
from fs.copy import copy_file

external_path = 'some/external/location/that/should/not/run/binary/files/Installer.exe'
external_fs = OSFS('//external/server')
in_memory_fs = MemoryFS()
copy_file(external_fs, external_path, in_memory_fs, 'Installer.exe')
data = in_memory_fs.readbytes('Installer.exe')

# Execute in a new process
execute_pe_in_new_process(data)

Critical Notes #

1. Architecture Matching: Your Python runtime (32-bit vs 64-bit) must match the PE binary's architecture. A 32-bit Python can't run a 64-bit exe, and vice versa.
2. PyInstaller Limitation: PyInstaller-generated exes typically unpack themselves to a temporary disk directory at runtime. To avoid this, you'd need to modify PyInstaller's bootloader to run entirely in memory—this is a more advanced task beyond basic PE loading.
3. Permissions: If your exe requires admin rights, your Python process must also run as administrator.
4. Antivirus Warnings: Memory-loaded PE execution is a technique used by malware, so your antivirus might flag this behavior. Test with exceptions or temporarily disable AV for testing.
5. Error Handling: The examples above include basic error handling, but you'll want to expand this for production use (e.g., handling invalid PE files, memory allocation failures).

内容的提问来源于stack exchange，提问作者Chris Collett