summaryrefslogtreecommitdiff
path: root/foreign/client_handling/lazagne/config/lib/memorpy/WinProcess.py
diff options
context:
space:
mode:
Diffstat (limited to 'foreign/client_handling/lazagne/config/lib/memorpy/WinProcess.py')
-rw-r--r--foreign/client_handling/lazagne/config/lib/memorpy/WinProcess.py312
1 files changed, 312 insertions, 0 deletions
diff --git a/foreign/client_handling/lazagne/config/lib/memorpy/WinProcess.py b/foreign/client_handling/lazagne/config/lib/memorpy/WinProcess.py
new file mode 100644
index 0000000..18c7054
--- /dev/null
+++ b/foreign/client_handling/lazagne/config/lib/memorpy/WinProcess.py
@@ -0,0 +1,312 @@
+# Author: Nicolas VERDIER
+# This file is part of memorpy.
+#
+# memorpy is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# memorpy is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with memorpy. If not, see <http://www.gnu.org/licenses/>.
+
+from ctypes import pointer, sizeof, windll, create_string_buffer, c_ulong, byref, GetLastError, c_bool, WinError
+from .structures import *
+import copy
+import struct
+# import utils
+import platform
+from .BaseProcess import BaseProcess, ProcessException
+
+psapi = windll.psapi
+kernel32 = windll.kernel32
+advapi32 = windll.advapi32
+
+IsWow64Process=None
+if hasattr(kernel32,'IsWow64Process'):
+ IsWow64Process=kernel32.IsWow64Process
+ IsWow64Process.restype = c_bool
+ IsWow64Process.argtypes = [c_void_p, POINTER(c_bool)]
+
+class WinProcess(BaseProcess):
+
+ def __init__(self, pid=None, name=None, debug=True):
+ """ Create and Open a process object from its pid or from its name """
+ super(WinProcess, self).__init__()
+ if pid:
+ self._open(int(pid), debug=debug)
+
+ elif name:
+ self._open_from_name(name, debug=debug)
+ else:
+ raise ValueError("You need to instanciate process with at least a name or a pid")
+
+ if self.is_64bit():
+ si = self.GetNativeSystemInfo()
+ self.max_addr = si.lpMaximumApplicationAddress
+ else:
+ si = self.GetSystemInfo()
+ self.max_addr = 2147418111
+ self.min_addr = si.lpMinimumApplicationAddress
+
+
+ def __del__(self):
+ self.close()
+
+ def is_64bit(self):
+ if not "64" in platform.machine():
+ return False
+ iswow64 = c_bool(False)
+ if IsWow64Process is None:
+ return False
+ if not IsWow64Process(self.h_process, byref(iswow64)):
+ raise WinError()
+ return not iswow64.value
+
+ @staticmethod
+ def list():
+ processes=[]
+ arr = c_ulong * 256
+ lpidProcess= arr()
+ cb = sizeof(lpidProcess)
+ cbNeeded = c_ulong()
+ hModule = c_ulong()
+ count = c_ulong()
+ modname = create_string_buffer(100)
+ PROCESS_QUERY_INFORMATION = 0x0400
+ PROCESS_VM_READ = 0x0010
+
+ psapi.EnumProcesses(byref(lpidProcess), cb, byref(cbNeeded))
+ nReturned = cbNeeded.value/sizeof(c_ulong())
+
+ pidProcess = [i for i in lpidProcess][:nReturned]
+ for pid in pidProcess:
+ proc={ "pid": int(pid) }
+ hProcess = kernel32.OpenProcess(PROCESS_QUERY_INFORMATION | PROCESS_VM_READ, False, pid)
+ if hProcess:
+ psapi.EnumProcessModules(hProcess, byref(hModule), sizeof(hModule), byref(count))
+ psapi.GetModuleBaseNameA(hProcess, hModule.value, modname, sizeof(modname))
+ proc["name"]=modname.value
+ kernel32.CloseHandle(hProcess)
+ processes.append(proc)
+ return processes
+
+ @staticmethod
+ def processes_from_name(processName):
+ processes = []
+ for process in WinProcess.list():
+ if processName == process.get("name", None) or (process.get("name","").lower().endswith(".exe") and process.get("name","")[:-4]==processName):
+ processes.append(process)
+
+ if len(processes) > 0:
+ return processes
+
+ @staticmethod
+ def name_from_process(dwProcessId):
+ process_list = WinProcess.list()
+ for process in process_list:
+ if process.pid == dwProcessId:
+ return process.get("name", None)
+
+ return False
+
+ def _open(self, dwProcessId, debug=False):
+ if debug:
+ ppsidOwner = DWORD()
+ ppsidGroup = DWORD()
+ ppDacl = DWORD()
+ ppSacl = DWORD()
+ ppSecurityDescriptor = SECURITY_DESCRIPTOR()
+
+ process = kernel32.OpenProcess(262144, 0, dwProcessId)
+ advapi32.GetSecurityInfo(kernel32.GetCurrentProcess(), 6, 0, byref(ppsidOwner), byref(ppsidGroup), byref(ppDacl), byref(ppSacl), byref(ppSecurityDescriptor))
+ advapi32.SetSecurityInfo(process, 6, DACL_SECURITY_INFORMATION | UNPROTECTED_DACL_SECURITY_INFORMATION, None, None, ppSecurityDescriptor.dacl, ppSecurityDescriptor.group)
+ kernel32.CloseHandle(process)
+ self.h_process = kernel32.OpenProcess(2035711, 0, dwProcessId)
+ if self.h_process is not None:
+ self.isProcessOpen = True
+ self.pid = dwProcessId
+ return True
+ return False
+
+ def close(self):
+ if self.h_process is not None:
+ ret = kernel32.CloseHandle(self.h_process) == 1
+ if ret:
+ self.h_process = None
+ self.pid = None
+ self.isProcessOpen = False
+ return ret
+ return False
+
+ def _open_from_name(self, processName, debug=False):
+ processes = self.processes_from_name(processName)
+ if not processes:
+ raise ProcessException("can't get pid from name %s" % processName)
+ elif len(processes)>1:
+ raise ValueError("There is multiple processes with name %s. Please select a process from its pid instead"%processName)
+ if debug:
+ self._open(processes[0]["pid"], debug=True)
+ else:
+ self._open(processes[0]["pid"], debug=False)
+
+ def GetSystemInfo(self):
+ si = SYSTEM_INFO()
+ kernel32.GetSystemInfo(byref(si))
+ return si
+
+ def GetNativeSystemInfo(self):
+ si = SYSTEM_INFO()
+ kernel32.GetNativeSystemInfo(byref(si))
+ return si
+
+ def VirtualQueryEx(self, lpAddress):
+ mbi = MEMORY_BASIC_INFORMATION()
+ if not VirtualQueryEx(self.h_process, lpAddress, byref(mbi), sizeof(mbi)):
+ raise ProcessException('Error VirtualQueryEx: 0x%08X' % lpAddress)
+ return mbi
+
+ def VirtualQueryEx64(self, lpAddress):
+ mbi = MEMORY_BASIC_INFORMATION64()
+ if not VirtualQueryEx64(self.h_process, lpAddress, byref(mbi), sizeof(mbi)):
+ raise ProcessException('Error VirtualQueryEx: 0x%08X' % lpAddress)
+ return mbi
+
+ def VirtualProtectEx(self, base_address, size, protection):
+ old_protect = c_ulong(0)
+ if not kernel32.VirtualProtectEx(self.h_process, base_address, size, protection, byref(old_protect)):
+ raise ProcessException('Error: VirtualProtectEx(%08X, %d, %08X)' % (base_address, size, protection))
+ return old_protect.value
+
+ def iter_region(self, start_offset=None, end_offset=None, protec=None, optimizations=None):
+
+ offset = start_offset or self.min_addr
+ end_offset = end_offset or self.max_addr
+
+ while True:
+ if offset >= end_offset:
+ break
+ mbi = self.VirtualQueryEx(offset)
+ offset = mbi.BaseAddress
+ chunk = mbi.RegionSize
+ protect = mbi.Protect
+ state = mbi.State
+ #print "offset: %s, chunk:%s"%(offset, chunk)
+ if state & MEM_FREE or state & MEM_RESERVE:
+ offset += chunk
+ continue
+ if protec:
+ if not protect & protec or protect & PAGE_NOCACHE or protect & PAGE_WRITECOMBINE or protect & PAGE_GUARD:
+ offset += chunk
+ continue
+ yield offset, chunk
+ offset += chunk
+
+ def write_bytes(self, address, data):
+ address = int(address)
+ if not self.isProcessOpen:
+ raise ProcessException("Can't write_bytes(%s, %s), process %s is not open" % (address, data, self.pid))
+ buffer = create_string_buffer(data)
+ sizeWriten = c_size_t(0)
+ bufferSize = sizeof(buffer) - 1
+ _address = address
+ _length = bufferSize + 1
+ try:
+ old_protect = self.VirtualProtectEx(_address, _length, PAGE_EXECUTE_READWRITE)
+ except:
+ pass
+
+ res = kernel32.WriteProcessMemory(self.h_process, address, buffer, bufferSize, byref(sizeWriten))
+ try:
+ self.VirtualProtectEx(_address, _length, old_protect)
+ except:
+ pass
+
+ return res
+
+ def read_bytes(self, address, bytes = 4, use_NtWow64ReadVirtualMemory64=False):
+ #print "reading %s bytes from addr %s"%(bytes, address)
+ if use_NtWow64ReadVirtualMemory64:
+ if NtWow64ReadVirtualMemory64 is None:
+ raise WindowsError("NtWow64ReadVirtualMemory64 is not available from a 64bit process")
+ RpM = NtWow64ReadVirtualMemory64
+ else:
+ RpM = ReadProcessMemory
+
+ address = int(address)
+ buffer = create_string_buffer(bytes)
+ bytesread = c_size_t(0)
+ data = b''
+ length = bytes
+ while length:
+ if RpM(self.h_process, address, buffer, bytes, byref(bytesread)) or (use_NtWow64ReadVirtualMemory64 and GetLastError() == 0):
+ if bytesread.value:
+ data += buffer.raw[:bytesread.value]
+ length -= bytesread.value
+ address += bytesread.value
+ if not len(data):
+ raise ProcessException('Error %s in ReadProcessMemory(%08x, %d, read=%d)' % (GetLastError(),
+ address,
+ length,
+ bytesread.value))
+ return data
+ else:
+ if GetLastError()==299: #only part of ReadProcessMemory has been done, let's return it
+ data += buffer.raw[:bytesread.value]
+ return data
+ raise WinError()
+ # data += buffer.raw[:bytesread.value]
+ # length -= bytesread.value
+ # address += bytesread.value
+ return data
+
+
+ def list_modules(self):
+ module_list = []
+ if self.pid is not None:
+ hModuleSnap = CreateToolhelp32Snapshot(TH32CS_CLASS.SNAPMODULE, self.pid)
+ if hModuleSnap is not None:
+ module_entry = MODULEENTRY32()
+ module_entry.dwSize = sizeof(module_entry)
+ success = Module32First(hModuleSnap, byref(module_entry))
+ while success:
+ if module_entry.th32ProcessID == self.pid:
+ module_list.append(copy.copy(module_entry))
+ success = Module32Next(hModuleSnap, byref(module_entry))
+
+ kernel32.CloseHandle(hModuleSnap)
+ return module_list
+
+ def get_symbolic_name(self, address):
+ for m in self.list_modules():
+ if int(m.modBaseAddr) <= int(address) < int(m.modBaseAddr + m.modBaseSize):
+ return '%s+0x%08X' % (m.szModule, int(address) - m.modBaseAddr)
+
+ return '0x%08X' % int(address)
+
+ def hasModule(self, module):
+ if module[-4:] != '.dll':
+ module += '.dll'
+ module_list = self.list_modules()
+ for m in module_list:
+ if module in m.szExePath.split('\\'):
+ return True
+ return False
+
+
+ def get_instruction(self, address):
+ """
+ Pydasm disassemble utility function wrapper. Returns the pydasm decoded instruction in self.instruction.
+ """
+ import pydasm
+ try:
+ data = self.read_bytes(int(address), 32)
+ except:
+ return 'Unable to disassemble at %08x' % address
+
+ return pydasm.get_instruction(data, pydasm.MODE_32)
+
generated by cgit v1.2.3 (git 2.39.1) at 2024-09-16 20:05:38 +0000