admin/hakril-PythonForWindows
1
0
Fork 0
mirror of https://github.com/hakril/PythonForWindows synced 2026-06-08 14:31:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

renamed k32testing.py -> winproxy.py

Browse Source
This commit is contained in:
Clement Rouault
2015-09-23 10:29:24 +02:00
parent b1d4f23bb7
commit 663de130b5
10 changed files with 81 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__init__.py
+7 -5
View File
@@ -11,15 +11,17 @@ Exported:
current_thread : :class:`windows.winobject.CurrentThread`
"""
import windows.k32testing
from .utils import VirtualProtected
import windows.winproxy
from .utils import VirtualProtected
from .winobject import System, CurrentProcess, CurrentThread
system = System()
current_process = CurrentProcess()
current_thread = CurrentThread()
import windows.vectored_exception
__all__ = ["system", "VirtualProtected", 'current_process', 'current_thread']
__all__ = ["system", "VirtualProtected", 'current_process', 'current_thread']
# Late import
# Do not move it: risk of circular import
import windows.vectored_exception
injection.py
-2
View File
@@ -19,8 +19,6 @@ def get_loadlib_getproc(target):
return exp['LoadLibraryA'], exp['GetProcAddress']
# 32 to 32 injection
def generate_python_exec_shellcode_32(target, PYDLL_addr, PyInit, PyRun, PYCODE_ADDR):
#LoadLibraryA = utils.get_func_addr('kernel32', 'LoadLibraryA')
native_exec/native_function.py
+2 -2
View File
@@ -4,7 +4,7 @@ import platform
import sys
import windows
import windows.k32testing as k32api
import windows.winproxy
from . import simple_x86 as x86
from . import simple_x64 as x64
@@ -43,7 +43,7 @@ class Win32MyMap(MyMap):
#access = mmap.ACCESS_READ | mmap.ACCESS_WRITE
#return cls(-1, size, access=access)
access = mmap.ACCESS_READ | mmap.ACCESS_WRITE
addr = k32api.VirtualAlloc(0, size, 0x1000, 0x40)
addr = windows.winproxy.VirtualAlloc(0, size, 0x1000, 0x40)
new_map = (ctypes.c_char * size).from_address(addr)
new_map.addr = addr
if new_map.addr == 0:
network.py
+8 -8
View File
@@ -1,5 +1,5 @@
import windows
import windows.k32testing
import windows.winproxy
import ctypes
import socket
import struct
@@ -51,7 +51,7 @@ class TCP4Connection(MIB_TCPROW_OWNER_PID):
closing.dwLocalPort = self.dwLocalPort
closing.dwRemoteAddr = self.dwRemoteAddr
closing.dwRemotePort = self.dwRemotePort
return windows.k32testing.SetTcpEntry(ctypes.byref(closing))
return windows.winproxy.SetTcpEntry(ctypes.byref(closing))
def __repr__(self):
if not self.established:
@@ -129,22 +129,22 @@ def get_MIB_TCP6TABLE_OWNER_PID_from_buffer(buffer):
def get_tcp_ipv4_sockets():
size = ctypes.c_uint(0)
try:
windows.k32testing.GetExtendedTcpTable(None, ctypes.byref(size), ulAf=windows.generated_def.windef.AF_INET)
except windows.k32testing.IphlpapiError as e:
windows.winproxy.GetExtendedTcpTable(None, ctypes.byref(size), ulAf=windows.generated_def.windef.AF_INET)
except windows.winproxy.IphlpapiError as e:
pass # Allow us to set size to the needed value
buffer = (ctypes.c_char * size.value)()
windows.k32testing.GetExtendedTcpTable(buffer, ctypes.byref(size), ulAf=windows.generated_def.windef.AF_INET)
windows.winproxy.GetExtendedTcpTable(buffer, ctypes.byref(size), ulAf=windows.generated_def.windef.AF_INET)
t = get_MIB_TCPTABLE_OWNER_PID_from_buffer(buffer)
return list(t.table)
def get_tcp_ipv6_sockets():
size = ctypes.c_uint(0)
try:
windows.k32testing.GetExtendedTcpTable(None, ctypes.byref(size), ulAf=windows.generated_def.windef.AF_INET6)
except windows.k32testing.IphlpapiError as e:
windows.winproxy.GetExtendedTcpTable(None, ctypes.byref(size), ulAf=windows.generated_def.windef.AF_INET6)
except windows.winproxy.IphlpapiError as e:
pass # Allow us to set size to the needed value
buffer = (ctypes.c_char * size.value)()
windows.k32testing.GetExtendedTcpTable(buffer, ctypes.byref(size), ulAf=windows.generated_def.windef.AF_INET6)
windows.winproxy.GetExtendedTcpTable(buffer, ctypes.byref(size), ulAf=windows.generated_def.windef.AF_INET6)
t = get_MIB_TCP6TABLE_OWNER_PID_from_buffer(buffer)
return list(t.table)
syswow64.py
+3 -3
View File
@@ -2,7 +2,7 @@ import struct
import ctypes
import codecs
import windows
import windows.k32testing as kernel32proxy
import windows.winproxy as winproxy
import windows.native_exec.simple_x64 as x64
from generated_def.winstructs import *
@@ -24,7 +24,7 @@ def execute_64bits_code_from_syswow(shellcode):
"""shellcode must not end by a ret"""
if not windows.current_process.is_wow_64:
raise ValueError("Calling execute_64bits_code_from_syswow from non-syswow process")
addr = windows.k32testing.VirtualAlloc(dwSize=0x1000)
addr = windows.winproxy.VirtualAlloc(dwSize=0x1000)
# post-exec 32bits stub (xor eax, eax; ret)
ret = "\xC3"
ret_addr = addr
@@ -131,7 +131,7 @@ def get_current_process_syswow_peb():
class CurrentProcessReadSyswow():
def read_memory(self, addr, size):
buffer_addr = ctypes.create_string_buffer(size)
kernel32proxy.NtWow64ReadVirtualMemory64(current_process.handle, addr, buffer_addr, size)
windows.winproxy.NtWow64ReadVirtualMemory64(current_process.handle, addr, buffer_addr, size)
return buffer_addr[:]
bitness = 64
peb_addr = get_current_process_syswow_peb_addr()
test/mytest.py
+4 -4
View File
@@ -86,16 +86,16 @@ class WindowsTestCase(unittest.TestCase):
imp = python_module.pe.imports
self.assertIn("kernel32.dll", imp.keys(), 'Kernel32.dll not in python imports')
current_proc_id_iat = [f for f in imp["kernel32.dll"] if f.name == "GetCurrentProcessId"][0]
k32_base = windows.k32testing.LoadLibraryA("kernel32.dll")
self.assertEqual(windows.k32testing.GetProcAddress(k32_base, "GetCurrentProcessId"), current_proc_id_iat.value)
k32_base = windows.winproxy.LoadLibraryA("kernel32.dll")
self.assertEqual(windows.winproxy.GetProcAddress(k32_base, "GetCurrentProcessId"), current_proc_id_iat.value)
def test_local_process_pe_exports(self):
mods = [m for m in windows.current_process.peb.modules if m.name == "kernel32.dll"]
self.assertTrue(mods, 'Could not find "kernel32.dll" in current process modules')
k32 = mods[0]
get_current_proc_id = k32.pe.exports['GetCurrentProcessId']
k32_base = windows.k32testing.LoadLibraryA("kernel32.dll")
self.assertEqual(windows.k32testing.GetProcAddress(k32_base, "GetCurrentProcessId"), get_current_proc_id)
k32_base = windows.winproxy.LoadLibraryA("kernel32.dll")
self.assertEqual(windows.winproxy.GetProcAddress(k32_base, "GetCurrentProcessId"), get_current_proc_id)
# Native execution
utils/winutils.py
+24 -24
View File
@@ -6,7 +6,7 @@ import sys
import code
import windows
from .. import k32testing
from .. import winproxy
from ..generated_def import windef
from ..generated_def.winstructs import *
@@ -30,7 +30,7 @@ def get_remote_func_addr(target, dll_name, func_name):
def is_wow_64(hProcess):
try:
fnIsWow64Process = get_func_addr("kernel32.dll", "IsWow64Process")
except k32testing.Kernel32Error:
except winproxy.Kernel32Error:
return False
IsWow64Process = ctypes.WINFUNCTYPE(BOOL, HANDLE, ctypes.POINTER(BOOL))(fnIsWow64Process)
Wow64Process = BOOL()
@@ -51,16 +51,16 @@ def get_handle_from_file(f):
def create_console():
"""Create a new console displaying STDOUT
Useful in injection of GUI process"""
k32testing.AllocConsole()
stdout_handle = k32testing.GetStdHandle(windef.STD_OUTPUT_HANDLE)
winproxy.AllocConsole()
stdout_handle = winproxy.GetStdHandle(windef.STD_OUTPUT_HANDLE)
console_stdout = create_file_from_handle(stdout_handle, "w")
sys.stdout = console_stdout
stdin_handle = k32testing.GetStdHandle(windef.STD_INPUT_HANDLE)
stdin_handle = winproxy.GetStdHandle(windef.STD_INPUT_HANDLE)
console_stdin = create_file_from_handle(stdin_handle, "r+")
sys.stdin = console_stdin
stderr_handle = k32testing.GetStdHandle(windef.STD_ERROR_HANDLE)
stderr_handle = winproxy.GetStdHandle(windef.STD_ERROR_HANDLE)
console_stderr = create_file_from_handle(stderr_handle, "w")
sys.stderr = console_stderr
@@ -72,7 +72,7 @@ def create_process(path, show_windows=False):
StartupInfo.cb = ctypes.sizeof(StartupInfo)
StartupInfo.dwFlags = 0
lpStartupInfo = ctypes.byref(StartupInfo)
windows.k32testing.CreateProcessA(path, lpProcessInformation=ctypes.byref(proc_info), lpStartupInfo=lpStartupInfo)
windows.winproxy.CreateProcessA(path, lpProcessInformation=ctypes.byref(proc_info), lpStartupInfo=lpStartupInfo)
proc = [p for p in windows.system.processes if p.pid == proc_info.dwProcessId][0]
return proc
@@ -82,17 +82,17 @@ def enable_privilege(lpszPrivilege, bEnablePrivilege):
luid = LUID()
hToken = HANDLE()
k32testing.OpenProcessToken(k32testing.GetCurrentProcess(), TOKEN_ALL_ACCESS, byref(hToken))
k32testing.LookupPrivilegeValueA(None, lpszPrivilege, byref(luid))
winproxy.OpenProcessToken(winproxy.GetCurrentProcess(), TOKEN_ALL_ACCESS, byref(hToken))
winproxy.LookupPrivilegeValueA(None, lpszPrivilege, byref(luid))
tp.PrivilegeCount = 1
tp.Privileges[0].Luid = luid
if bEnablePrivilege:
tp.Privileges[0].Attributes = SE_PRIVILEGE_ENABLED
else:
tp.Privileges[0].Attributes = 0
k32testing.AdjustTokenPrivileges(hToken, False, byref(tp), sizeof(TOKEN_PRIVILEGES))
k32testing.CloseHandle(hToken)
if k32testing.GetLastError() == windef.ERROR_NOT_ALL_ASSIGNED:
winproxy.AdjustTokenPrivileges(hToken, False, byref(tp), sizeof(TOKEN_PRIVILEGES))
winproxy.CloseHandle(hToken)
if winproxy.GetLastError() == windef.ERROR_NOT_ALL_ASSIGNED:
raise ValueError("Failed to get privilege {0}".format(lpszPrivilege))
return True
@@ -104,9 +104,9 @@ def check_is_elevated():
cbsize = DWORD()
bcsize = sizeof(elevation)
k32testing.OpenProcessToken(k32testing.GetCurrentProcess(), TOKEN_ALL_ACCESS, byref(hToken))
k32testing.GetTokenInformation(hToken, TokenElevation, byref(elevation), sizeof(elevation), byref(cbsize))
k32testing.CloseHandle(hToken)
winproxy.OpenProcessToken(winproxy.GetCurrentProcess(), TOKEN_ALL_ACCESS, byref(hToken))
winproxy.GetTokenInformation(hToken, TokenElevation, byref(elevation), sizeof(elevation), byref(cbsize))
winproxy.CloseHandle(hToken)
return elevation.TokenIsElevated
def check_debug():
@@ -116,9 +116,9 @@ def check_debug():
cbsize = DWORD(1024)
bufferres = (c_char * cbsize.value)()
k32testing.RegOpenKeyExA(HKEY_LOCAL_MACHINE, "System\\CurrentControlSet\\Control", 0, KEY_READ, byref(hkresult))
k32testing.RegGetValueA(hkresult, None, "SystemStartOptions", RRF_RT_REG_SZ, None, byref(bufferres), byref(cbsize))
k32testing.RegCloseKey(hkresult)
winproxy.RegOpenKeyExA(HKEY_LOCAL_MACHINE, "System\\CurrentControlSet\\Control", 0, KEY_READ, byref(hkresult))
winproxy.RegGetValueA(hkresult, None, "SystemStartOptions", RRF_RT_REG_SZ, None, byref(bufferres), byref(cbsize))
winproxy.RegCloseKey(hkresult)
control = bufferres[:]
if "DEBUG" not in control:
@@ -144,10 +144,10 @@ def pop_shell():
def get_kernel_modules():
cbsize = DWORD()
k32testing.NtQuerySystemInformation(SystemModuleInformation, None, 0, byref(cbsize))
winproxy.NtQuerySystemInformation(SystemModuleInformation, None, 0, byref(cbsize))
raw_buffer = (cbsize.value * c_char)()
buffer = SYSTEM_MODULE_INFORMATION.from_address(ctypes.addressof(raw_buffer))
k32testing.NtQuerySystemInformation(SystemModuleInformation, byref(raw_buffer), sizeof(raw_buffer), byref(cbsize))
winproxy.NtQuerySystemInformation(SystemModuleInformation, byref(raw_buffer), sizeof(raw_buffer), byref(cbsize))
modules = (SYSTEM_MODULE * buffer.ModulesCount).from_address(addressof(buffer) + SYSTEM_MODULE_INFORMATION.Modules.offset)
return list(modules)
@@ -168,20 +168,20 @@ class VirtualProtected(object):
def __enter__(self):
self.old_protect = DWORD()
k32testing.VirtualProtect(self.addr, self.size, self.new_protect, ctypes.byref(self.old_protect))
winproxy.VirtualProtect(self.addr, self.size, self.new_protect, ctypes.byref(self.old_protect))
return self
def __exit__(self, exc_type, exc_value, traceback):
k32testing.VirtualProtect(self.addr, self.size, self.old_protect.value, ctypes.byref(self.old_protect))
winproxy.VirtualProtect(self.addr, self.size, self.old_protect.value, ctypes.byref(self.old_protect))
return False
class DisableWow64FsRedirection(object):
"""A context manager that disable the Wow64 Fs Redirection"""
def __enter__(self):
self.OldValue = PVOID()
k32testing.Wow64DisableWow64FsRedirection(ctypes.byref(self.OldValue))
winproxy.Wow64DisableWow64FsRedirection(ctypes.byref(self.OldValue))
return self
def __exit__(self, exc_type, exc_value, traceback):
k32testing.Wow64RevertWow64FsRedirection(self.OldValue)
winproxy.Wow64RevertWow64FsRedirection(self.OldValue)
return False
vectored_exception.py
+2 -2
View File
@@ -184,9 +184,9 @@ class WithExceptionHandler(object):
self.handler = VectoredException(handler)
def __enter__(self):
self.value = windows.k32testing.AddVectoredExceptionHandler(0, self.handler)
self.value = windows.winproxy.AddVectoredExceptionHandler(0, self.handler)
return self
def __exit__(self, exc_type, exc_value, traceback):
windows.k32testing.RemoveVectoredExceptionHandler(self.value)
windows.winproxy.RemoveVectoredExceptionHandler(self.value)
return False
winobject.py
+31 -31
View File
@@ -7,7 +7,7 @@ import struct
import windows
import windows.syswow64
import windows.k32testing as kernel32proxy
import windows.winproxy as winproxy
import windows.injection as injection
import windows.native_exec as native_exec
import windows.native_exec.simple_x86 as x86
@@ -43,7 +43,7 @@ class AutoHandle(object):
def __del__(self):
if hasattr(self, "_handle") and self._handle:
kernel32proxy.CloseHandle(self._handle)
winproxy.CloseHandle(self._handle)
class System(object):
"""Represent the current windows system python is running on"""
@@ -82,11 +82,11 @@ class System(object):
def enumerate_processes():
process_entry = WinProcess()
process_entry.dwSize = ctypes.sizeof(process_entry)
snap = kernel32proxy.CreateToolhelp32Snapshot(windef.TH32CS_SNAPPROCESS, 0)
kernel32proxy.Process32First(snap, process_entry)
snap = winproxy.CreateToolhelp32Snapshot(windef.TH32CS_SNAPPROCESS, 0)
winproxy.Process32First(snap, process_entry)
res = []
res.append(utils.swallow_ctypes_copy(process_entry))
while kernel32proxy.Process32Next(snap, process_entry):
while winproxy.Process32Next(snap, process_entry):
res.append(utils.swallow_ctypes_copy(process_entry))
return res
@@ -94,11 +94,11 @@ class System(object):
def enumerate_threads():
thread_entry = WinThread()
thread_entry.dwSize = ctypes.sizeof(thread_entry)
snap = kernel32proxy.CreateToolhelp32Snapshot(windef.TH32CS_SNAPTHREAD, 0)
snap = winproxy.CreateToolhelp32Snapshot(windef.TH32CS_SNAPTHREAD, 0)
threads = []
kernel32proxy.Thread32First(snap, thread_entry)
winproxy.Thread32First(snap, thread_entry)
threads.append(copy.copy(thread_entry))
while kernel32proxy.Thread32Next(snap, thread_entry):
while winproxy.Thread32Next(snap, thread_entry):
threads.append(copy.copy(thread_entry))
return threads
@@ -129,23 +129,23 @@ class WinThread(THREADENTRY32, AutoHandle):
def context(self):
x = windows.vectored_exception.EnhancedCONTEXT()
x.ContextFlags = CONTEXT_FULL
kernel32proxy.GetThreadContext(self.handle, x)
winproxy.GetThreadContext(self.handle, x)
return x
def set_context(self, context):
return kernel32proxy.SetThreadContext(self.handle, context)
return winproxy.SetThreadContext(self.handle, context)
def exit(self, code=0):
return kernel32proxy.TerminateThread(self.handle, code)
return winproxy.TerminateThread(self.handle, code)
def resume(self):
return kernel32proxy.ResumeThread(self.handle)
return winproxy.ResumeThread(self.handle)
def suspend(self):
return kernel32proxy.SuspendThread(self.handle)
return winproxy.SuspendThread(self.handle)
def _get_handle(self):
return kernel32proxy.OpenThread(dwThreadId=self.tid)
return winproxy.OpenThread(dwThreadId=self.tid)
def __repr__(self):
owner = self.owner
@@ -157,7 +157,7 @@ class WinThread(THREADENTRY32, AutoHandle):
@staticmethod
def _from_handle(handle):
tid = kernel32proxy.GetThreadId(handle)
tid = winproxy.GetThreadId(handle)
print(tid)
try:
return [t for t in System().threads if t.tid == tid][0]
@@ -210,7 +210,7 @@ class CurrentThread(AutoHandle):
@utils.fixedpropety
def tid(self):
"""Thread ID"""
return kernel32proxy.GetCurrentThreadId()
return winproxy.GetCurrentThreadId()
@utils.fixedpropety
def owner(self):
@@ -221,14 +221,14 @@ class CurrentThread(AutoHandle):
return windows.current_process
def _get_handle(self):
return kernel32proxy.GetCurrentThread()
return winproxy.GetCurrentThread()
def __del__(self):
pass
def exit(self, code=0):
"""Exit the thread"""
return kernel32proxy.ExitThread(code)
return winproxy.ExitThread(code)
@@ -259,7 +259,7 @@ class CurrentProcess(Process):
return get_peb
def _get_handle(self):
return kernel32proxy.GetCurrentProcess()
return winproxy.GetCurrentProcess()
def __del__(self):
pass
@@ -303,7 +303,7 @@ class CurrentProcess(Process):
:returns: int
"""
return kernel32proxy.VirtualAlloc(dwSize=size)
return winproxy.VirtualAlloc(dwSize=size)
def write_memory(self, addr, data):
"""Write data at addr"""
@@ -323,12 +323,12 @@ class CurrentProcess(Process):
.. note::
CreateThread https://msdn.microsoft.com/en-us/library/windows/desktop/ms682453%28v=vs.85%29.aspx
"""
handle = kernel32proxy.CreateThread(lpStartAddress=lpStartAddress, lpParameter=lpParameter, dwCreationFlags=dwCreationFlags)
handle = winproxy.CreateThread(lpStartAddress=lpStartAddress, lpParameter=lpParameter, dwCreationFlags=dwCreationFlags)
return WinThread._from_handle(handle)
def exit(self, code=0):
"""Exit the process"""
return kernel32proxy.ExitProcess(code)
return winproxy.ExitProcess(code)
class WinProcess(PROCESSENTRY32, Process):
"""A Process on the system"""
@@ -360,7 +360,7 @@ class WinProcess(PROCESSENTRY32, Process):
return self.th32ParentProcessID
def _get_handle(self):
return kernel32proxy.OpenProcess(dwProcessId=self.pid)
return winproxy.OpenProcess(dwProcessId=self.pid)
def __repr__(self):
return '<{0} "{1}" pid {2} at {3}>'.format(self.__class__.__name__, self.name, self.pid, hex(id(self)))
@@ -370,19 +370,19 @@ class WinProcess(PROCESSENTRY32, Process):
:returns: int
"""
return kernel32proxy.VirtualAllocEx(self.handle, dwSize=size)
return winproxy.VirtualAllocEx(self.handle, dwSize=size)
def write_memory(self, addr, data):
"""Write `data` at `addr`"""
return kernel32proxy.WriteProcessMemory(self.handle, addr, lpBuffer=data)
return winproxy.WriteProcessMemory(self.handle, addr, lpBuffer=data)
def low_read_memory(self, addr, buffer_addr, size):
if windows.current_process.bitness == 32 and self.bitness == 64:
#OptionalExport can be None (see k32testing.py)
if kernel32proxy.NtWow64ReadVirtualMemory64 is None:
#OptionalExport can be None (see winproxy.py)
if winproxy.NtWow64ReadVirtualMemory64 is None:
raise ValueError("NtWow64ReadVirtualMemory64 non available in ntdll: cannot write into 64bits processus")
return kernel32proxy.NtWow64ReadVirtualMemory64(self.handle, addr, buffer_addr, size)
return kernel32proxy.ReadProcessMemory(self.handle, addr, lpBuffer=buffer_addr, nSize=size)
return winproxy.NtWow64ReadVirtualMemory64(self.handle, addr, buffer_addr, size)
return winproxy.ReadProcessMemory(self.handle, addr, lpBuffer=buffer_addr, nSize=size)
def read_memory(self, addr, size):
"""Read `size` from `addr`"""
@@ -417,7 +417,7 @@ class WinProcess(PROCESSENTRY32, Process):
"""Create a remote thread"""
if windows.current_process.bitness == 32 and self.bitness == 64:
return windows.syswow64.NtCreateThreadEx_32_to_64(self, addr, param)
return WinThread._from_handle(kernel32proxy.CreateRemoteThread(hProcess=self.handle, lpStartAddress=addr, lpParameter=param))
return WinThread._from_handle(winproxy.CreateRemoteThread(hProcess=self.handle, lpStartAddress=addr, lpParameter=param))
def load_library(self, dll_path):
"""Load the library in remote process"""
@@ -466,7 +466,7 @@ class WinProcess(PROCESSENTRY32, Process):
def exit(self, code=0):
"""Exit the process"""
return kernel32proxy.TerminateProcess(self.handle, code)
return winproxy.TerminateProcess(self.handle, code)
class LoadedModule(LDR_DATA_TABLE_ENTRY):
k32testing.py → winproxy.py
View File