mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
Merge pull request #74 from hakril/improve_injection_exceptions
Improve injection error reporting
This commit is contained in:
hakril
@@ -0,0 +1,75 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import pytest
|
||||
|
||||
import weakref
|
||||
import shutil
|
||||
import time
|
||||
import os
|
||||
|
||||
import windows
|
||||
import windows.generated_def as gdef
|
||||
|
||||
from .conftest import pop_proc_32, pop_proc_64
|
||||
from .pfwtest import DEFAULT_CREATION_FLAGS
|
||||
|
||||
@pytest.fixture(params=
|
||||
[(pop_proc_32, DEFAULT_CREATION_FLAGS),
|
||||
(pop_proc_32, gdef.CREATE_SUSPENDED),
|
||||
(pop_proc_64, DEFAULT_CREATION_FLAGS),
|
||||
(pop_proc_64, gdef.CREATE_SUSPENDED)],
|
||||
ids=["proc32", "proc32susp", "proc64", "proc64susp"])
|
||||
def proc_3264_runsus(request):
|
||||
"""Fixture for process 32/64 both running & suspended"""
|
||||
proc_poper, dwCreationFlags = request.param
|
||||
proc = proc_poper(dwCreationFlags=dwCreationFlags)
|
||||
time.sleep(0.2) # Give time to the process to load :)
|
||||
print("Created {0} ({1}bits) for test".format(proc, proc.bitness))
|
||||
yield weakref.proxy(proc) # provide the fixture value
|
||||
try:
|
||||
proc.exit(0)
|
||||
except WindowsError as e:
|
||||
if not proc.is_exit:
|
||||
raise
|
||||
# print("DEL PROC")
|
||||
del proc
|
||||
|
||||
# Its really the same test as test_process.test_load_library but with suspended process as well
|
||||
def test_dll_injection(proc_3264_runsus):
|
||||
assert (not proc_3264_runsus.peb.Ldr) or ("wintrust.dll" not in [mod.name for mod in proc_3264_runsus.peb.modules])
|
||||
modaddr = windows.injection.load_dll_in_remote_process(proc_3264_runsus, "wintrust.dll")
|
||||
wintrustmod = [mod for mod in proc_3264_runsus.peb.modules if mod.name == "wintrust.dll"][0]
|
||||
assert wintrustmod.baseaddr == modaddr
|
||||
|
||||
def test_dll_injection_error_reporting(proc_3264_runsus):
|
||||
with pytest.raises(windows.injection.InjectionFailedError) as excinfo:
|
||||
windows.injection.load_dll_in_remote_process(proc_3264_runsus, "NO_A_DLL.dll")
|
||||
assert excinfo.value.__cause__.winerror == gdef.ERROR_MOD_NOT_FOUND
|
||||
|
||||
def test_dll_injection_access_denied(proc_3264_runsus, tmpdir):
|
||||
"""Emulate injection of MsStore python, were its DLL are not executable by any other append
|
||||
See: https://github.com/hakril/PythonForWindows/issues/72
|
||||
"""
|
||||
mybitness = windows.current_process.bitness
|
||||
if proc_3264_runsus.bitness == mybitness:
|
||||
DLLPATH = r"c:\windows\system32\wintrust.dll"
|
||||
elif mybitness == 64: # target is 32
|
||||
DLLPATH = r"c:\windows\syswow64\wintrust.dll"
|
||||
elif mybitness == 32: # target is 64
|
||||
DLLPATH = r"c:\windows\sysnative\wintrust.dll"
|
||||
else:
|
||||
raise Value("WTF ARE THE BITNESS ?")
|
||||
targetname = os.path.join(str(tmpdir), "wintrust_noexec.dll")
|
||||
shutil.copy(DLLPATH, targetname)
|
||||
# Deny Execute; allow read for everyone
|
||||
sd = windows.security.SecurityDescriptor.from_string("D:(D;;GXFX;;;WD)(A;;1;;;WD)")
|
||||
sd.to_filename(targetname)
|
||||
|
||||
try:
|
||||
with pytest.raises(windows.injection.InjectionFailedError) as excinfo:
|
||||
windows.injection.load_dll_in_remote_process(proc_3264_runsus, targetname)
|
||||
assert excinfo.value.__cause__.winerror == gdef.ERROR_ACCESS_DENIED
|
||||
finally:
|
||||
proc_3264_runsus.exit()
|
||||
proc_3264_runsus.wait()
|
||||
time.sleep(0.5) # Fail on Azure CI of no sleep
|
||||
os.unlink(targetname)
|
||||
@@ -370,6 +370,11 @@ class TestProcessWithCheckGarbage(object):
|
||||
proc32_64.load_library(DLL)
|
||||
assert DLL in [m.name for m in proc32_64.peb.modules]
|
||||
|
||||
def test_load_library_suspended(self, proc32_64_suspended):
|
||||
DLL = "wintrust.dll"
|
||||
proc32_64_suspended.load_library(DLL)
|
||||
assert DLL in [m.name for m in proc32_64_suspended.peb.modules]
|
||||
|
||||
def test_load_library_unicode_name(self, proc32_64, tmpdir):
|
||||
mybitness = windows.current_process.bitness
|
||||
UNICODE_FILENAME = u'\u4e2d\u56fd\u94f6\u884c\u7f51\u94f6\u52a9\u624b.dll'
|
||||
|
||||
+92
-24
@@ -40,6 +40,14 @@ def perform_manual_getproc_loadlib_32(target, dll_name):
|
||||
code += x86.Call(":FUNC_GETPROCADDRESS32")
|
||||
code += x86.Push(x86.mem("[ECX + 8]"))
|
||||
code += x86.Call("EAX") # LoadLibrary
|
||||
code += x86.Cmp("EAX", 0)
|
||||
code += x86.Jnz(":end")
|
||||
# GetLastError()
|
||||
# I really don't want to resolve another function
|
||||
# For a field that have been the same since XP/Win2003
|
||||
code += x86.Mov('EAX', x86.mem('fs:[0x34]'))
|
||||
code += x86.Add("EAX", 0x80000000)
|
||||
code += x86.Label(":end")
|
||||
code += x86.Pop("ECX")
|
||||
code += x86.Pop("ECX")
|
||||
code += x86.Ret()
|
||||
@@ -60,9 +68,18 @@ def perform_manual_getproc_loadlib_32(target, dll_name):
|
||||
|
||||
t = target.execute(RemoteManualLoadLibray.get_code(), addr4)
|
||||
t.wait()
|
||||
if not t.exit_code:
|
||||
raise InjectionFailedError("Injection of <{0}> failed".format(dll_name))
|
||||
return True
|
||||
module_baseaddr = t.exit_code
|
||||
|
||||
if module_baseaddr & 0x80000000:
|
||||
# Not a possible userland addr -> its a GetLastError()
|
||||
error_code = module_baseaddr & 0x7fffffff
|
||||
module_baseaddr = None
|
||||
real_error = ctypes.WinError(error_code)
|
||||
myexc = InjectionFailedError(u"Injection of <{0}> failed due to error <{1}> in injected process".format(dll_name, str(real_error)))
|
||||
myexc.__cause__ = real_error
|
||||
raise myexc
|
||||
|
||||
return module_baseaddr
|
||||
|
||||
def perform_manual_getproc_loadlib_64(target, dll_name):
|
||||
dll = get_kernel32_dll_name()
|
||||
@@ -76,18 +93,23 @@ def perform_manual_getproc_loadlib_64(target, dll_name):
|
||||
code += x64.Mov("RDX", x64.mem("[R15 + 8]"))
|
||||
code += x64.Call(":FUNC_GETPROCADDRESS64")
|
||||
code += x64.Mov("RCX", x64.mem("[R15 + 0x10]"))
|
||||
code += x64.Push("RCX")
|
||||
code += x64.Push("RCX")
|
||||
code += x64.Push("RCX")
|
||||
code += (x64.Push("RCX") * 3)
|
||||
code += x64.Call("RAX") # LoadLibrary
|
||||
code += x64.Pop("RCX")
|
||||
code += x64.Pop("RCX")
|
||||
code += x64.Pop("RCX")
|
||||
code += (x64.Pop("RCX") * 3)
|
||||
code += x64.Mov("RCX", x64.mem("[R15]"))
|
||||
code += x64.Mov(x64.mem("[RCX]"), "RAX")
|
||||
# GetLastError()
|
||||
# I really don't want to resolve another function
|
||||
# For a field that have been the same since XP/Win2003
|
||||
code += x64.Mov('RAX', x64.mem('gs:[0x68]'))
|
||||
code += x64.Ret()
|
||||
|
||||
RemoteManualLoadLibray += GetProcAddress64
|
||||
|
||||
with target.allocated_memory(0x1000) as addr:
|
||||
# Addr contains the name of kernel32
|
||||
# The data at addr are discadable after the call
|
||||
# So, on return it contains the return PVOID64 value of LoadLibraryA
|
||||
addr2 = addr + len(dll)
|
||||
addr3 = addr2 + len(api)
|
||||
addr4 = addr3 + len(dll_to_load)
|
||||
@@ -101,17 +123,55 @@ def perform_manual_getproc_loadlib_64(target, dll_name):
|
||||
|
||||
t = target.execute(RemoteManualLoadLibray.get_code(), addr4)
|
||||
t.wait()
|
||||
if not t.exit_code:
|
||||
raise InjectionFailedError("Injection of <{0}> failed".format(dll_name))
|
||||
return True
|
||||
module_baseaddr = target.read_ptr(addr)
|
||||
if not module_baseaddr:
|
||||
module_baseaddr = None
|
||||
real_error = ctypes.WinError(t.exit_code)
|
||||
myexc = InjectionFailedError(u"Injection of <{0}> failed due to error <{1}> in injected process".format(dll_name, str(real_error)))
|
||||
myexc.__cause__ = real_error
|
||||
raise myexc
|
||||
|
||||
def generate_simple_LoadLibraryW_64(load_libraryW, remote_store):
|
||||
return module_baseaddr
|
||||
|
||||
def generate_simple_LoadLibraryW_32_with_error(k32):
|
||||
"""A shellcode that execute LoadLibraryW(param) and returns the value.
|
||||
If LoadLibraryW fails -> returns (GetLastError | 0x10000000)
|
||||
|
||||
As a valid 32b modules will never be in >=0x80000000,
|
||||
this allow to determine if the call was successful of not"""
|
||||
load_libraryW = k32.pe.exports["LoadLibraryW"]
|
||||
GetLastError = k32.pe.exports["GetLastError"]
|
||||
|
||||
code = x86.MultipleInstr()
|
||||
code += x86.Mov("EAX", x86.mem("[ESP + 4]"))
|
||||
code += x86.Push("EAX")
|
||||
code += x86.Mov("EAX", load_libraryW)
|
||||
code += x86.Call("EAX")
|
||||
code += x86.Cmp("EAX", 0)
|
||||
code += x86.Jnz(":end")
|
||||
code += x86.Mov("EAX", GetLastError)
|
||||
code += x86.Call("EAX")
|
||||
code += x86.Add("EAX", 0x80000000)
|
||||
code += x86.Label(":end")
|
||||
code += x86.Ret()
|
||||
return code.get_code()
|
||||
|
||||
def generate_simple_LoadLibraryW_64_with_error(k32, remote_store):
|
||||
"""A shellcode that execute LoadLibraryW(param) and store the value at a fixed address.
|
||||
This allow a 32b process to inject and retrieve a 64bit module address
|
||||
|
||||
Thread return value is the result of GetLastError()
|
||||
"""
|
||||
load_libraryW = k32.pe.exports["LoadLibraryW"]
|
||||
GetLastError = k32.pe.exports["GetLastError"]
|
||||
code = RemoteLoadLibrayStub = x64.MultipleInstr()
|
||||
code += x64.Mov("RAX", load_libraryW)
|
||||
code += (x64.Push("RDI") * 5) # Prepare stack
|
||||
code += x64.Call("RAX")
|
||||
code += (x64.Pop("RDI") * 5) # Clean stack
|
||||
code += x64.Mov(x64.deref(remote_store), "RAX")
|
||||
code += x64.Mov("RAX", GetLastError) # Add a jump ?
|
||||
code += x64.Call("RAX")
|
||||
code += (x64.Pop("RDI") * 5) # Clean stack
|
||||
code += x64.Ret()
|
||||
return RemoteLoadLibrayStub.get_code()
|
||||
|
||||
@@ -136,17 +196,21 @@ def load_dll_in_remote_process(target, dll_path):
|
||||
if k32:
|
||||
# We have kernel32 \o/
|
||||
k32 = k32[0]
|
||||
try:
|
||||
load_libraryW = k32.pe.exports["LoadLibraryW"]
|
||||
except KeyError:
|
||||
raise ValueError("Kernel32 have no export <LoadLibraryA> (wtf)")
|
||||
|
||||
with target.allocated_memory(0x1000) as addr:
|
||||
if target.bitness == 32:
|
||||
target.write_memory(addr, (dll_path + "\x00").encode('utf-16le'))
|
||||
t = target.create_thread(load_libraryW, addr)
|
||||
shellcode32 = generate_simple_LoadLibraryW_32_with_error(k32)
|
||||
encoded_dll_name = (dll_path + "\x00").encode('utf-16le')
|
||||
paramaddr = addr
|
||||
target.write_memory(addr, encoded_dll_name)
|
||||
shellcode_addr = addr + len(encoded_dll_name)
|
||||
target.write_memory(shellcode_addr, shellcode32)
|
||||
t = target.create_thread(shellcode_addr, paramaddr)
|
||||
t.wait()
|
||||
module_baseaddr = t.exit_code
|
||||
exit_code = module_baseaddr = t.exit_code
|
||||
if module_baseaddr & 0x80000000:
|
||||
# Not a possible userland addr -> its a GetLastError()
|
||||
module_baseaddr = None
|
||||
exit_code = exit_code & 0x7fffffff
|
||||
else:
|
||||
# For 64b target we need a special stub as the return value of
|
||||
# load_libraryW does not fit in t.exit_code (DWORD)
|
||||
@@ -158,14 +222,18 @@ def load_dll_in_remote_process(target, dll_path):
|
||||
param_addr = addr
|
||||
addr += len(full_dll_name)
|
||||
shellcode_addr = addr
|
||||
shellcode = generate_simple_LoadLibraryW_64(load_libraryW, retval_addr)
|
||||
shellcode = generate_simple_LoadLibraryW_64_with_error(k32, retval_addr)
|
||||
target.write_memory(shellcode_addr, shellcode)
|
||||
t = target.create_thread(shellcode_addr, param_addr)
|
||||
t.wait()
|
||||
exit_code = t.exit_code
|
||||
module_baseaddr = target.read_ptr(retval_addr)
|
||||
|
||||
if not module_baseaddr:
|
||||
raise InjectionFailedError(u"Injection of <{0}> failed".format(dll_path))
|
||||
real_error = ctypes.WinError(exit_code)
|
||||
myexc = InjectionFailedError(u"Injection of <{0}> failed due to error <{1}> in injected process".format(dll_path, str(real_error)))
|
||||
myexc.__cause__ = real_error
|
||||
raise myexc
|
||||
dbgprint("DLL Injected via LoadLibray", "DLLINJECT")
|
||||
# Cannot return the full return value of load_libraryW in 64b target.. (exit_code is a DWORD)
|
||||
return module_baseaddr
|
||||
|
||||
Reference in New Issue
Block a user