16. Samples of code¶
16.1. Processes¶
16.1.1. windows.current_process¶
import sys
import os.path
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
import windows.native_exec.simple_x86 as x86
import windows.native_exec.simple_x64 as x64
# Here is our current process
cp = windows.current_process
print("current process is {cp}".format(cp=windows.current_process))
print("current process is a <{cp.bitness}> bits process".format(cp=cp))
print("current process is a SysWow64 process ? <{cp.is_wow_64}>".format(cp=cp))
print("current process pid <{cp.pid}> and ppid <{cp.ppid}>".format(cp=cp))
print("Here are the current process threads: <{cp.threads}>".format(cp=cp))
print("Let's execute some native code ! (0x41 + 1)")
if windows.current_process.bitness == 32:
# Let's generate some native code
code = x86.MultipleInstr()
code += x86.Mov("Eax", 0x41)
code += x86.Inc("EAX")
code += x86.Ret()
else:
code = x64.MultipleInstr()
code += x64.Mov("RAX", 0x41)
code += x64.Inc("RAX")
code += x64.Ret()
native_code = code.get_code()
v = windows.current_process.execute(native_code)
print("Native code returned <{0}>".format(hex(v)))
print("Allocating memory in current process")
addr = cp.virtual_alloc(0x1000) # Default alloc is RWX (so secure !)
print("Allocated memory is at <{0}>".format(hex(addr)))
print("Writing 'SOME STUFF' in allocation memory")
cp.write_memory(addr, "SOME STUFF")
print("Reading memory : <{0}>".format(repr(cp.read_memory(addr, 20))))
Output:
(cmd λ) python32.exe process\current_process.py
current process is <windows.winobject.process.CurrentProcess object at 0x030A2590>
current process is a <32> bits process
current process is a SysWow64 process ? <True>
current process pid <8264> and ppid <4100>
Here are the current process threads: <[<WinThread 13540 owner "python.exe" at 0x32d3210>]>
Let's execute some native code ! (0x41 + 1)
Native code returned <0x42>
Allocating memory in current process
Allocated memory is at <0xd60000>
Writing 'SOME STUFF' in allocation memory
Reading memory : <'SOME STUFF\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'>
16.1.2. Remote process : WinProcess¶
import sys
import os.path
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
import windows.native_exec.simple_x86 as x86
import windows.native_exec.simple_x64 as x64
print("Creating a notepad") ## Replaced calc.exe by notepad.exe cause of windows 10.
notepad = windows.utils.create_process(r"C:\windows\system32\notepad.exe")
# You don't need to do that in our case, but it's useful to now
print("Looking for notepads in the processes")
all_notepads = [proc for proc in windows.system.processes if proc.name == "notepad.exe"]
print("They are currently <{0}> notepads running on the system".format(len(all_notepads)))
print("Let's play with our notepad: <{notepad}>".format(notepad=notepad))
print("Our notepad pid is {notepad.pid}".format(notepad=notepad))
print("Our notepad is a <{notepad.bitness}> bits process".format(notepad=notepad))
print("Our notepad is a SysWow64 process ? <{notepad.is_wow_64}>".format(notepad=notepad))
print("Our notepad have threads ! <{notepad.threads}>".format(notepad=notepad))
# PEB STUFF
peb = notepad.peb
print("Exploring our notepad PEB ! {peb}".format(peb=peb))
print("Command line is {peb.commandline}".format(peb=peb))
modules = peb.modules
print("Here are 3 loaded modules: {0}".format(modules[:3]))
# See iat_hook.py for module exploration
# Remote alloc / read / write
print("Allocating memory in our notepad")
addr = notepad.virtual_alloc(0x1000)
print("Allocated memory is at <{0}>".format(hex(addr)))
print("Writing 'SOME STUFF' in allocated memory")
notepad.write_memory(addr, "SOME STUFF")
print("Reading allocated memory : <{0}>".format(repr(notepad.read_memory(addr, 20))))
# Remote Execution
print("Execution some native code in our notepad (write 0x424242 at allocated address + return 0x1337)")
if notepad.bitness == 32:
# Let's generate some native code
code = x86.MultipleInstr()
code += x86.Mov(x86.deref(addr), 0x42424242)
code += x86.Mov("EAX", 0x1337)
code += x86.Ret()
else:
code = x64.MultipleInstr()
code += x64.Mov('RAX', addr)
code += x64.Mov(x64.mem("[RAX]"), 0x42424242)
code += x64.Mov("RAX", 0x1337)
code += x64.Ret()
print("Executing native code !")
t = notepad.execute(code.get_code())
t.wait()
print("Return code = {0}".format(hex(t.exit_code)))
print("Reading allocated memory : <{0}>".format(repr(notepad.read_memory(addr, 20))))
print("Executing python code !")
# Make 'windows' importable in remote python
notepad.execute_python("import sys; sys.path.append(r'{0}')".format(sys.path[-1]))
notepad.execute_python("import windows")
# Let's write in the notepad 'current_process' memory :)
notepad.execute_python("addr = {addr}; windows.current_process.write_memory(addr, 'HELLO FROM notepad')".format(addr=addr))
print("Reading allocated memory : <{0}>".format(repr(notepad.read_memory(addr, 20))))
# python_execute is 'safe':
# - it waits for the thread completion
# - it raise an error if remote code raised some
try:
print("Trying to import in remote module 'FAKE_MODULE'")
notepad.execute_python("def func():\n import FAKE_MODULE\nfunc()")
except windows.injection.RemotePythonError as e:
print("Remote ERROR !")
print(e)
print("That's all ! killing the notepad")
notepad.exit()
Output:
(cmd λ) python.exe process\remote_process.py
Creating a notepad
Looking for notepads in the processes
They are currently <1> notepads running on the system
Let's play with our notepad: <<WinProcess "notepad.exe" pid 2044 at 0x40ce850>>
Our notepad pid is 2044
Our notepad is a <32> bits process
Our notepad is a SysWow64 process ? <True>
Our notepad have threads ! <[<WinThread 7700 owner "notepad.exe" at 0x41faee0>, <WinThread 7264 owner "notepad.exe" at 0x41faf30>, ...]>
Exploring our notepad PEB ! <windows.winobject.process.RemotePEB object at 0x03F6CDA0>
Command line is <RemoteWinUnicodeString ""C:\windows\system32\notepad.exe"" at 0x3f6cf80>
Here are 3 loaded modules: [<RemoteLoadedModule "notepad.exe" at 0x3f6cf30>, <RemoteLoadedModule "ntdll.dll" at 0x3f6ce40>, <RemoteLoadedModule "kernel32.dll" at 0x3f6cee0>]
Allocating memory in our notepad
Allocated memory is at <0x6f80000>
Writing 'SOME STUFF' in allocated memory
Reading allocated memory : <'SOME STUFF\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'>
Execution some native code in our notepad (write 0x424242 at allocated address + return 0x1337)
Executing native code !
Return code = 0x1337L
Reading allocated memory : <'BBBB STUFF\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'>
Executing python code !
Reading allocated memory : <'HELLO FROM notepad\x00\x00'>
Trying to import in remote module 'FAKE_MODULE'
Remote ERROR !
Traceback (most recent call last):
File "<string>", line 3, in <module>
File "<string>", line 2, in func
ImportError: No module named FAKE_MODULE
That's all ! killing the notepad
16.1.3. PEB exploration¶
import sys
import os.path
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
print("Exploring the current process PEB")
peb = windows.current_process.peb
print("PEB is <{0}>".format(peb))
commandline = peb.commandline
print("Commandline object is {0}".format(commandline))
print("Commandline string is {0}".format(repr(commandline.Buffer)))
imagepath = peb.imagepath
print("Imagepath {0}".format(imagepath))
modules = peb.modules
print("Printing some modules: {0}".format("\n".join(str(m) for m in modules[:6])))
print("=== K32 ===")
print("Looking for kernel32.dll")
k32 = [m for m in modules if m.name == "kernel32.dll"][0]
print("Kernel32 module: {0}".format(k32))
print("Module name = <{0}> | Fullname = <{1}>".format(k32.name, k32.fullname))
print("Kernel32 is loaded at address {0}".format(hex(k32.baseaddr)))
print("=== K32 PE ===")
k32pe = k32.pe
print("PE Representation of k32: {0}".format(k32pe))
exports = k32pe.exports
some_exports = dict((k,v) for k,v in exports.items() if k in [0, 42, "VirtualAlloc", "CreateFileA"])
print("Here are some exports {0}".format(some_exports))
imports = k32pe.imports
print("Import DLL dependancies are (without api-*): {0}".format([x for x in imports.keys() if not x.startswith("api-")]))
NtCreateFile_iat = [x for x in imports["ntdll.dll"] if x.name == "NtCreateFile"][0]
print("IAT Entry for ntdll!NtCreateFile = {0} | addr = {1}".format(NtCreateFile_iat, hex(NtCreateFile_iat.addr)))
print("Sections: {0}".format(k32pe.sections))
Output:
(cmd λ) python.exe process\peb.py
Exploring the current process PEB
PEB is <<windows.winobject.PEB object at 0x02649B70>>
Commandline object is <WinUnicodeString "python.exe peb.py " at 0x2649c60>
Commandline string is u'python.exe peb.py '
Imagepath <WinUnicodeString "C:\Python27\python.exe" at 0x2649d50>
Printing some modules: <LoadedModule "python.exe" at 0x272a030>
<LoadedModule "ntdll.dll" at 0x272a080>
<LoadedModule "kernel32.dll" at 0x272acb0>
<LoadedModule "kernelbase.dll" at 0x272ad00>
<LoadedModule "python27.dll" at 0x272ad50>
<LoadedModule "msvcr90.dll" at 0x272ada0>
=== K32 ===
Looking for kernel32.dll
Kernel32 module: <LoadedModule "kernel32.dll" at 0x272acb0>
Module name = <kernel32.dll> | Fullname = <C:\Windows\SYSTEM32\KERNEL32.DLL>
Kernel32 is loaded at address 0x774c0000
=== K32 PE ===
PE Representation of k32: <windows.pe_parse.PEFile object at 0x0272D350>
Here are some exports {0: 2001566688L, u'CreateFileA': 2001635616L, 42: 2001647872L, u'VirtualAlloc': 2001570704L}
Import DLL dependancies are (without api-*): [u'ntdll.dll', u'kernelbase.dll']
IAT Entry for ntdll!NtCreateFile = <IATEntry "NtCreateFile" ordinal 253> | addr = 0x77541128L
Sections: [<PESection ".text">, <PESection ".rdata">, <PESection ".data">, <PESection ".rsrc">, <PESection ".reloc">]
16.1.4. IAT hooking¶
import sys
import os.path
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import _winreg
import windows
# Here is a demo of IAT hooking in python
# We will hook the 'RegOpenKeyExA' entry of Python27.dll because it is easy to trigger !
# First: let's create our hook
# windows.hooks.RegOpenKeyExACallback is generated based on windows.generated_def.winfuncs
@windows.hooks.RegOpenKeyExACallback
def open_reg_hook(hKey, lpSubKey, ulOptions, samDesired, phkResult, real_function):
print("<in hook> Hook called | hKey = {0} | lpSubKey = <{1}>".format(hex(hKey), lpSubKey.value))
# Our hook can choose to call the real_function or not
if "SECRET" in lpSubKey.value:
print("<in hook> Secret key asked, returning magic handle 0x12345678")
# We must respect the hooked method return-value interface
phkResult[0] = 0x12345678
return 0
if "FAIL" in lpSubKey.value:
print("<in hook> Asked for a failing key: returning 0x2a")
return 42
print("<in hook> Non-secret key : calling normal function")
return real_function()
# Get the peb of our process
peb = windows.current_process.peb
# Get the pythonxx.dll module
pythondll_module = [m for m in peb.modules if m.name.startswith("python") and m.name.endswith(".dll")][0]
# Get the iat entries for DLL advapi32.dll
adv_imports = pythondll_module.pe.imports['advapi32.dll']
# Get RegOpenKeyExA iat entry
RegOpenKeyExA_iat = [n for n in adv_imports if n.name == "RegOpenKeyExA"][0]
# Setup our hook
RegOpenKeyExA_iat.set_hook(open_reg_hook)
# Use python native module _winreg that call 'RegOpenKeyExA'
print("Asking for <MY_SECRET_KEY>")
v = _winreg.OpenKey(1234567, "MY_SECRET_KEY")
print("Result = " + hex(v.handle))
print("")
print("Asking for <MY_FAIL_KEY>")
try:
v = _winreg.OpenKey(1234567, "MY_FAIL_KEY")
print("Result = " + hex(v.handle))
except WindowsError as e:
print(repr(e))
print("")
print("Asking for <HKEY_CURRENT_USER/Software>")
try:
v = _winreg.OpenKey(_winreg.HKEY_CURRENT_USER, "Software")
print("Result = " + hex(v.handle))
except WindowsError as e:
print(repr(e))
Output:
(cmd λ) python process\iat_hook.py
Asking for <MY_SECRET_KEY>
<in hook> Hook called | hKey = 0x12d687 | lpSubKey = <MY_SECRET_KEY>
<in hook> Secret key asked, returning magic handle 0x12345678
Result = 0x12345678
Asking for <MY_FAIL_KEY>
<in hook> Hook called | hKey = 0x12d687 | lpSubKey = <MY_FAIL_KEY>
<in hook> Asked for a failing key: returning 0x2a
WindowsError(42, 'Windows Error 0x2A')
Asking for <HKEY_CURRENT_USER/Software>
<in hook> Hook called | hKey = 0x80000001L | lpSubKey = <Software>
<in hook> Non-secret key : calling normal function
Result = 0x108
16.2. windows.system¶
import sys
import os.path
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
system = windows.system
print("Basic system infos:")
print(" version = {0}".format(system.version))
print(" bitness = {0}".format(system.bitness))
print(" computer_name = {0}".format(system.computer_name))
print(" product_type = {0}".format(system.product_type))
print(" version_name = {0}".format(system.version_name))
print("")
print("There is {0} processes".format(len(system.processes)))
print("There is {0} threads".format(len(system.threads)))
print("")
print("Dumping first logical drive:")
drive = system.logicaldrives[0]
print(" " + str(drive))
print((" " * 8) + "name = {0}".format(drive.name))
print((" " * 8) + "type = {0}".format(drive.type))
print((" " * 8) + "path = {0}".format(drive.path))
print("")
print("Dumping first service:")
serv = windows.system.services[0]
print(" " + str(serv))
print((" " * 8) + "name = {0}".format(serv.name))
print((" " * 8) + "description = {0}".format(serv.description))
print((" " * 8) + "status = {0}".format(serv.status))
print((" " * 8) + "process = {0}".format(repr(serv.process)))
print("")
print("Finding a service in a user process:")
serv = [s for s in windows.system.services if s.process][0]
print(" " + str(serv))
print((" " * 8) + "name = {0}".format(serv.name))
print((" " * 8) + "description = {0}".format(serv.description))
print((" " * 8) + "status = {0}".format(serv.status))
print((" " * 8) + "process = {0}".format(repr(serv.process)))
print("")
print("Enumerating handles:")
handles = system.handles
print(" There are {0} handles:".format(len(handles)))
print(" First handle is: " + str(handles[0]))
print(" Enumerating handles of the current process:")
cp_handles = [h for h in system.handles if h.dwProcessId == windows.current_process.pid]
print(" There are {0} handles for this process".format(len(cp_handles)))
print(" Looking for a File handle:")
file_h = [h for h in cp_handles if h.type == "File"][0]
print(" Handle is {0}".format(file_h))
print(" Name is <{0}>".format(file_h.name))
Output:
(cmd λ) python system.py
Basic system infos:
version = (6, 3)
bitness = 64
computer_name = HAKRIL-PC
product_type = VER_NT_WORKSTATION(0x1L)
version_name = Windows 8.1
There is 117 processes
There is 1246 threads
Dumping first logical drive:
<LogicalDrive "C:\" (DRIVE_FIXED)>
name = C:\
type = DRIVE_FIXED(0x3L)
path = \Device\HarddiskVolume2
Dumping first service:
<ServiceA "ACPI">
name = ACPI
description = Microsoft ACPI Driver
status = ServiceStatus(type=SERVICE_KERNEL_DRIVER(0x1L), state=SERVICE_RUNNING(0x4L), control_accepted=1L, flags=0L)
process = None
Finding a service in a user process:
<ServiceA "Appinfo">
name = Appinfo
description = Application Information
status = ServiceStatus(type=SERVICE_WIN32_SHARE_PROCESS(0x20L), state=SERVICE_RUNNING(0x4L), control_accepted=129L, flags=0L)
process = <WinProcess "svchost.exe" pid 988 at 0x2e64750>
Enumerating handles:
There are 40664 handles:
First handle is: <Handle value=<0x4> in process pid=4>
Enumerating handles of the current process:
There are 255 handles for this process
Looking for a File handle:
Handle is <Handle value=<0x4> in process pid=14340>
Name is <\Device\ConDrv>
16.3. Network - socket exploration¶
import sys
import os.path
import socket
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
if not windows.utils.check_is_elevated():
print("!!! Demo will fail because closing a connection require elevated process !!!")
print("Working on ipv4")
conns = windows.system.network.ipv4
print("== Listening ==")
print("Some listening connections: {0}".format([c for c in conns if not c.established][:3]))
print("Listening ports are : {0}".format([c.local_port for c in conns if not c.established]))
print("== Established ==")
print("Some established connections: {0}".format([c for c in conns if c.established][:3]))
TARGET_HOST = "localhost"
TARGET_PORT = 80
print("== connection to {0}:{1} ==".format(TARGET_HOST, TARGET_PORT))
s = socket.create_connection((TARGET_HOST, TARGET_PORT))
our_connection = [c for c in windows.system.network.ipv4 if c.established and c.remote_port == TARGET_PORT and c.remote_addr == s.getpeername()[0]]
print("Our connection is {0}".format(our_connection))
print("Sending YOP")
s.send("YOP")
print("Closing socket")
our_connection[0].close()
print("Sending LAIT")
s.send("LAIT")
Output:
(cmd λ) python.exe network\network.py
Working on ipv4
== Listening ==
Some listening connections: [<TCP IPV4 Listening socket on 0.0.0.0:80>, <TCP IPV4 Listening socket on 0.0.0.0:135>, <TCP IPV4 Listening socket on 0.0.0.0:443>]
Listening ports are : [80, 135, 443, 445, 902, 912, 5357, 49152, 49153, 49154, 49155, 49157, 49159, 8307, 25340, 139, 139]
== Established ==
Some established connections: [<TCP IPV4 Connection 127.0.0.1:25340 -> 127.0.0.1:49472>, <TCP IPV4 Connection 127.0.0.1:49173 -> 127.0.0.1:49174>, <TCP IPV4 Connection 127.0.0.1:49174 -> 127.0.0.1:49173>]
== connection to localhost:80 ==
Our connection is [<TCP IPV4 Connection 127.0.0.1:49616 -> 127.0.0.1:80>]
Sending YOP
Closing socket
Sending LAIT
Traceback (most recent call last):
File ".\network.py", line 45, in <module>
s.send("LAIT")
socket.error: [Errno 10054] An existing connection was forcibly closed by the remote host
16.4. Registry¶
import sys
import os.path
import pprint
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
registry = windows.system.registry
print("Registry is <{0}>".format(registry))
current_user = registry("HKEY_CURRENT_USER")
print("HKEY_CURRENT_USER is <{0}>".format(current_user))
subkeys_name = [s.name for s in current_user.subkeys]
print("HKEY_CURRENT_USER subkeys names are:")
pprint.pprint(subkeys_name)
print("Opening 'Software' in HKEY_CURRENT_USER: {0}".format(current_user("Software")))
print("We can also open it in one access: {0}".format(registry(r"HKEY_CURRENT_USER\Sofware")))
print("Looking at CurrentVersion")
windows_info = registry("HKEY_LOCAL_MACHINE\SOFTWARE\Microsoft\Windows NT\CurrentVersion")
print("Key is {0}".format(windows_info))
print("values are:")
pprint.pprint(windows_info.values)
registered_owner = windows_info["RegisteredOwner"]
print("registered owner = <{0}>".format(registered_owner))
Output:
(cmd λ) python.exe registry\registry.py
Registry is <<windows.registry.Registry object at 0x02941290>>
HKEY_CURRENT_USER is <<PyHKey "HKEY_CURRENT_USER">>
HKEY_CURRENT_USER subkeys names are:
['AppEvents',
'AppXBackupContentType',
'Console',
'Control Panel',
'Environment',
'EUDC',
'Identities',
'Keyboard Layout',
'Network',
'Printers',
'Software',
'System',
'Volatile Environment']
Opening 'Software' in HKEY_CURRENT_USER: <PyHKey "HKEY_CURRENT_USER\Software">
We can also open it in one access: <PyHKey "HKEY_CURRENT_USER\Sofware">
Looking at CurrentVersion
Key is <PyHKey "HKEY_LOCAL_MACHINE\SOFTWARE\Microsoft\Windows NT\CurrentVersion">
values are:
[KeyValue(name='SoftwareType', value=u'System', type=1),
KeyValue(name='RegisteredOwner', value=u'hakril', type=1),
KeyValue(name='InstallDate', value=0, type=4),
...
KeyValue(name='PathName', value=u'C:\\Windows', type=1)]
registered owner = <KeyValue(name='RegisteredOwner', value=u'hakril', type=1)>
16.5. windows.wintrust¶
import sys
import os.path
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows.wintrust
TARGET_FILE = r"C:\windows\system32\ntdll.dll"
print("Checking signature of <{0}>".format(TARGET_FILE))
print(" is_signed: <{0}>".format(windows.wintrust.is_signed(TARGET_FILE)))
print(" check_signature: <{0}>".format(windows.wintrust.check_signature(TARGET_FILE)))
sign_info = windows.wintrust.full_signature_information(TARGET_FILE)
print(" full_signature_information:")
print(" * signed <{0}>".format(sign_info.signed))
print(" * catalog <{0}>".format(sign_info.catalog))
print(" * catalogsigned <{0}>".format(sign_info.catalogsigned))
print(" * additionalinfo <{0}>".format(sign_info.additionalinfo))
print("Checking signature of some loaded DLL")
for module in windows.current_process.peb.modules[:5]:
path = module.fullname
is_signed = windows.wintrust.is_signed(path)
if is_signed:
print("<{0}> : {1}".format(path, is_signed))
else:
sign_info = windows.wintrust.full_signature_information(path)
print("<{0}> : {1} ({2})".format(path, is_signed, sign_info[3]))
Output:
(cmd λ) python crypto\wintrust.py
Checking signature of <C:\windows\system32\ntdll.dll>
is_signed: <True>
check_signature: <0>
full_signature_information:
* signed <True>
* catalog <C:\Windows\system32\CatRoot\{F750E6C3-38EE-11D1-85E5-00C04FC295EE}\Package_35_for_KB3128650~31bf3856ad364e35~amd64~~6.3.1.2.cat>
* catalogsigned <True>
* additionalinfo <0>
Checking signature of some loaded DLL
<c:\python27\python.exe> : False (TRUST_E_NOSIGNATURE(0x800b0100L))
<c:\windows\system32\ntdll.dll> : True
<c:\windows\system32\kernel32.dll> : True
<c:\windows\system32\kernelbase.dll> : True
<c:\windows\system32\python27.dll> : False (TRUST_E_NOSIGNATURE(0x800b0100L))
16.6. VectoredException()¶
16.6.1. In local process¶
import sys
import os.path
import pprint
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import ctypes
import windows
from windows.winobject.exception import VectoredException
import windows.generated_def.windef as windef
from windows.generated_def.winstructs import *
@VectoredException
def handler(exc):
print("==Entry of VEH handler==")
if exc[0].ExceptionRecord[0].ExceptionCode == EXCEPTION_ACCESS_VIOLATION:
target_addr = ctypes.cast(exc[0].ExceptionRecord[0].ExceptionInformation[1], ctypes.c_void_p).value
print("Instr at {0} accessed to addr {1}".format(hex(exc[0].ExceptionRecord[0].ExceptionAddress), hex(target_addr)))
print("Resetting page protection to <PAGE_READWRITE>")
windows.winproxy.VirtualProtect(target_page, 0x1000, windef.PAGE_READWRITE)
exc[0].ContextRecord[0].EEFlags.TF = 1
return windef.EXCEPTION_CONTINUE_EXECUTION
else:
print("Exception of type {0}".format(exc[0].ExceptionRecord[0].ExceptionCode))
print("Resetting page protection to <PAGE_NOACCESS>")
windows.winproxy.VirtualProtect(target_page, 0x1000, windef.PAGE_NOACCESS)
return windef.EXCEPTION_CONTINUE_EXECUTION
windows.winproxy.AddVectoredExceptionHandler(0, handler)
target_page = windows.current_process.virtual_alloc(0x1000)
print("Protected page is at <{0}>".format(hex(target_page)))
print("Setting page protection to <PAGE_NOACCESS>")
windows.winproxy.VirtualProtect(target_page, 0x1000, windef.PAGE_NOACCESS)
print("")
v = ctypes.c_uint.from_address(target_page).value
print("Value 1 read")
print("")
v = ctypes.c_uint.from_address(target_page + 0x10).value
print("Value 2 read")
Output:
(cmd λ) python.exe process\veh_segv.py
Protected page is at <0x1db0000>
Setting page protection to <PAGE_NOACCESS>
==Entry of VEH handler==
Instr at 0x1d1ab574 accessed to addr 0x1db0000
Resetting page protection to <PAGE_READWRITE>
==Entry of VEH handler==
Exception of type EXCEPTION_SINGLE_STEP(0x80000004L)
Resetting page protection to <PAGE_NOACCESS>
Value 1 read
==Entry of VEH handler==
Instr at 0x1d1ab574 accessed to addr 0x1db0010
Resetting page protection to <PAGE_READWRITE>
==Entry of VEH handler==
Exception of type EXCEPTION_SINGLE_STEP(0x80000004L)
Resetting page protection to <PAGE_NOACCESS>
Value 2 read
16.6.2. In remote process¶
import sys
import os.path
import pprint
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
import windows.test
from windows.generated_def.winstructs import *
python_code = """
import windows
import ctypes
import windows
from windows.winobject.exception import VectoredException
import windows.generated_def.windef as windef
from windows.generated_def.winstructs import *
windows.utils.create_console()
module_to_trace = "gdi32.dll"
nb_repeat = [5]
@VectoredException
def handler(exc):
if exc[0].ExceptionRecord[0].ExceptionCode == EXCEPTION_ACCESS_VIOLATION:
print("")
target_addr = ctypes.cast(exc[0].ExceptionRecord[0].ExceptionInformation[1], ctypes.c_void_p).value
print("Instr at {0} accessed to addr {1} ({2})".format(hex(exc[0].ExceptionRecord[0].ExceptionAddress), hex(target_addr), module_to_trace))
windows.winproxy.VirtualProtect(target_page, code_size, windef.PAGE_EXECUTE_READWRITE)
nb_repeat[0] -= 1
if nb_repeat[0]:
exc[0].ContextRecord[0].EEFlags.TF = 1
else:
print("No more tracing !")
return windef.EXCEPTION_CONTINUE_EXECUTION
else:
print("Exception of type {0}".format(exc[0].ExceptionRecord[0].ExceptionCode))
print("Resetting page protection to <PAGE_READWRITE>")
windows.winproxy.VirtualProtect(target_page, code_size, windef.PAGE_READWRITE)
return windef.EXCEPTION_CONTINUE_EXECUTION
windows.winproxy.AddVectoredExceptionHandler(0, handler)
print("Tracing execution in module: <{0}>".format(module_to_trace))
module = [x for x in windows.current_process.peb.modules if x.name == module_to_trace][0]
target_page = module.baseaddr
code_size = module.pe.get_OptionalHeader().SizeOfCode
print("Protected page is at {0}".format(hex(target_page)))
windows.winproxy.VirtualProtect(target_page, code_size, windef.PAGE_READWRITE)
"""
c = windows.test.pop_proc_64(dwCreationFlags=CREATE_SUSPENDED)
x = c.execute_python(python_code)
c.threads[0].resume()
import time
time.sleep(0.1)
for t in c.threads:
t.suspend()
time.sleep(1)
c.exit()
Output:
(cmd λ) python.exe process\remote_veh_segv.py
(In another console)
Tracing execution in module: <gdi32.dll>
Protected page is at 0x7ffa3c700000L
Instr at 0x7ffa3c70f0f0L accessed to addr 0x7ffa3c70f0f0L (gdi32.dll)
Exception of type EXCEPTION_SINGLE_STEP(0x80000004L)
Resetting page protection to <PAGE_READWRITE>
Instr at 0x7ffa3c70f0f5L accessed to addr 0x7ffa3c70f0f5L (gdi32.dll)
Exception of type EXCEPTION_SINGLE_STEP(0x80000004L)
Resetting page protection to <PAGE_READWRITE>
Instr at 0x7ffa3c70f0faL accessed to addr 0x7ffa3c70f0faL (gdi32.dll)
Exception of type EXCEPTION_SINGLE_STEP(0x80000004L)
Resetting page protection to <PAGE_READWRITE>
Instr at 0x7ffa3c70f0ffL accessed to addr 0x7ffa3c70f0ffL (gdi32.dll)
Exception of type EXCEPTION_SINGLE_STEP(0x80000004L)
Resetting page protection to <PAGE_READWRITE>
Instr at 0x7ffa3c70f100L accessed to addr 0x7ffa3c70f100L (gdi32.dll)
No more tracing !
16.7. Debugging¶
16.7.1. Debugger¶
import sys
import os.path
import pprint
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
import windows.test
import windows.debug
from windows.generated_def.winstructs import *
class MyDebugger(windows.debug.Debugger):
def on_exception(self, exception):
code = exception.ExceptionRecord.ExceptionCode
addr = exception.ExceptionRecord.ExceptionAddress
print("Got exception {0} at 0x{1:x}".format(code, addr))
class PrintUnicodeString(windows.debug.Breakpoint):
def __init__(self, addr, argument_position):
super(PrintUnicodeString, self).__init__(addr)
self.arg_pos = argument_position
def trigger(self, dbg, exc):
p = dbg.current_process
t = dbg.current_thread
esp = t.context.Esp
unicode_string_addr = p.read_ptr(esp + (self.arg_pos + 1) * 4)
wstring_addr = p.read_ptr(unicode_string_addr + 4)
dll_loaded = p.read_wstring(wstring_addr)
print("Loading <{0}>".format(dll_loaded))
if dll_loaded.endswith("ole32.dll"):
print("Ask to load <ole32.dll>: exiting process")
dbg.current_process.exit()
calc = windows.test.pop_proc_32(dwCreationFlags=DEBUG_PROCESS)
d = MyDebugger(calc)
d.add_bp(PrintUnicodeString("ntdll!LdrLoadDll", argument_position=2))
d.loop()
Ouput:
(cmd λ) python.exe debug\debugger_print_LdrLoaddll.py
Loading <KERNEL32.DLL>
Got exception EXCEPTION_BREAKPOINT(0x80000003L) at 0x77a73bad
Loading <C:\Windows\system32\IMM32.DLL>
Loading <C:\Windows\system32\uxtheme.dll>
Loading <C:\Windows\system32\uxtheme.dll>
Loading <C:\Windows\system32\uxtheme.dll>
Loading <C:\Windows\system32\uxtheme.dll>
Loading <kernel32.dll>
Loading <C:\Windows\WinSxS\x86_microsoft.windows.gdiplus_6595b64144ccf1df_1.1.9600.17415_none_dad8722c5bcc2d8f\gdiplus.dll>
Loading <comctl32.dll>
Loading <comctl32.dll>
Loading <comctl32.dll>
Loading <C:\Windows\system32\shell32.dll>
Loading <C:\Windows\SYSTEM32\WINMM.dll>
Loading <C:\Windows\system32\ole32.dll>
Ask to load <ole32.dll>: exiting process
16.7.1.1. Single stepping¶
import sys
import os.path
import pprint
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
import windows.test
import windows.debug
import windows.native_exec.simple_x86 as x86
from windows.generated_def.winstructs import *
class MyDebugger(windows.debug.Debugger):
def __init__(self, *args, **kwargs):
super(MyDebugger, self).__init__(*args, **kwargs)
self.single_step_counter = 0
def on_exception(self, exception):
code = exception.ExceptionRecord.ExceptionCode
addr = exception.ExceptionRecord.ExceptionAddress
print("Got exception {0} at 0x{1:x}".format(code, addr))
def on_single_step(self, exception):
code = exception.ExceptionRecord.ExceptionCode
addr = exception.ExceptionRecord.ExceptionAddress
print("Got single_step {0} at 0x{1:x}".format(code, addr))
self.single_step_counter -= 1
if self.single_step_counter > 0:
return self.single_step()
else:
print("No more single step: exiting")
self.current_process.exit()
class SingleStepOnWrite(windows.debug.MemoryBreakpoint):
"""Check that BP/dbg can trigger single step and that instruction follows"""
def trigger(self, dbg, exc):
fault_addr = exc.ExceptionRecord.ExceptionInformation[1]
eip = dbg.current_thread.context.pc
print("Instruction at <{0:#x}> wrote at <{1:#x}>".format(eip, fault_addr))
dbg.single_step_counter = 4
return dbg.single_step()
calc = windows.test.pop_proc_32(dwCreationFlags=DEBUG_PROCESS)
d = MyDebugger(calc)
code = calc.virtual_alloc(0x1000)
data = calc.virtual_alloc(0x1000)
injected = x86.MultipleInstr()
injected += x86.Mov("EAX", 0)
injected += x86.Mov(x86.deref(data), "EAX")
injected += x86.Add("EAX", 4)
injected += x86.Mov(x86.deref(data + 4), "EAX")
injected += x86.Add("EAX", 8)
injected += x86.Mov(x86.deref(data + 8), "EAX")
injected += x86.Nop()
injected += x86.Nop()
injected += x86.Ret()
calc.write_memory(code, injected.get_code())
d.add_bp(SingleStepOnWrite(data, size=8, events="W"))
calc.create_thread(code, 0)
d.loop()
Ouput:
(cmd λ) python.exe debug\debugger_membp_singlestep.py
Got exception EXCEPTION_BREAKPOINT(0x80000003L) at 0x77ae3c7d
Instruction at <0x8d0006> wrote at <0x8e0000>
Got single_step EXCEPTION_SINGLE_STEP(0x80000004L) at 0x8d000c
Got single_step EXCEPTION_SINGLE_STEP(0x80000004L) at 0x8d0011
Instruction at <0x8d0011> wrote at <0x8e0004>
Got single_step EXCEPTION_SINGLE_STEP(0x80000004L) at 0x8d0017
Got single_step EXCEPTION_SINGLE_STEP(0x80000004L) at 0x8d001c
Got single_step EXCEPTION_SINGLE_STEP(0x80000004L) at 0x8d0022
Got single_step EXCEPTION_SINGLE_STEP(0x80000004L) at 0x8d0023
No more single step: exiting
16.7.1.2. windows.debug.FunctionBP¶
import sys
import os.path
import pprint
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
import windows.test
import windows.debug
from windows.generated_def.winstructs import *
class FollowNtCreateFile(windows.debug.FunctionBP):
TARGET = windows.winproxy.NtCreateFile
COUNTER = 3
def trigger(self, dbg, exc):
if not self.COUNTER:
print("Exiting process")
dbg.current_process.exit()
return
params = self.extract_arguments(dbg.current_process, dbg.current_thread)
filename = params["ObjectAttributes"].contents.ObjectName.contents.Buffer
handle_addr = params["FileHandle"].value
self.data = (filename, handle_addr)
self.break_on_ret(dbg, exc)
def ret_trigger(self, dbg, exc):
filename, handle_addr = self.data
ret_value = dbg.current_thread.context.func_result # EAX / RAX depending of bitness
handle_value = dbg.current_process.read_ptr(handle_addr)
if ret_value:
print("NtCreateFile of <{0}> FAILED (result={1:#x})".format(filename, ret_value))
return
print("NtCreateFile of <{0}>: handle = {1:#x}".format(filename, handle_value))
# Manual verification
fhandle = [h for h in windows.system.handles if h.dwProcessId == dbg.current_process.pid and h.wValue == handle_value]
if not fhandle:
raise ValueError("handle not found!")
fhandle = fhandle[0]
print("Handle manually found! typename=<{0}>, name=<{1}>".format(fhandle.type, fhandle.name))
print("")
self.COUNTER -= 1
if __name__ == "__main__":
calc = windows.test.pop_proc_32(dwCreationFlags=DEBUG_PROCESS)
d = windows.debug.Debugger(calc)
d.add_bp(FollowNtCreateFile())
d.loop()
Ouput:
(cmd λ) python.exe debug\debug_functionbp.py
NtCreateFile of <\??\C:\Windows\syswow64\en-US\calc.exe.mui>: handle = 0xac
Handle manually found! typename=<File>, name=<\Device\HarddiskVolume2\Windows\SysWOW64\en-US\calc.exe.mui>
NtCreateFile of <\Device\DeviceApi\CMApi>: handle = 0x108
Handle manually found! typename=<File>, name=<\Device\DeviceApi>
NtCreateFile of <\??\C:\Windows\Fonts\staticcache.dat>: handle = 0x154
Handle manually found! typename=<File>, name=<\Device\HarddiskVolume2\Windows\Fonts\StaticCache.dat>
Exiting process
16.7.1.3. Debugger.attach¶
import sys
import os.path
import pprint
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
import windows.test
import windows.debug
from windows.generated_def.winstructs import *
# Just a debugger that follow NtCreateFile and print filename & handler
from debug_functionbp import FollowNtCreateFile
def follow_create_file(pid):
print("Finding process with pid <{0}>".format(pid))
target = [p for p in windows.system.processes if p.pid == pid][0]
print("Target is {0}".format(target))
dbg = windows.debug.Debugger.attach(target)
print("Debugger attached: {0}".format(dbg))
print("")
dbg.add_bp(FollowNtCreateFile())
dbg.loop()
if __name__ == "__main__":
# Create a non-debugged process safe to debug
calc = windows.test.pop_proc_32(dwCreationFlags=0)
# Give ovnly the PID to follow_create_file
follow_create_file(calc.pid)
Ouput:
(cmd λ) python.exe debug\attach.py
Finding process with pid <11392>
Target is <WinProcess "notepad.exe" pid 11392 at 0x471a750>
Debugger attached: <windows.debug.debugger.Debugger object at 0x04707EF0>
NtCreateFile of <\??\C:\Windows\Fonts\staticcache.dat>: handle = 0x288
Handle manually found! typename=<File>, name=<\Device\HarddiskVolume4\Windows\Fonts\StaticCache.dat>
NtCreateFile of <\??\C:\WINDOWS\Registration\R000000000015.clb>: handle = 0x320
Handle manually found! typename=<File>, name=<\Device\HarddiskVolume4\Windows\Registration\R000000000015.clb>
NtCreateFile of <\??\C:\WINDOWS\Globalization\Sorting\sortdefault.nls>: handle = 0x334
Handle manually found! typename=<File>, name=<\Device\HarddiskVolume4\Windows\Globalization\Sorting\SortDefault.nls>
Exiting process
16.7.1.4. Native code tester¶
import sys
import argparse
import windows
import windows.test
import windows.debug as dbg
import windows.native_exec.simple_x86 as x86
import windows.native_exec.simple_x64 as x64
from windows.generated_def import *
def hexdump(string, start_addr=0):
result = ""
if len(string) == 0:
return
ascii = list("."*256)
for i in range(1,0x7f):
ascii[i] = chr(i)
ascii[0x0] = "."
ascii[0x7] = "."
ascii[0x8] = "."
ascii[0x9] = "."
ascii[0xa] = "."
ascii[0x1b] = "."
ascii[0xd] = "."
ascii[0xff] = "\x13"
ascii = "".join(ascii)
offset = 0
while (offset+0x10) <= len(string):
line = string[offset:(offset+0x10)]
linebuf = " %08X " % (offset + start_addr)
for i in range(0,16):
if i == 8:
linebuf += " "
linebuf += "%02X " % ord(line[i])
linebuf += " "
for i in range(0,16):
linebuf += ascii[ord(line[i])]
result += linebuf+"\n"
offset += 0x10
if (len(string) % 0x10) > 0:
linebuf = " %08X " % (offset + start_addr)
for i in range((len(string)-(len(string) % 0x10)),(len(string))):
if i == 8:
linebuf += " "
linebuf += "%02X " % ord(string[i])
linebuf += " "*(0x10-(len(string) % 0x10))
linebuf += " "
for i in range((len(string)-(len(string) % 0x10)),(len(string))):
linebuf += ascii[ord(string[i])]
result += linebuf+"\n"
return result
class CodeTesteur(dbg.Debugger):
def __init__(self, process, code, register_start={}):
super(CodeTesteur, self).__init__(process)
self.initial_code = code
code += "\xcc"
code_addr = self.write_code_in_target(process, code)
register_start["pc"] = code_addr
self.thread_exec = process.threads[0]
self.context_exec = self.thread_exec.context
self.setup_target_context(self.context_exec, register_start)
print("Startup context is:")
self.context_exec.dump()
print(self.context_exec.EEFlags)
self.thread_exec.suspend()
self.thread_exec.set_context(self.context_exec)
self.thread_exec.resume()
self.init_breakpoint = False
def write_code_in_target(self, process, code):
addr = process.virtual_alloc(len(code))
process.write_memory(addr, code)
return addr
def setup_target_context(self, ctx, register_start):
for name, value in register_start.items():
if not hasattr(ctx, name):
raise ValueError("Unknown register to setup <{0}>".format(name))
setattr(ctx, name, value)
def on_exception(self, x):
exc_code = x.ExceptionRecord.ExceptionCode
exc_addr = x.ExceptionRecord.ExceptionAddress
if not self.init_breakpoint and exc_code == EXCEPTION_BREAKPOINT:
self.init_breakpoint = True
return
ctx = self.current_thread.context
print("==Post-exec context==")
ctx.dump()
print(ctx.EEFlags)
if exc_code == EXCEPTION_BREAKPOINT and exc_addr == self.context_exec.pc + len(self.initial_code):
print("<Normal terminaison>")
else:
print("<{0}> at <{1:#x}>".format(exc_code, exc_addr))
self.report_ctx_diff(self.context_exec, ctx)
self.current_process.exit()
return
def report_ctx_diff(self, start, now):
print("==DIFF==")
for name, start_value in start.regs():
now_value = getattr(now, name)
if start_value != now_value:
diff = now_value - start_value
print("{0}: {1:#x} -> {2:#x} ({3:+#x})".format(name, start_value, now_value, diff))
if start.sp > now.sp:
print("Negative Stack: dumping:")
data = self.current_process.read_memory(now.sp, start.sp - now.sp)
print(hexdump(data, start.sp))
def test_code_x86(code, regs=None, raw=False, **kwargs):
print("Testing x86 code")
process = windows.test.pop_proc_32(dwCreationFlags=DEBUG_PROCESS)
if raw:
code = code.replace(" ", "").decode('hex')
else:
code = x86.assemble(code)
start_register = {}
if regs:
for name_value in regs.split(";"):
name, value = name_value.split("=")
name = name.strip().capitalize()
if name == "Eflags":
name = "EFlags"
value = int(value.strip(), 0)
start_register[name] = value
x = CodeTesteur(process, code, start_register)
x.loop()
def test_code_x64(code, regs=None, raw=False, **kwargs):
print("Testing x64 code")
if windows.current_process.bitness == 32:
raise ValueError("Cannot debug a 64b process from 32b python")
process = windows.test.pop_proc_64(dwCreationFlags=DEBUG_PROCESS)
if raw:
code = code.replace(" ", "").decode('hex')
else:
code = x64.assemble(code)
start_register = {}
if regs:
for name_value in regs.split(";"):
name, value = name_value.split("=")
name = name.strip().capitalize()
if name == "Eflags":
name = "EFlags"
value = int(value.strip(), 0)
start_register[name] = value
x = CodeTesteur(process, code, start_register)
x.loop()
parser = argparse.ArgumentParser(prog=__file__)
parser.add_argument('--x64', action='store_const', dest="func", const=test_code_x64, default=test_code_x86, help='Code is x64')
parser.add_argument('--raw', action='store_true', help='argument is raw assembled code (in hex)')
parser.add_argument('code', help='The code to execute')
parser.add_argument('regs', nargs="?", help='The default values of the registers')
res = parser.parse_args()
res.func(**res.__dict__)
Ouput:
(cmd λ) python.exe test_code.py "mov eax, 0x42424242" "eax=0x11223344"
Testing x86 code
Startup context is:
Eip -> 0x3f0000L
Esp -> 0x3bfae4L
Eax -> 0x11223344L
Ebx -> 0x5a6000L
Ecx -> 0x0L
Edx -> 0x0L
Ebp -> 0x0L
Edi -> 0x0L
Esi -> 0x0L
EFlags -> 0x202L
EEflags(0x202L:IF)
==Post-exec context==
Eip -> 0x3f0007L
Esp -> 0x3bfae4L
Eax -> 0x42424242L
Ebx -> 0x5a6000L
Ecx -> 0x0L
Edx -> 0x0L
Ebp -> 0x0L
Edi -> 0x0L
Esi -> 0x0L
EFlags -> 0x202L
EEflags(0x202L:IF)
<Normal terminaison>
==DIFF==
Eip: 0x3f0000 -> 0x3f0007 (+0x7)
Eax: 0x11223344 -> 0x42424242 (+0x31200efe)
(cmd λ) python64 test_code.py --x64 "mov r15, 0x11223344; push r14; call r15" "rcx=1; r14=0x4242424243434343"
Testing x64 code
Startup context is:
Rip -> 0x205a1d60000L
Rsp -> 0xe24a88fa88L
Rax -> 0x0L
Rbx -> 0x0L
Rcx -> 0x1L
Rdx -> 0xe24aaf9000L
Rbp -> 0x0L
Rdi -> 0x0L
Rsi -> 0x0L
R8 -> 0x0L
R9 -> 0x0L
R10 -> 0x0L
R11 -> 0x0L
R12 -> 0x0L
R13 -> 0x0L
R14 -> 0x4242424243434343L
R15 -> 0x0L
EFlags -> 0x200L
EEflags(0x200L:IF)
==Post-exec context==
Rip -> 0x11223344L
Rsp -> 0xe24a88fa78L
Rax -> 0x0L
Rbx -> 0x0L
Rcx -> 0x1L
Rdx -> 0xe24aaf9000L
Rbp -> 0x0L
Rdi -> 0x0L
Rsi -> 0x0L
R8 -> 0x0L
R9 -> 0x0L
R10 -> 0x0L
R11 -> 0x0L
R12 -> 0x0L
R13 -> 0x0L
R14 -> 0x4242424243434343L
R15 -> 0x11223344L
EFlags -> 0x10202L
EEflags(0x10202L:IF|RF)
<EXCEPTION_ACCESS_VIOLATION(0xc0000005L)> at <0x11223344>
==DIFF==
Rip: 0x205a1d60000 -> 0x11223344 (-0x20590b3ccbc)
Rsp: 0xe24a88fa88 -> 0xe24a88fa78 (-0x10)
R15: 0x0 -> 0x11223344 (+0x11223344)
EFlags: 0x200 -> 0x10202 (+0x10002)
Negative Stack: dumping:
E24A88FA88 0C 00 D6 A1 05 02 00 00 43 43 43 43 42 42 42 42 ........CCCCBBBB
16.7.2. LocalDebugger¶
16.7.2.1. In current process¶
import sys
import os.path
import pprint
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
from windows.generated_def.winstructs import *
import windows.native_exec.simple_x86 as x86
class SingleSteppingDebugger(windows.debug.LocalDebugger):
SINGLE_STEP_COUNT = 4
def on_exception(self, exc):
code = self.get_exception_code()
context = self.get_exception_context()
print("EXCEPTION !!!! Got a {0} at 0x{1:x}".format(code, context.pc))
self.SINGLE_STEP_COUNT -= 1
if self.SINGLE_STEP_COUNT:
return self.single_step()
return EXCEPTION_CONTINUE_EXECUTION
class RewriteBreakpoint(windows.debug.HXBreakpoint):
def trigger(self, dbg, exc):
context = dbg.get_exception_context()
print("GOT AN HXBP at 0x{0:x}".format(context.pc))
# Rewrite the infinite loop with 2 nop
windows.current_process.write_memory(self.addr, "\x90\x90")
# Ask for a single stepping
return dbg.single_step()
d = SingleSteppingDebugger()
# Infinite loop + nop + ret
code = x86.assemble("label :begin; jmp :begin; nop; ret")
func = windows.native_exec.create_function(code, [PVOID])
print("Code addr = 0x{0:x}".format(func.code_addr))
# Create a thread that will infinite loop
t = windows.current_process.create_thread(func.code_addr, 0)
# Add a breakpoint on the infitine loop
d.add_bp(RewriteBreakpoint(func.code_addr))
t.wait()
print("Done!")
Ouput:
(cmd λ) python.exe debug\local_debugger.py
Code addr = 0xcf0002
GOT AN HXBP at 0xcf0002
EXCEPTION !!!! Got a EXCEPTION_SINGLE_STEP(0x80000004L) at 0xcf0003
EXCEPTION !!!! Got a EXCEPTION_SINGLE_STEP(0x80000004L) at 0xcf0004
EXCEPTION !!!! Got a EXCEPTION_SINGLE_STEP(0x80000004L) at 0xcf0005
EXCEPTION !!!! Got a EXCEPTION_SINGLE_STEP(0x80000004L) at 0x770d7c04
Done!
16.7.2.2. In remote process¶
import sys
import os.path
import pprint
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import ctypes
import windows
import windows.test
from windows.generated_def.winstructs import *
remote_code = """
import windows
from windows.generated_def.winstructs import *
windows.utils.create_console()
class YOLOHXBP(windows.debug.HXBreakpoint):
def trigger(self, dbg, exc):
p = windows.current_process
arg_pos = 2
context = dbg.get_exception_context()
esp = context.Esp
unicode_string_addr = p.read_ptr(esp + (arg_pos + 1) * 4)
wstring_addr = p.read_ptr(unicode_string_addr + 4)
dll_loaded = p.read_wstring(wstring_addr)
print("I AM LOADING <{0}>".format(dll_loaded))
d = windows.debug.LocalDebugger()
exp = windows.current_process.peb.modules[1].pe.exports
#windows.utils.FixedInteractiveConsole(locals()).interact()
ldr = exp["LdrLoadDll"]
d.add_bp(YOLOHXBP(ldr))
"""
c = windows.test.pop_proc_32(dwCreationFlags=CREATE_SUSPENDED)
c.execute_python(remote_code)
c.threads[0].resume()
import time
time.sleep(2)
c.exit()
Ouput:
(cmd λ) python.exe debug\local_debugger_remote_process.py
(In another console)
I AM LOADING <C:\Windows\system32\uxtheme.dll>
I AM LOADING <C:\Windows\system32\uxtheme.dll>
I AM LOADING <C:\Windows\system32\uxtheme.dll>
I AM LOADING <C:\Windows\system32\uxtheme.dll>
I AM LOADING <kernel32.dll>
I AM LOADING <C:\Windows\WinSxS\x86_microsoft.windows.gdiplus_6595b64144ccf1df_1.1.9600.17415_none_dad8722c5bcc2d8f\gdiplus.dll>
I AM LOADING <comctl32.dll>
I AM LOADING <comctl32.dll>
I AM LOADING <comctl32.dll>
I AM LOADING <comctl32.dll>
I AM LOADING <comctl32.dll>
I AM LOADING <comctl32>
I AM LOADING <C:\Windows\SysWOW64\oleacc.dll>
I AM LOADING <OLEAUT32.DLL>
I AM LOADING <C:\Windows\system32\ole32.dll>
I AM LOADING <C:\Windows\system32\MSCTF.dll>
I AM LOADING <C:\Windows\SysWOW64\msxml6.dll>
I AM LOADING <C:\Windows\system32\shell32.dll>
I AM LOADING <C:\Windows\SYSTEM32\WINMM.dll>
I AM LOADING <C:\Windows\system32\ole32.dll>
16.8. WMI requests¶
import sys
import os.path
import pprint
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
print("WMI requester is {0}".format(windows.system.wmi))
print("Selecting * from 'Win32_Process'")
result = windows.system.wmi.select("Win32_Process")
print("They are <{0}> processes".format(len(result)))
print("Looking for ourself via pid")
us = [p for p in result if int(p["ProcessId"]) == windows.current_process.pid][0]
print("Some info about our process:")
print(" * {0} -> {1}".format("Name", us["Name"]))
print(" * {0} -> {1}".format("ProcessId", us["ProcessId"]))
print(" * {0} -> {1}".format("OSName", us["OSName"]))
print(" * {0} -> {1}".format("UserModeTime", us["UserModeTime"]))
print(" * {0} -> {1}".format("WindowsVersion", us["WindowsVersion"]))
print(" * {0} -> {1}".format("CommandLine", us["CommandLine"]))
print("<Select Caption,FileSystem,FreeSpace from Win32_LogicalDisk>:")
for vol in windows.system.wmi.select("Win32_LogicalDisk", ["Caption", "FileSystem", "FreeSpace"]):
print(" * " + str(vol))
Ouput:
(cmd λ) python wmi\wmi_request.py
WMI requester is <windows.winobject.wmi.WmiRequester object at 0x02B37EF0>
Selecting * from 'Win32_Process'
They are <92> processes
Looking for ourself via pid
Some info about our process:
* Name -> python.exe
* ProcessId -> 7968
* OSName -> Microsoft Windows 8.1 Pro|C:\Windows|\Device\Harddisk0\Partition2
* UserModeTime -> 2812500
* WindowsVersion -> 6.3.9600
* CommandLine -> python.exe .\samples\wmi_request.py
<Select Caption,FileSystem,FreeSpace from Win32_LogicalDisk>:
* {'Caption': u'C:', 'FreeSpace': u'43991547904', 'FileSystem': u'NTFS'}
* {'Caption': u'E:', 'FreeSpace': u'82776027136', 'FileSystem': u'NTFS'}
* {'Caption': u'F:', 'FreeSpace': u'5711265792', 'FileSystem': u'FAT32'}
* {'Caption': u'G:', 'FreeSpace': None, 'FileSystem': None}
16.9. using COM: INetFwPolicy2¶
import sys
import os.path
import pprint
sys.path.append(os.path.abspath(__file__ + "\..\.."))
import windows
import windows.generated_def as gdef
from windows.generated_def import interfaces
# This code is a simple version of the firewall code in windows.winoject.network
print("Initialisation of COM")
windows.com.init()
print("Creating INetFwPolicy2 variable")
firewall = interfaces.INetFwPolicy2()
print("{0} (value = {1})".format(firewall, firewall.value))
print("")
print("Generating CLSID")
NetFwPolicy2CLSID = windows.com.IID.from_string("E2B3C97F-6AE1-41AC-817A-F6F92166D7DD")
print(NetFwPolicy2CLSID)
print("")
print("Creating COM instance")
windows.com.create_instance(NetFwPolicy2CLSID, firewall)
print("{0} (value = 0x{1:0})".format(firewall, firewall.value))
print("")
print("Checking for enabled profiles")
for profile in [gdef.NET_FW_PROFILE2_DOMAIN, gdef.NET_FW_PROFILE2_PRIVATE, gdef.NET_FW_PROFILE2_PUBLIC]:
enabled = gdef.VARIANT_BOOL()
firewall.get_FirewallEnabled(profile, enabled)
print(" * {0} -> {1}".format(profile, enabled.value))
Output:
(cmd λ) python com\com_inetfwpolicy2.py
Initialisation of COM
Creating INetFwPolicy2 variable
<INetFwPolicy2 object at 0x02DC8210> (value = None)
Generating CLSID
<IID "E2B3C97F-6AE1-41AC-817A-F6F92166D7DD">
Creating COM instance
<INetFwPolicy2 object at 0x02DC8210> (value = 0x8984848)
Checking for enabled profiles
* NET_FW_PROFILE2_DOMAIN(0x1L) -> True
* NET_FW_PROFILE2_PRIVATE(0x2L) -> True
* NET_FW_PROFILE2_PUBLIC(0x4L) -> True
16.10. windows.crypto¶
16.10.1. Encryption demo¶
This sample is a working POC able to generate key-pair, encrypt and decrypt file.
import argparse
import windows.crypto as crypto
from windows import winproxy
from windows.generated_def import *
import windows.crypto.generation as gencrypt
# http://stackoverflow.com/questions/1461272/basic-questions-on-microsoft-cryptoapi
def crypt(src, dst, certs, **kwargs):
"""Encrypt the content of 'src' file with the certifacts in 'certs' into 'dst'"""
# Open every certificates in the certs list
certlist = [crypto.CertificateContext.from_file(x) for x in certs]
# Encrypt the content of 'src' with all the public keys(certs)
res = crypto.encrypt(certlist, src.read())
print("Encryption done. Result:")
print(repr(res))
# Write the result in 'dst'
dst.write(res)
dst.close()
src.close()
def decrypt(src, pfxfile, password, **kwargs):
"""Decrypt the content of 'src' with the private key in 'pfxfile'. the 'pfxfile' is open using the 'password'"""
# Open the 'pfx' with the given password
pfx = crypto.import_pfx(pfxfile.read(), password)
# Decrypt the content of the file
decrypted = crypto.decrypt(pfx, src.read())
print(u"Result = <{0}>".format(decrypted))
return decrypted
PFW_TMP_KEY_CONTAINER = "PythonForWindowsTMPContainer"
def genkeys(common_name, pfxpassword, outname, **kwargs):
"""Generate a SHA256/RSA key pair. A self-signed certificate with 'common_name' is stored as 'outname'.cer.
The private key is stored in 'outname'.pfx protected with 'pfxpassword'"""
cert_store = crypto.EHCERTSTORE.new_in_memory()
# Create a TMP context that will hold our newly generated key-pair
with crypto.CryptContext(PFW_TMP_KEY_CONTAINER, None, PROV_RSA_FULL, 0, retrycreate=True) as ctx:
key = HCRYPTKEY()
# Generate a key-pair that is exportable
winproxy.CryptGenKey(ctx, AT_KEYEXCHANGE, CRYPT_EXPORTABLE, key)
# It does NOT destroy the key-pair from the container,
# It only release the key handle
# https://msdn.microsoft.com/en-us/library/windows/desktop/aa379918(v=vs.85).aspx
winproxy.CryptDestroyKey(key)
# Descrption of the key-container that will be used to generate the certificate
KeyProvInfo = CRYPT_KEY_PROV_INFO()
KeyProvInfo.pwszContainerName = PFW_TMP_KEY_CONTAINER
KeyProvInfo.pwszProvName = None
KeyProvInfo.dwProvType = PROV_RSA_FULL
KeyProvInfo.dwFlags = 0
KeyProvInfo.cProvParam = 0
KeyProvInfo.rgProvParam = None
#KeyProvInfo.dwKeySpec = AT_SIGNATURE
KeyProvInfo.dwKeySpec = AT_KEYEXCHANGE
crypt_algo = CRYPT_ALGORITHM_IDENTIFIER()
crypt_algo.pszObjId = szOID_RSA_SHA256RSA
certif_name = "CN={0}".format(common_name)
# Generate a self-signed certificate based on the given key-container and signature algorithme
certif = gencrypt.generate_selfsigned_certificate(certif_name, key_info=KeyProvInfo, signature_algo=crypt_algo)
# Add the newly created certificate to our TMP cert-store
cert_store.add_certificate(certif)
# Generate a pfx from the TMP cert-store
pfx = gencrypt.generate_pfx(cert_store, pfxpassword)
if outname is None:
outname = common_name.lower()
# Dump the certif (public key) and pfx (public + private keys)
with open(outname + ".cer", "wb") as f:
# The encoded certif only contains the public key
f.write(certif.encoded)
with open(outname + ".pfx", "wb") as f:
f.write(pfx)
print(certif)
# Destroy the TMP key container
prov = HCRYPTPROV()
winproxy.CryptAcquireContextW(prov, PFW_TMP_KEY_CONTAINER, None, PROV_RSA_FULL, CRYPT_DELETEKEYSET)
parser = argparse.ArgumentParser(prog=__file__)
subparsers = parser.add_subparsers(description='valid subcommands',)
cryptparse = subparsers.add_parser('crypt')
cryptparse.set_defaults(func=crypt)
cryptparse.add_argument('src', type=argparse.FileType('rb'), help='File to encrypt')
cryptparse.add_argument('dst', type=argparse.FileType('wb'), help='The encrypted file')
cryptparse.add_argument('certs', type=str, nargs='+',
help='List of certfile used to encrypt the src')
decryptparse = subparsers.add_parser('decrypt')
decryptparse.set_defaults(func=decrypt)
decryptparse.add_argument('src', type=argparse.FileType('rb'), help='File to decrypt')
decryptparse.add_argument('pfxfile', type=argparse.FileType('rb'), help='PFX file to use')
decryptparse.add_argument('password', help='Password of the PFX')
genkeysparse = subparsers.add_parser('genkey')
genkeysparse.set_defaults(func=genkeys)
genkeysparse.add_argument('common_name', nargs='?', metavar='CommonName', default='DEFAULT', help='the common name of the certificate')
genkeysparse.add_argument('outname', nargs='?',help='The filename base for the generated files')
genkeysparse.add_argument('--pfxpassword', nargs='?', help='Password to protect the PFX')
res = parser.parse_args()
res.func(**res.__dict__)
Ouput:
(cmd λ) python crypto\encryption_demo.py genkey YOLOCERTIF mykey --pfxpassword MYPASSWORD
<CertificatContext "YOLOCERTIF" serial="1b a4 3e 17 f7 ed ec ab 4f f8 11 46 48 e9 29 25">
(cmd λ) ls
mykey.cer mykey.pfx
(cmd λ) echo|set /p="my secret message" > message.txt
(cmd λ) python crypto\encryption_demo.py crypt message.txt message.crypt mykey.cer
Encryption done. Result:
bytearray(b'0\x82\x01\x19\x06\t*\x86H\x86\xf7\r\x01\x07\x03\xa0\x82\x01\n0\x82\x01\x06\x02\x01\x001\x81\xc30\x81
\xc0\x02\x01\x000)0\x151\x130\x11\x06\x03U\x04\x03\x13\nYOLOCERTIF\x02\x10\x1b\xa4>\x17\xf7\xed\xec\xabO\xf8\x11
FH\xe9)%0\r\x06\t*\x86H\x86\xf7\r\x01\x01\x01\x05\x00\x04\x81\x80V\x89)\xf5\xaaM\x99cEA\x17^\xa2D~\x94\xe3\xf2\x1f
\x05Y\xc2\xbb\xb2\xbbYBpU6\x870\xce\xe7\xd2M{\xbb\xb9K\xa0\xf5\xe5\x93\xca\xedF\x80.x\xdc\xf2\x0c\xa6UO\x01\r\xaf
\xd0Z\xd9\xabnzR\xd4j=\xca\xc2RG\xcd\x11u\x82\x7f\x8c\xd8t\xb9\xf9\xe8%\xfal\xaaHPj;\xecKk]\t%\xfd\x91\xcc\xe0lWf
\xc6\x12x\x1am\xc8\x01t\xac\xa6\xf3#\x02\xd4J \x8eZ\xbb\x10W\xe1 0;\x06\t*\x86H\x86\xf7\r\x01\x07\x010\x14\x06\x08*
\x86H\x86\xf7\r\x03\x07\x04\x08\x14F\x04\xad\xed9\xed<\x80\x18\x80]6\xccTV\xbc\xb8*\x84QY!~\xb3\n\x1aV\xd4\rf\xd1n:')
(cmd λ) python crypto\encryption_demo.py decrypt message.crypt mykey.pfx BADPASS
Traceback (most recent call last):
File "..\samples\encryption_demo.py", line 103, in <module>
res.func(**res.__dict__)
File "..\samples\encryption_demo.py", line 26, in decrypt
pfx = crypto.import_pfx(pfxfile.read(), password)
File "c:\users\hakril\documents\work\pythonforwindows\windows\crypto\certificate.py", line 153, in import_pfx
cert_store = winproxy.PFXImportCertStore(pfx, password, flags)
File "c:\users\hakril\documents\work\pythonforwindows\windows\winproxy.py", line 1065, in PFXImportCertStore
return PFXImportCertStore.ctypes_function(pPFX, szPassword, dwFlags)
File "c:\users\hakril\documents\work\pythonforwindows\windows\winproxy.py", line 148, in perform_call
return self._cprototyped(*args)
File "c:\users\hakril\documents\work\pythonforwindows\windows\winproxy.py", line 69, in kernel32_error_check
raise Kernel32Error(func_name)
windows.winproxy.Kernel32Error: PFXImportCertStore: [Error 86] The specified network password is not correct.
(cmd λ) python crypto\encryption_demo.py decrypt message.crypt mykey.pfx MYPASSWORD
Result = <my secret message>
16.10.2. Certificate demo¶
import hashlib
import windows.crypto
windowscert = """-----BEGIN CERTIFICATE-----
MIIFBDCCA+ygAwIBAgITMwAAAQZuwyXEMckYDgAAAAABBjANBgkqhkiG9w0BAQsF
ADCBhDELMAkGA1UEBhMCVVMxEzARBgNVBAgTCldhc2hpbmd0b24xEDAOBgNVBAcT
B1JlZG1vbmQxHjAcBgNVBAoTFU1pY3Jvc29mdCBDb3Jwb3JhdGlvbjEuMCwGA1UE
AxMlTWljcm9zb2Z0IFdpbmRvd3MgUHJvZHVjdGlvbiBQQ0EgMjAxMTAeFw0xNjEw
MTEyMDM5MzFaFw0xODAxMTEyMDM5MzFaMHAxCzAJBgNVBAYTAlVTMRMwEQYDVQQI
EwpXYXNoaW5ndG9uMRAwDgYDVQQHEwdSZWRtb25kMR4wHAYDVQQKExVNaWNyb3Nv
ZnQgQ29ycG9yYXRpb24xGjAYBgNVBAMTEU1pY3Jvc29mdCBXaW5kb3dzMIIBIjAN
BgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAyWcaCYghNInk3ecpyu2uZ7LCV9QS
7GWYr41ufTkcL66ewHxlAoWjmkKG6W2Bp9BYYQok10iDeDGACE9Vjr6m4Jdh+YuN
RLxMnHC8JTGzk96CzmdBPAuUWdAcHNmTkIWQF6AXzsbBWsekQejvDBygAOCuIYh4
sBgNa5cjTxQc7Iyp9c7RxBmThV5BNFTOnSN6D9N8zU+ENgIZuyHxGvqzRdrhU4G4
Cg/h1CkI4TgeZQZCeUNPnWV6DMuvPCiqGEia5phOJZyENKND0Sx6eQZrYnuz1gMn
YaEnO+ggegtt4pWpqg8Ch0jNrkL1fb3Kzz7E34/K9dcTgaOymfF6qUKabQIDAQAB
o4IBgDCCAXwwHwYDVR0lBBgwFgYKKwYBBAGCNwoDBgYIKwYBBQUHAwMwHQYDVR0O
BBYEFBEciVg/vsVmKtr/hmHt7KM6g8lSMFIGA1UdEQRLMEmkRzBFMQ0wCwYDVQQL
EwRNT1BSMTQwMgYDVQQFEysyMjk4NzkrMTQ3NDQ5YmUtMTVhOC00ZWJhLTkzZjMt
ZDExMGE1YzQ1NTUyMB8GA1UdIwQYMBaAFKkpAjmOFsSXeM2Q+Z5PmuF8Va9TMFQG
A1UdHwRNMEswSaBHoEWGQ2h0dHA6Ly93d3cubWljcm9zb2Z0LmNvbS9wa2lvcHMv
Y3JsL01pY1dpblByb1BDQTIwMTFfMjAxMS0xMC0xOS5jcmwwYQYIKwYBBQUHAQEE
VTBTMFEGCCsGAQUFBzAChkVodHRwOi8vd3d3Lm1pY3Jvc29mdC5jb20vcGtpb3Bz
L2NlcnRzL01pY1dpblByb1BDQTIwMTFfMjAxMS0xMC0xOS5jcnQwDAYDVR0TAQH/
BAIwADANBgkqhkiG9w0BAQsFAAOCAQEAvYC1iawgKoxXAotQXaN0lj1J5VX01/un
7JybZF4sPMG4acoFT85Ao5U6TK5ATPB7yPUulAivp8908DwTGqN+Ju6iH+UkvAb+
a/WcHVEMxQXK5eOFNE6yekUArBGbMNWlTFrpwklmVTnL9R+4aApTEe6ITT1KLDio
5uFw98n5Sqgh+In073czyiTG7MVhBexbOfhgnciXoufeyhwy1pYgjouSqSQZs4bj
cUwQTwGlS2Gd5a+3nblhjn+QhSszIo1K5n1udLPFWtn29BuGlSrtTXPv5OCfNtLO
l2ec6CyjDQc6HcQBNCsbJVq6qGtQbYNE+ih+KhIU4tO5jf25xthf2g==
-----END CERTIFICATE-----"""
raw_cert = ("".join(windowscert.split("\n")[1:-1])).decode('base64')
cert = windows.crypto.CertificateContext.from_buffer(raw_cert)
print("Analysing certificate: {0}".format(cert))
print(" * name: <{0}>".format(cert.name))
print(" * issuer: <{0}>".format(cert.issuer))
print(" * raw_serial: <{0}>".format(cert.raw_serial))
print(" * serial: <{0}>".format(cert.serial))
print(" * encoded start: <{0!r}>".format(cert.encoded[:20]))
print ""
chains = cert.chains
print("This certificate has {0} certificate chain(s)".format(len(chains)))
for i, chain in enumerate(chains):
print("Chain {0}:".format(i))
for ccert in chain:
print(" {0}:".format(ccert))
print(" * issuer: <{0}>".format(ccert.issuer))
print ""
cert_to_verif = ccert
print("Looking for <{0}> in trusted certificates".format(cert_to_verif.name))
root_store = windows.crypto.EHCERTSTORE.from_system_store("Root")
# This is not the correct way verify the validity of a certificate chain.
# I would say that if the goal is to verify the signature of the certificate: use wintrust.
# (or maybe CertVerifyCertificateChainPolicy : https://msdn.microsoft.com/en-us/library/windows/desktop/aa377163(v=vs.85).aspx)
matchs = [c for c in root_store.certs if c == cert_to_verif]
print("matches = {0}".format(matchs))
if matchs:
print("Found it !")
else:
print("Not found :(")
## Extract certificates of a PE file
## This code is not a fixed API and the current state of my tests
print ("")
print ("== PE Analysis ==")
TARGET_FILE = r"C:\windows\system32\ntdll.dll"
print("Target sha1 = <{0}>".format(hashlib.sha1(open(TARGET_FILE, "rb").read()).hexdigest()))
cryptobj = windows.crypto.CryptObject(TARGET_FILE)
print("Analysing {0}".format(cryptobj))
print("File has {0} signer(s):".format(cryptobj.crypt_msg.nb_signer))
for i, signer in enumerate(cryptobj.crypt_msg.signers):
print("Signer {0}:".format(i))
print(" * Issuer: {0!r}".format(windows.crypto.ECRYPT_DATA_BLOB(signer.Issuer.cbData, signer.Issuer.pbData).data))
print(" * HashAlgorithme: {0}".format(signer.HashAlgorithm.pszObjId))
cert = cryptobj.cert_store.find(signer.Issuer, signer.SerialNumber)
print(" * Certificate: {0}".format(cert))
print("")
print("File embdeds {0} certificate(s):".format(cryptobj.crypt_msg.nb_cert))
for i, certificate in enumerate(cryptobj.crypt_msg.certs):
print(" * {0}) {1}".format(i, certificate))
Ouput:
(cmd λ) python crypto\certificate.py
Analysing certificate: <CertificateContext "Microsoft Windows" serial="33 00 00 01 06 6e c3 25 c4 31 c9 18 0e 00 00 00 00 01 06">
* name: <Microsoft Windows>
* issuer: <Microsoft Windows Production PCA 2011>
* raw_serial: <[51, 0, 0, 1, 6, 110, 195, 37, 196, 49, 201, 24, 14, 0, 0, 0, 0, 1, 6]>
* serial: <33 00 00 01 06 6e c3 25 c4 31 c9 18 0e 00 00 00 00 01 06>
* encoded start: <bytearray(b'0\x82\x05\x040\x82\x03\xec\xa0\x03\x02\x01\x02\x02\x133\x00\x00\x01\x06')>
This certificate has 1 certificate chain(s)
Chain 0:
<CertificateContext "Microsoft Windows" serial="33 00 00 01 06 6e c3 25 c4 31 c9 18 0e 00 00 00 00 01 06">:
* issuer: <Microsoft Windows Production PCA 2011>
<CertificateContext "Microsoft Windows Production PCA 2011" serial="61 07 76 56 00 00 00 00 00 08">:
* issuer: <Microsoft Root Certificate Authority 2010>
<CertificateContext "Microsoft Root Certificate Authority 2010" serial="28 cc 3a 25 bf ba 44 ac 44 9a 9b 58 6b 43 39 aa">:
* issuer: <Microsoft Root Certificate Authority 2010>
Looking for <Microsoft Root Certificate Authority 2010> in trusted certificates
matches = [<CertificateContext "Microsoft Root Certificate Authority 2010" serial="28 cc 3a 25 bf ba 44 ac 44 9a 9b 58 6b 43 39 aa">]
Found it !
== PE Analysis ==
Target sha1 = <eb90bc0e33f3e62b0eac4afa8bfcf42a5d4e7bbb>
Analysing <CryptObject "C:\windows\system32\ntdll.dll" content_type=CERT_QUERY_CONTENT_PKCS7_SIGNED_EMBED(0xaL)>
File has 1 signer(s):
Signer 0:
* Issuer: bytearray(b'0\x81\x841\x0b0\t\x06\x03U\x04\x06\x13\x02US1\x130\x11\x06\x03U\x04\x08\x13\nWashington1\x100\x0e\x06\x03U\x04\x07\x13\x07Redmond1\x1e0\x1c\x06\x03U\x04\n\x13\x15Microsoft Corporation1.0,\x06\x03U\x04\x03\x13%Microsoft Windows Production PCA 2011')
* HashAlgorithme: 2.16.840.1.101.3.4.2.1
* Certificate: <CertificateContext "Microsoft Windows" serial="33 00 00 01 06 6e c3 25 c4 31 c9 18 0e 00 00 00 00 01 06">
File embdeds 2 certificate(s):
* 0) <CertificateContext "Microsoft Windows" serial="33 00 00 01 06 6e c3 25 c4 31 c9 18 0e 00 00 00 00 01 06">
* 1) <CertificateContext "Microsoft Windows Production PCA 2011" serial="61 07 76 56 00 00 00 00 00 08">
16.11. windows.alpc¶
16.11.1. simple alpc communication¶
import multiprocessing
import windows.alpc
from windows.generated_def import LPC_CONNECTION_REQUEST, LPC_REQUEST
PORT_NAME = r"\RPC Control\PythonForWindowsPORT"
def alpc_server():
server = windows.alpc.AlpcServer(PORT_NAME) # Create the ALPC Port
print("[SERV] PORT <{0}> CREATED".format(PORT_NAME))
msg = server.recv() # Wait for a message
print("[SERV] Message type = {0:#x}".format(msg.type))
print("[SERV] Received data: <{0}>".format(msg.data))
assert msg.type & 0xfff == LPC_CONNECTION_REQUEST # Check that message is a connection request
print("[SERV] Connection request")
server.accept_connection(msg)
msg = server.recv() # Wait for a real message
print ""
print("[SERV] Received message: <{0}>".format(msg.data))
print("[SERV] Message type = {0:#x}".format(msg.type))
assert msg.type & 0xfff == LPC_REQUEST
# We can reply by two ways:
# - Send the same message with modified data
# - Recreate a Message and copy the MessageId
msg.data = "REQUEST '{0}' DONE".format(msg.data)
server.send(msg)
def alpc_client():
print("Client pid = {0}".format(windows.current_process.pid))
# Creation an 'AlpcClient' with a port name will connect to the port with an empty message
client = windows.alpc.AlpcClient(PORT_NAME)
print("[CLIENT] Connected: {0}".format(client))
# Send a message / wait for the response
response = client.send_receive("Hello world !")
print("[CLIENT] Response: <{0}>".format(response.data))
# You can also send message without waiting for a response with 'client.send'
if __name__ == "__main__":
proc = multiprocessing.Process(target=alpc_server, args=())
proc.start()
import time; time.sleep(0.5)
alpc_client()
print("BYE")
proc.terminate()
Ouput:
(cmd λ) python alpc\simple_alpc.py
[SERV] PORT <\RPC Control\PythonForWindowsPORT> CREATED
Client pid = 15044
[SERV] Message type = 0x300a
[SERV] Received data: <>
[SERV] Connection request
[CLIENT] Connected: <windows.alpc.AlpcClient object at 0x0377FDB0>
[SERV] Received message: <Hello world !>
[SERV] Message type = 0x3001
[CLIENT] Response: <REQUEST 'Hello world !' DONE>
BYE
16.11.2. advanced alpc communication¶
import multiprocessing
import windows.alpc
from windows.generated_def import LPC_CONNECTION_REQUEST, LPC_REQUEST
import windows.generated_def as gdef
import ctypes
import tempfile
PORT_NAME = r"\RPC Control\PythonForWindowsPORT_2"
PORT_CONTEXT = 0x11223344
def full_alpc_server():
print("server pid = {0}".format(windows.current_process.pid))
server = windows.alpc.AlpcServer(PORT_NAME)
print("[SERV] PORT <{0}> CREATED".format(PORT_NAME))
msg = server.recv()
print("[SERV] == Message received ==")
if msg.type & 0xfff == LPC_CONNECTION_REQUEST:
print(" * ALPC connection request: <{0}>".format(msg.data))
msg.data = "Connection message response"
server.accept_connection(msg, port_context=PORT_CONTEXT)
else:
raise ValueError("Expected connection")
while True:
msg = server.recv()
print("[SERV] == Message received ==")
# print(" * Data: {0}".format(msg.data))
# print("[SERV] RECV Message type = {0:#x}".format(msg.type))
# print("[SERV] RECV Message Valid ATTRS = {0:#x}".format(msg.attributes.ValidAttributes))
# print("[SERV] RECV Message ATTRS = {0:#x}".format(msg.attributes.AllocatedAttributes))
if msg.type & 0xfff == LPC_REQUEST:
print(" * ALPC request: <{0}>".format(msg.data))
print(" * view_is_valid <{0}>".format(msg.view_is_valid))
if msg.view_is_valid:
print(" * message view attribute:")
windows.utils.print_ctypes_struct(msg.view_attribute, " - VIEW", hexa=True)
view_data = windows.current_process.read_string(msg.view_attribute.ViewBase)
print(" * Reading view content: <{0}>".format(view_data))
print(" * security_is_valid <{0}>".format(msg.security_is_valid))
print(" * handle_is_valid <{0}>".format(msg.handle_is_valid))
if msg.handle_is_valid:
if msg.handle_attribute.Handle:
print(" * message handle attribute:")
windows.utils.print_ctypes_struct(msg.handle_attribute, " - HANDLE", hexa=True)
if msg.handle_attribute.ObjectType == 1:
f = windows.utils.create_file_from_handle(msg.handle_attribute.Handle)
print(" - File: {0}".format(f))
print(" - content: <{0}>".format(f.read()))
else:
print(" - unknow object type == {0}".format(msg.handle_attribute.ObjectType))
msg.attributes.ValidAttributes -= gdef.ALPC_MESSAGE_HANDLE_ATTRIBUTE
print(" * context_is_valid <{0}>".format(msg.context_is_valid))
if msg.context_is_valid:
print(" * message context attribute:")
windows.utils.print_ctypes_struct(msg.context_attribute, " - CTX", hexa=True)
if msg.attributes.ValidAttributes & gdef.ALPC_MESSAGE_TOKEN_ATTRIBUTE:
print(" * message token attribute:")
token_struct = msg.attributes.get_attribute(gdef.ALPC_MESSAGE_TOKEN_ATTRIBUTE)
windows.utils.print_ctypes_struct(token_struct, " - TOKEN", hexa=True)
# We can reply by to way:
# - Send the same message with modified data
# - Recreate a Message and copy the MessageId
msg.data = "REQUEST '{0}' DONE".format(msg.data)
server.send(msg)
else:
raise ValueError("Unexpected message type")
def send_message_with_handle(client):
print ""
print("[Client] == Sending a message with a handle ==")
# Craft a file with some data
f = tempfile.NamedTemporaryFile()
f.write("Tempfile data <3")
f.seek(0)
# New message with a Handle
msg = windows.alpc.AlpcMessage()
msg.attributes.ValidAttributes |= gdef.ALPC_MESSAGE_HANDLE_ATTRIBUTE
msg.handle_attribute.Flags = gdef.ALPC_HANDLEFLG_DUPLICATE_SAME_ACCESS
msg.handle_attribute.Handle = windows.utils.get_handle_from_file(f)
msg.handle_attribute.ObjectType = 0
msg.handle_attribute.DesiredAccess = 0
msg.data = "some message with a file"
client.send_receive(msg)
def send_message_with_view(client):
print ""
print("[Client] == Sending a message with a view ==")
# Create View
section = client.create_port_section(0, 0, 0x4000)
view = client.map_section(section[0], 0x4000)
# New message with a View
msg = windows.alpc.AlpcMessage(0x2000)
msg.attributes.ValidAttributes |= gdef.ALPC_MESSAGE_VIEW_ATTRIBUTE
msg.view_attribute.Flags = 0
msg.view_attribute.ViewBase = view.ViewBase
msg.view_attribute.SectionHandle = view.SectionHandle
msg.view_attribute.ViewSize = 0x4000
msg.data = "some message with a view"
windows.current_process.write_memory(view.ViewBase, "The content of the view :)\x00")
client.send_receive(msg)
def alpc_client():
print("Client pid = {0}".format(windows.current_process.pid))
client = windows.alpc.AlpcClient()
# You can create a non-connected AlpcClient and send a custom
# 'AlpcMessage' for complexe alpc port connection.
connect_message = windows.alpc.AlpcMessage()
connect_message.data = "Connection request client message"
print("[CLIENT] == Connecting to port ==")
connect_response = client.connect_to_port(PORT_NAME, connect_message)
print("[CLIENT] Connected with response: <{0}>".format(connect_response.data))
# AlpcClient send/recv/send_receive methods accept both string or
# AlpcMessage for complexe message.
print""
print("[CLIENT] == Sending a message ==")
msg = windows.alpc.AlpcMessage()
msg.data = "Complex Message 1"
print(" * Sending Message <{0}>".format(msg.data))
response = client.send_receive(msg)
print("[CLIENT] Server response: <{0}>".format(response.data))
print("[CLIENT] RESP Message Valid ATTRS = {0}".format(response.valid_attributes))
send_message_with_handle(client)
send_message_with_view(client)
if __name__ == "__main__":
proc = multiprocessing.Process(target=full_alpc_server, args=())
proc.start()
import time; time.sleep(0.5)
alpc_client()
print("BYE")
proc.terminate()
Output:
(cmd λ) python alpc\advanced_alpc.py
server pid = 13000
[SERV] PORT <\RPC Control\PythonForWindowsPORT_2> CREATED
Client pid = 2100
[CLIENT] == Connecting to port ==
[SERV] == Message received ==
* ALPC connection request: <Connection request client message>
[CLIENT] Connected with response: <Connection message response>
[CLIENT] == Sending a message ==
* Sending Message <Complex Message 1>
[SERV] == Message received ==
* ALPC request: <Complex Message 1>
* view_is_valid <False>
* security_is_valid <False>
* handle_is_valid <False>
* context_is_valid <True>
* message context attribute:
- CTX.PortContext -> 0x11223344
- CTX.MessageContext -> None
- CTX.Sequence -> 0x1L
- CTX.MessageId -> 0x0L
- CTX.CallbackId -> 0x0L
* message token attribute:
- TOKEN.TokenId -> 0x1e4ecaccL
- TOKEN.AuthenticationId -> 0x48989L
- TOKEN.ModifiedId -> 0x48995L
[CLIENT] Server response: <REQUEST 'Complex Message 1' DONE>
[CLIENT] RESP Message Valid ATTRS = [ALPC_MESSAGE_CONTEXT_ATTRIBUTE(0x20000000L)]
[Client] == Sending a message with a handle ==
[SERV] == Message received ==
* ALPC request: <some message with a file>
* view_is_valid <False>
* security_is_valid <False>
* handle_is_valid <True>
* message handle attribute:
- HANDLE.Flags -> 0x0L
- HANDLE.Handle -> 0x260
- HANDLE.ObjectType -> 0x1L
- HANDLE.DesiredAccess -> 0x13019fL
- File: <open file '<fdopen>', mode 'r' at 0x02D529C0>
- content: <Tempfile data <3>
* context_is_valid <True>
* message context attribute:
- CTX.PortContext -> 0x11223344
- CTX.MessageContext -> None
- CTX.Sequence -> 0x2L
- CTX.MessageId -> 0x0L
- CTX.CallbackId -> 0x0L
* message token attribute:
- TOKEN.TokenId -> 0x1e4ecaccL
- TOKEN.AuthenticationId -> 0x48989L
- TOKEN.ModifiedId -> 0x48995L
[Client] == Sending a message with a view ==
[SERV] == Message received ==
* ALPC request: <some message with a view>
* view_is_valid <True>
* message view attribute:
- VIEW.Flags -> 0x0L
- VIEW.SectionHandle -> None
- VIEW.ViewBase -> 0x2770000
- VIEW.ViewSize -> 0x4000
* Reading view content: <The content of the view :)>
* security_is_valid <False>
* handle_is_valid <False>
* context_is_valid <True>
* message context attribute:
- CTX.PortContext -> 0x11223344
- CTX.MessageContext -> None
- CTX.Sequence -> 0x3L
- CTX.MessageId -> 0x0L
- CTX.CallbackId -> 0x0L
* message token attribute:
- TOKEN.TokenId -> 0x1e4ecaccL
- TOKEN.AuthenticationId -> 0x48989L
- TOKEN.ModifiedId -> 0x48995L
BYE
16.12. windows.rpc¶
16.12.1. Manual UAC¶
import argparse
import sys
import windows.rpc
import windows.generated_def as gdef
from windows.rpc import ndr
# NDR Descriptions
class NDRPoint(ndr.NdrStructure):
MEMBERS = [ndr.NdrLong, ndr.NdrLong]
class NdrUACStartupInfo(ndr.NdrStructure):
MEMBERS = [ndr.NdrUniquePTR(ndr.NdrWString),
ndr.NdrLong,
ndr.NdrLong,
ndr.NdrLong,
ndr.NdrLong,
ndr.NdrLong,
ndr.NdrLong,
ndr.NdrLong,
ndr.NdrLong,
ndr.NdrLong,
NDRPoint]
class RAiLaunchAdminProcessParameters(ndr.NdrParameters):
MEMBERS = [ndr.NdrUniquePTR(ndr.NdrWString),
ndr.NdrUniquePTR(ndr.NdrWString),
ndr.NdrLong,
ndr.NdrLong,
ndr.NdrWString,
ndr.NdrWString,
NdrUACStartupInfo,
ndr.NdrLong,
ndr.NdrLong]
class NdrProcessInformation(ndr.NdrParameters):
MEMBERS = [ndr.NdrLong] * 4
# Parsing args
parser = argparse.ArgumentParser(prog=__file__)
parser.add_argument('--target', default=sys.executable, help='Executable to launch')
parser.add_argument('--cmdline', default="", help='The commandline for the process')
parser.add_argument('--uacflags', type=lambda x: int(x, 0), default=0x11)
parser.add_argument('--creationflags', type=lambda x: int(x, 0), default=gdef.CREATE_UNICODE_ENVIRONMENT)
params = parser.parse_args()
print(params)
# Connecting to RPC Interface.
UAC_UIID = "201ef99a-7fa0-444c-9399-19ba84f12a1a"
client = windows.rpc.find_alpc_endpoint_and_connect(UAC_UIID)
iid = client.bind(UAC_UIID)
# Marshalling parameters.
parameters = RAiLaunchAdminProcessParameters.pack([
params.target + "\x00", # Application Path
params.cmdline + "\x00", # Commandline
params.uacflags, # UAC-Request Flag
params.creationflags, # dwCreationFlags
"\x00", # StartDirectory
"WinSta0\\Default\x00", # Station
# Startup Info
(None, # Title
0, # dwX
0, # dwY
0, # dwXSize
0, # dwYSize
0, # dwXCountChars
0, # dwYCountChars
0, # dwFillAttribute
0, # dwFlags
5, # wShowWindow
# Point structure: Use MonitorFromPoint to setup StartupInfo.hStdOutput
(0, 0)),
0, # Window-Handle to know if UAC can steal focus
0xffffffff]) # UAC Timeout
result = client.call(iid, 0, parameters)
stream = ndr.NdrStream(result)
ph, th, pid, tid = NdrProcessInformation.unpack(stream)
return_value = ndr.NdrLong.unpack(stream)
print("Return value = {0:#x}".format(return_value))
target = windows.winobject.process.WinProcess(handle=ph)
print("Created process is {0}".format(target))
print(" * bitness is {0}".format(target.bitness))
print(" * integrity: {0}".format(target.token.integrity))
print(" * elevated: {0}".format(target.token.is_elevated))
Output:
(cmd λ) python rpc\uac.py
Namespace(cmdline='', creationflags=CREATE_UNICODE_ENVIRONMENT(0x400L), target='C:\\Python27\\python.exe', uacflags=17)
# UAC pop - asking to execute python.exe | Clicking Yes
Return value = 0x6
Created process is <WinProcess "python.exe" pid 19304 at 0x455f7d0>
* bitness is 32
* integrity: SECURITY_MANDATORY_HIGH_RID(0x3000L)
* elevated: True
# The new python.exe in another window
>>> windows.current_process.token.integrity
SECURITY_MANDATORY_HIGH_RID(0x3000L)
>>> windows.current_process.token.is_elevated
True
16.12.2. Manual LsarEnumeratePrivileges¶
import windows.rpc
from windows.rpc import ndr
# Ndr stuff
class NdrContext(ndr.NdrStructure):
MEMBERS = [ndr.NdrLong, ndr.NdrLong, ndr.NdrLong, ndr.NdrLong, ndr.NdrLong]
class PLSAPR_OBJECT_ATTRIBUTES(ndr.NdrStructure):
MEMBERS = [ndr.NdrLong,
ndr.NdrUniquePTR(ndr.NdrWString),
ndr.NdrUniquePTR(ndr.NdrLong), # We dont care if the subtype as we will pass None
ndr.NdrLong,
ndr.NdrUniquePTR(ndr.NdrLong), # We dont care if the subtype as we will pass None
ndr.NdrUniquePTR(ndr.NdrLong)] # We dont care if the subtype as we will pass None
class LsarOpenPolicy2Parameter(ndr.NdrParameters):
MEMBERS = [ndr.NdrUniquePTR(ndr.NdrWString),
PLSAPR_OBJECT_ATTRIBUTES,
ndr.NdrLong]
class LsarEnumeratePrivilegesParameter(ndr.NdrParameters):
MEMBERS = [NdrContext,
ndr.NdrLong,
ndr.NdrLong]
class LSAPR_POLICY_PRIVILEGE_DEF(object):
@classmethod
def unpack(cls, stream):
size1 = ndr.NdrShort.unpack(stream)
ptr = ndr.NdrShort.unpack(stream)
size2 = ndr.NdrLong.unpack(stream)
luid = ndr.NdrHyper.unpack(stream)
return ptr, luid
class LSAPR_PRIVILEGE_ENUM_BUFFER(object):
@classmethod
def unpack(cls, stream):
entries = ndr.NdrLong.unpack(stream)
array_size = ndr.NdrLong.unpack(stream)
array_ptr = ndr.NdrLong.unpack(stream)
# Unpack pointed array
array_size2 = ndr.NdrLong.unpack(stream)
assert array_size == array_size2
x = []
# unpack each elements LSAPR_POLICY_PRIVILEGE_DEF
for i in range(array_size2):
ptr, luid = LSAPR_POLICY_PRIVILEGE_DEF.unpack(stream)
if ptr:
x.append(luid)
# unpack pointed strings
result = []
for luid in x:
name = ndr.NdrWcharConformantVaryingArrays.unpack(stream)
result.append((luid, name))
return result
# Actual code
## LSASS alpc endpoints is fixed, no need for the epmapper
client = windows.rpc.RPCClient(r"\RPC Control\lsasspirpc")
## Bind to the desired interface
iid = client.bind('12345778-1234-abcd-ef00-0123456789ab', version=(0,0))
## Craft parameters and call 'LsarOpenPolicy2'
params = LsarOpenPolicy2Parameter.pack([None, (0, None, None, 0, None, None), 0x20000000])
res = client.call(iid, 44, params)
## Unpack the resulting handle
handle = NdrContext.unpack(ndr.NdrStream(res))
## Craft parameters and call 'LsarEnumeratePrivileges'
x = LsarEnumeratePrivilegesParameter.pack([handle, 0, 10000]);
res = client.call(iid, 2, x)
## Unpack the resulting 'LSAPR_PRIVILEGE_ENUM_BUFFER'
priviledges = LSAPR_PRIVILEGE_ENUM_BUFFER.unpack(ndr.NdrStream(res))
for priv in priviledges:
print priv
Output:
(cmd λ) python rpc\lsass.py
(2, u'SeCreateTokenPrivilege')
(3, u'SeAssignPrimaryTokenPrivilege')
(4, u'SeLockMemoryPrivilege')
(5, u'SeIncreaseQuotaPrivilege')
(6, u'SeMachineAccountPrivilege')
(7, u'SeTcbPrivilege')
(8, u'SeSecurityPrivilege')
(9, u'SeTakeOwnershipPrivilege')
(10, u'SeLoadDriverPrivilege')
(11, u'SeSystemProfilePrivilege')
(12, u'SeSystemtimePrivilege')
(13, u'SeProfileSingleProcessPrivilege')
(14, u'SeIncreaseBasePriorityPrivilege')
(15, u'SeCreatePagefilePrivilege')
(16, u'SeCreatePermanentPrivilege')
(17, u'SeBackupPrivilege')
(18, u'SeRestorePrivilege')
(19, u'SeShutdownPrivilege')
(20, u'SeDebugPrivilege')
(21, u'SeAuditPrivilege')
(22, u'SeSystemEnvironmentPrivilege')
(23, u'SeChangeNotifyPrivilege')
(24, u'SeRemoteShutdownPrivilege')
(25, u'SeUndockPrivilege')
(26, u'SeSyncAgentPrivilege')
(27, u'SeEnableDelegationPrivilege')
(28, u'SeManageVolumePrivilege')
(29, u'SeImpersonatePrivilege')
(30, u'SeCreateGlobalPrivilege')
(31, u'SeTrustedCredManAccessPrivilege')
(32, u'SeRelabelPrivilege')
(33, u'SeIncreaseWorkingSetPrivilege')
(34, u'SeTimeZonePrivilege')
(35, u'SeCreateSymbolicLinkPrivilege')
(36, u'SeDelegateSessionUserImpersonatePrivilege')