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

Add a new FlagMapper to easily transform int to Flag and change to code use it

Browse Source
This commit is contained in:
Clement Rouault
2017-11-10 13:47:27 +01:00
parent 4efae2194d
commit 8ba325afd2
13 changed files with 72 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ctypes_generation/extended_structs/_MEMORY_BASIC_INFORMATION32.py
+7 -7
View File
@@ -1,30 +1,30 @@
INITIAL_MEMORY_BASIC_INFORMATION32 = _MEMORY_BASIC_INFORMATION32
class _MEMORY_BASIC_INFORMATION32(INITIAL_MEMORY_BASIC_INFORMATION32):
STATE_MAPPER = {MEM_COMMIT: MEM_COMMIT, MEM_FREE: MEM_FREE, MEM_RESERVE:MEM_RESERVE}
TYPE_MAPPER = {MEM_IMAGE: MEM_IMAGE, MEM_MAPPED: MEM_MAPPED, MEM_PRIVATE:MEM_PRIVATE}
PROTECT_MAPPER = {x:x for x in [PAGE_NOACCESS, PAGE_READONLY, PAGE_READWRITE,
STATE_MAPPER = FlagMapper(MEM_COMMIT, MEM_FREE, MEM_RESERVE)
TYPE_MAPPER = FlagMapper(MEM_IMAGE, MEM_MAPPED, MEM_PRIVATE)
PROTECT_MAPPER = FlagMapper(PAGE_NOACCESS, PAGE_READONLY, PAGE_READWRITE,
PAGE_WRITECOPY, PAGE_EXECUTE, PAGE_EXECUTE_READ, PAGE_EXECUTE_READWRITE,
PAGE_EXECUTE_WRITECOPY]}
PAGE_EXECUTE_WRITECOPY)
@property
def State(self):
raw_state = super(_MEMORY_BASIC_INFORMATION32, self).State
# Finally make a chooser somewhere ?
return self.STATE_MAPPER.get(raw_state, raw_state)
return self.STATE_MAPPER[raw_state]
@property
def Type(self):
raw_type = super(_MEMORY_BASIC_INFORMATION32, self).Type
# Finally make a chooser somewhere ?
return self.TYPE_MAPPER.get(raw_type, raw_type)
return self.TYPE_MAPPER[raw_type]
@property
def Protect(self):
raw_protect = super(_MEMORY_BASIC_INFORMATION32, self).Protect
# Finally make a chooser somewhere ?
return self.PROTECT_MAPPER.get(raw_protect, raw_protect)
return self.PROTECT_MAPPER[raw_protect]
def __repr__(self):
return "<MEMORY_BASIC_INFORMATION32 BaseAddress={0:#08x} RegionSize={1:#08x} State={2} Type={3} Protect={4}>".format(
ctypes_generation/extended_structs/_MEMORY_BASIC_INFORMATION64.py
+7 -7
View File
@@ -1,30 +1,30 @@
INITIAL_MEMORY_BASIC_INFORMATION64 = _MEMORY_BASIC_INFORMATION64
class _MEMORY_BASIC_INFORMATION64(INITIAL_MEMORY_BASIC_INFORMATION64):
STATE_MAPPER = {MEM_COMMIT: MEM_COMMIT, MEM_FREE: MEM_FREE, MEM_RESERVE:MEM_RESERVE}
TYPE_MAPPER = {MEM_IMAGE: MEM_IMAGE, MEM_MAPPED: MEM_MAPPED, MEM_PRIVATE:MEM_PRIVATE}
PROTECT_MAPPER = {x:x for x in [PAGE_NOACCESS, PAGE_READONLY, PAGE_READWRITE,
STATE_MAPPER = FlagMapper(MEM_COMMIT, MEM_FREE, MEM_RESERVE)
TYPE_MAPPER = FlagMapper(MEM_IMAGE, MEM_MAPPED, MEM_PRIVATE)
PROTECT_MAPPER = FlagMapper(PAGE_NOACCESS, PAGE_READONLY, PAGE_READWRITE,
PAGE_WRITECOPY, PAGE_EXECUTE, PAGE_EXECUTE_READ, PAGE_EXECUTE_READWRITE,
PAGE_EXECUTE_WRITECOPY]}
PAGE_EXECUTE_WRITECOPY)
@property
def State(self):
raw_state = super(_MEMORY_BASIC_INFORMATION64, self).State
# Finally make a chooser somewhere ?
return self.STATE_MAPPER.get(raw_state, raw_state)
return self.STATE_MAPPER[raw_state]
@property
def Type(self):
raw_type = super(_MEMORY_BASIC_INFORMATION64, self).Type
# Finally make a chooser somewhere ?
return self.TYPE_MAPPER.get(raw_type, raw_type)
return self.TYPE_MAPPER[raw_type]
@property
def Protect(self):
raw_protect = super(_MEMORY_BASIC_INFORMATION64, self).Protect
# Finally make a chooser somewhere ?
return self.PROTECT_MAPPER.get(raw_protect, raw_protect)
return self.PROTECT_MAPPER[raw_protect]
def __repr__(self):
return "<MEMORY_BASIC_INFORMATION64 BaseAddress={0:#08x} RegionSize={1:#08x} State={2} Type={3} Protect={4}>".format(
ctypes_generation/generate.py
+6
View File
@@ -230,6 +230,12 @@ class InitialDefGenerator(CtypesGenerator):
return Flag(name, value)
return StrFlag(name, value)
class FlagMapper(dict):
def __init__(self, *values):
self.update({x:x for x in values})
def __missing__(self, key):
return key
""")
IMPORT_HEADER = "{deps}"
windows/alpc.py
+2 -2
View File
@@ -184,7 +184,7 @@ KNOWN_ALPC_ATTRIBUTES = (gdef.ALPC_MESSAGE_SECURITY_ATTRIBUTE,
gdef.ALPC_MESSAGE_DIRECT_ATTRIBUTE,
gdef.ALPC_MESSAGE_WORK_ON_BEHALF_ATTRIBUTE)
KNOWN_ALPC_ATTRIBUTES_MAPPING = {x:x for x in KNOWN_ALPC_ATTRIBUTES}
KNOWN_ALPC_ATTRIBUTES_MAPPING = gdef.FlagMapper(*KNOWN_ALPC_ATTRIBUTES)
class MessageAttribute(gdef.ALPC_MESSAGE_ATTRIBUTES):
@@ -270,7 +270,7 @@ class MessageAttribute(gdef.ALPC_MESSAGE_ATTRIBUTES):
for mask in (1 << i for i in range(64)):
if value & mask:
attrs.append(mask)
return [KNOWN_ALPC_ATTRIBUTES_MAPPING.get(x, x) for x in attrs]
return [KNOWN_ALPC_ATTRIBUTES_MAPPING[x] for x in attrs]
@property
def valid_list(self):
windows/crypto/certificate.py
+2 -2
View File
@@ -29,7 +29,7 @@ CRYPT_OBJECT_FORMAT_TYPE = [
gdef.CERT_QUERY_CONTENT_PFX_AND_LOAD
]
CRYPT_OBJECT_FORMAT_TYPE_DICT = {x:x for x in CRYPT_OBJECT_FORMAT_TYPE}
CRYPT_OBJECT_FORMAT_TYPE_DICT = gdef.FlagMapper(*CRYPT_OBJECT_FORMAT_TYPE)
## Move CryptObject to new .py ?
@@ -68,7 +68,7 @@ class CryptObject(object):
self.cert_store = hStore if hStore else None
self.crypt_msg = hMsg if hMsg else None
self.encoding = dwEncoding
self.content_type = CRYPT_OBJECT_FORMAT_TYPE_DICT.get(dwContentType.value, dwContentType)
self.content_type = CRYPT_OBJECT_FORMAT_TYPE_DICT[dwContentType.value]
def _signers_and_certs_generator(self):
for signer in self.crypt_msg.signers:
windows/generated_def/windef.py
+6
View File
@@ -50,6 +50,12 @@ def make_flag(name, value):
return Flag(name, value)
return StrFlag(name, value)
class FlagMapper(dict):
def __init__(self, *values):
self.update({x:x for x in values})
def __missing__(self, key):
return key
from ntstatus import *
INVALID_HANDLE_VALUE = make_flag("INVALID_HANDLE_VALUE", ( ( -1 ) & NATIVE_WORD_MAX_VALUE ))
windows/generated_def/winstructs.py
+14 -14
View File
@@ -1358,30 +1358,30 @@ PMEMORY_BASIC_INFORMATION32 = POINTER(_MEMORY_BASIC_INFORMATION32)
INITIAL_MEMORY_BASIC_INFORMATION32 = _MEMORY_BASIC_INFORMATION32
class _MEMORY_BASIC_INFORMATION32(INITIAL_MEMORY_BASIC_INFORMATION32):
STATE_MAPPER = {MEM_COMMIT: MEM_COMMIT, MEM_FREE: MEM_FREE, MEM_RESERVE:MEM_RESERVE}
TYPE_MAPPER = {MEM_IMAGE: MEM_IMAGE, MEM_MAPPED: MEM_MAPPED, MEM_PRIVATE:MEM_PRIVATE}
PROTECT_MAPPER = {x:x for x in [PAGE_NOACCESS, PAGE_READONLY, PAGE_READWRITE,
STATE_MAPPER = FlagMapper(MEM_COMMIT, MEM_FREE, MEM_RESERVE)
TYPE_MAPPER = FlagMapper(MEM_IMAGE, MEM_MAPPED, MEM_PRIVATE)
PROTECT_MAPPER = FlagMapper(PAGE_NOACCESS, PAGE_READONLY, PAGE_READWRITE,
PAGE_WRITECOPY, PAGE_EXECUTE, PAGE_EXECUTE_READ, PAGE_EXECUTE_READWRITE,
PAGE_EXECUTE_WRITECOPY]}
PAGE_EXECUTE_WRITECOPY)
@property
def State(self):
raw_state = super(_MEMORY_BASIC_INFORMATION32, self).State
# Finally make a chooser somewhere ?
return self.STATE_MAPPER.get(raw_state, raw_state)
return self.STATE_MAPPER[raw_state]
@property
def Type(self):
raw_type = super(_MEMORY_BASIC_INFORMATION32, self).Type
# Finally make a chooser somewhere ?
return self.TYPE_MAPPER.get(raw_type, raw_type)
return self.TYPE_MAPPER[raw_type]
@property
def Protect(self):
raw_protect = super(_MEMORY_BASIC_INFORMATION32, self).Protect
# Finally make a chooser somewhere ?
return self.PROTECT_MAPPER.get(raw_protect, raw_protect)
return self.PROTECT_MAPPER[raw_protect]
def __repr__(self):
return "<MEMORY_BASIC_INFORMATION32 BaseAddress={0:#08x} RegionSize={1:#08x} State={2} Type={3} Protect={4}>".format(
@@ -1407,30 +1407,30 @@ MEMORY_BASIC_INFORMATION64 = _MEMORY_BASIC_INFORMATION64
INITIAL_MEMORY_BASIC_INFORMATION64 = _MEMORY_BASIC_INFORMATION64
class _MEMORY_BASIC_INFORMATION64(INITIAL_MEMORY_BASIC_INFORMATION64):
STATE_MAPPER = {MEM_COMMIT: MEM_COMMIT, MEM_FREE: MEM_FREE, MEM_RESERVE:MEM_RESERVE}
TYPE_MAPPER = {MEM_IMAGE: MEM_IMAGE, MEM_MAPPED: MEM_MAPPED, MEM_PRIVATE:MEM_PRIVATE}
PROTECT_MAPPER = {x:x for x in [PAGE_NOACCESS, PAGE_READONLY, PAGE_READWRITE,
STATE_MAPPER = FlagMapper(MEM_COMMIT, MEM_FREE, MEM_RESERVE)
TYPE_MAPPER = FlagMapper(MEM_IMAGE, MEM_MAPPED, MEM_PRIVATE)
PROTECT_MAPPER = FlagMapper(PAGE_NOACCESS, PAGE_READONLY, PAGE_READWRITE,
PAGE_WRITECOPY, PAGE_EXECUTE, PAGE_EXECUTE_READ, PAGE_EXECUTE_READWRITE,
PAGE_EXECUTE_WRITECOPY]}
PAGE_EXECUTE_WRITECOPY)
@property
def State(self):
raw_state = super(_MEMORY_BASIC_INFORMATION64, self).State
# Finally make a chooser somewhere ?
return self.STATE_MAPPER.get(raw_state, raw_state)
return self.STATE_MAPPER[raw_state]
@property
def Type(self):
raw_type = super(_MEMORY_BASIC_INFORMATION64, self).Type
# Finally make a chooser somewhere ?
return self.TYPE_MAPPER.get(raw_type, raw_type)
return self.TYPE_MAPPER[raw_type]
@property
def Protect(self):
raw_protect = super(_MEMORY_BASIC_INFORMATION64, self).Protect
# Finally make a chooser somewhere ?
return self.PROTECT_MAPPER.get(raw_protect, raw_protect)
return self.PROTECT_MAPPER[raw_protect]
def __repr__(self):
return "<MEMORY_BASIC_INFORMATION64 BaseAddress={0:#08x} RegionSize={1:#08x} State={2} Type={3} Protect={4}>".format(
windows/rpc/client.py
+4 -5
View File
@@ -6,20 +6,19 @@ import windows.com
import windows.generated_def as gdef
KNOW_REQUEST_TYPE = {x:x for x in [gdef.RPC_REQUEST_TYPE_CALL, gdef.RPC_REQUEST_TYPE_BIND]}
KNOW_REQUEST_TYPE = gdef.FlagMapper(gdef.RPC_REQUEST_TYPE_CALL, gdef.RPC_REQUEST_TYPE_BIND)
KNOW_RESPONSE_TYPE = {x:x for x in [gdef.RPC_RESPONSE_TYPE_FAIL, gdef.RPC_RESPONSE_TYPE_SUCESS, gdef.RPC_RESPONSE_TYPE_BIND_OK]}
KNOW_RESPONSE_TYPE = gdef.FlagMapper(gdef.RPC_RESPONSE_TYPE_FAIL, gdef.RPC_RESPONSE_TYPE_SUCESS, gdef.RPC_RESPONSE_TYPE_BIND_OK)
KNOWN_RPC_ERROR_CODE = {x:x for x in [
KNOWN_RPC_ERROR_CODE = gdef.FlagMapper(
gdef.ERROR_INVALID_HANDLE,
gdef.RPC_X_BAD_STUB_DATA,
gdef.RPC_S_UNKNOWN_IF,
gdef.RPC_S_PROTOCOL_ERROR,
gdef.RPC_S_UNSUPPORTED_TRANS_SYN,
gdef.RPC_S_PROCNUM_OUT_OF_RANGE,
]}
gdef.RPC_S_PROCNUM_OUT_OF_RANGE)
NOT_USED = 0xBAADF00D
windows/winobject/process.py
+3 -3
View File
@@ -12,6 +12,7 @@ import windows
import windows.native_exec.simple_x86 as x86
import windows.native_exec.simple_x64 as x64
import windows.remotectypes as rctypes
import windows.generated_def as gdef
from windows import injection
from windows import native_exec
@@ -21,7 +22,6 @@ from windows import utils
from windows.dbgprint import dbgprint
from windows.generated_def.winstructs import *
from windows.generated_def.ntstatus import NtStatusException
from windows.generated_def import windef
from windows.winobject import exception
from windows.winobject import sid
@@ -1139,7 +1139,7 @@ SECURITY_MANDATORY_HIGH_RID,
SECURITY_MANDATORY_SYSTEM_RID,
SECURITY_MANDATORY_PROTECTED_PROCESS_RID]
know_integrity_level_mapper = {x:x for x in KNOW_INTEGRITY_LEVEL}
know_integrity_level_mapper = gdef.FlagMapper(*KNOW_INTEGRITY_LEVEL)
# Create ProcessToken and Thread Token objects ?
class Token(AutoHandle):
@@ -1158,7 +1158,7 @@ class Token(AutoHandle):
sid = ctypes.cast(buffer, POINTER(TOKEN_MANDATORY_LABEL))[0].Label.Sid
count = winproxy.GetSidSubAuthorityCount(sid)
integrity = winproxy.GetSidSubAuthority(sid, count[0] - 1)[0]
return know_integrity_level_mapper.get(integrity, integrity)
return know_integrity_level_mapper[integrity]
def set_integrity(self, integrity):
"""Set the integrity level of a token
windows/winobject/service.py
+8 -7
View File
@@ -5,13 +5,14 @@ from collections import namedtuple
from contextlib import contextmanager
from windows import utils
import windows.generated_def as gdef
from windows.generated_def import *
SERVICE_TYPE = {x:x for x in [SERVICE_KERNEL_DRIVER, SERVICE_FILE_SYSTEM_DRIVER, SERVICE_WIN32_OWN_PROCESS, SERVICE_WIN32_SHARE_PROCESS, SERVICE_INTERACTIVE_PROCESS]}
SERVICE_STATE = {x:x for x in [SERVICE_STOPPED, SERVICE_START_PENDING, SERVICE_STOP_PENDING, SERVICE_RUNNING, SERVICE_CONTINUE_PENDING, SERVICE_PAUSE_PENDING, SERVICE_PAUSED]}
SERVICE_CONTROLE_ACCEPTED = {x:x for x in []}
SERVICE_FLAGS = {x:x for x in [SERVICE_RUNS_IN_SYSTEM_PROCESS]}
SERVICE_TYPE = gdef.FlagMapper(SERVICE_KERNEL_DRIVER, SERVICE_FILE_SYSTEM_DRIVER, SERVICE_WIN32_OWN_PROCESS, SERVICE_WIN32_SHARE_PROCESS, SERVICE_INTERACTIVE_PROCESS)
SERVICE_STATE = gdef.FlagMapper(SERVICE_STOPPED, SERVICE_START_PENDING, SERVICE_STOP_PENDING, SERVICE_RUNNING, SERVICE_CONTINUE_PENDING, SERVICE_PAUSE_PENDING, SERVICE_PAUSED)
SERVICE_CONTROLE_ACCEPTED = gdef.FlagMapper()
SERVICE_FLAGS = gdef.FlagMapper(SERVICE_RUNS_IN_SYSTEM_PROCESS)
ServiceStatus = namedtuple("ServiceStatus", ["type", "state", "control_accepted", "flags"])
@@ -70,10 +71,10 @@ class Service(object):
:type: :class:`ServiceStatus`
"""
status = self.ServiceStatusProcess
stype = SERVICE_TYPE.get(status.dwServiceType, status.dwServiceType)
sstate = SERVICE_STATE.get(status.dwCurrentState, status.dwCurrentState)
stype = SERVICE_TYPE[status.dwServiceType]
sstate = SERVICE_STATE[status.dwCurrentState]
scontrol = status.dwControlsAccepted
sflags = SERVICE_FLAGS.get(status.dwServiceFlags, status.dwServiceFlags)
sflags = SERVICE_FLAGS[status.dwServiceFlags]
return ServiceStatus(stype, sstate, scontrol, sflags)
@utils.fixedpropety
windows/winobject/system.py
+7 -6
View File
@@ -6,7 +6,8 @@ import struct
import windows
from windows import winproxy
from windows import utils
from windows.generated_def import windef
import windows.generated_def as gdef
from windows.winobject import process
from windows.winobject import network
@@ -166,6 +167,7 @@ class System(object):
else:
return "Unknow Windows <version={0} | is_workstation={1}>".format(version, is_workstation)
VERSION_MAPPER = gdef.FlagMapper(VER_NT_WORKSTATION, VER_NT_DOMAIN_CONTROLLER, VER_NT_SERVER)
@utils.fixedpropety
def product_type(self):
"""The product type, value might be:
@@ -176,9 +178,8 @@ class System(object):
:type: :class:`long` or :class:`int` (or subclass)
"""
version_map = {x:x for x in [VER_NT_WORKSTATION, VER_NT_DOMAIN_CONTROLLER, VER_NT_SERVER]}
version = self.get_version()
return version_map.get(version.wProductType, version.wProductType)
return self.VERSION_MAPPER[version.wProductType]
@utils.fixedpropety
@@ -217,7 +218,7 @@ class System(object):
dbgprint("Enumerating processes with CreateToolhelp32Snapshot", "SLOW")
process_entry = PROCESSENTRY32()
process_entry.dwSize = ctypes.sizeof(process_entry)
snap = winproxy.CreateToolhelp32Snapshot(windef.TH32CS_SNAPPROCESS, 0)
snap = winproxy.CreateToolhelp32Snapshot(gdef.TH32CS_SNAPPROCESS, 0)
winproxy.Process32First(snap, process_entry)
res = []
res.append(process.WinProcess._from_PROCESSENTRY32(process_entry))
@@ -232,7 +233,7 @@ class System(object):
dbgprint("Enumerating threads with CreateToolhelp32Snapshot <generator>", "SLOW")
thread_entry = THREADENTRY32()
thread_entry.dwSize = ctypes.sizeof(thread_entry)
snap = winproxy.CreateToolhelp32Snapshot(windef.TH32CS_SNAPTHREAD, 0)
snap = winproxy.CreateToolhelp32Snapshot(gdef.TH32CS_SNAPTHREAD, 0)
dbgprint("New handle CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD) <generator> | {0:#x}".format(snap), "HANDLE")
try:
winproxy.Thread32First(snap, thread_entry)
@@ -257,7 +258,7 @@ class System(object):
dbgprint("Enumerating threads with CreateToolhelp32Snapshot and setup owner", "SLOW")
# One snap for both enum to be prevent race
snap = winproxy.CreateToolhelp32Snapshot(windef.TH32CS_SNAPTHREAD | windef.TH32CS_SNAPPROCESS, 0)
snap = winproxy.CreateToolhelp32Snapshot(gdef.TH32CS_SNAPTHREAD | gdef.TH32CS_SNAPPROCESS, 0)
process_entry = PROCESSENTRY32()
process_entry.dwSize = ctypes.sizeof(process_entry)
windows/winobject/volume.py
+3 -2
View File
@@ -2,14 +2,15 @@ import ctypes
import windows
from windows import winproxy
import windows.generated_def as gdef
from windows.generated_def.winstructs import *
class LogicalDrive(object):
DRIVE_TYPE = {x:x for x in [DRIVE_UNKNOWN, DRIVE_NO_ROOT_DIR, DRIVE_REMOVABLE,
DRIVE_FIXED, DRIVE_REMOTE, DRIVE_CDROM, DRIVE_RAMDISK]}
DRIVE_TYPE = gdef.FlagMapper(DRIVE_UNKNOWN, DRIVE_NO_ROOT_DIR, DRIVE_REMOVABLE,
DRIVE_FIXED, DRIVE_REMOTE, DRIVE_CDROM, DRIVE_RAMDISK)
def __init__(self, name):
self.name = name
windows/wintrust.py
+3 -2
View File
@@ -3,6 +3,7 @@ import struct
import windows
from collections import namedtuple
from windows import winproxy
import windows.generated_def as gdef
from windows.generated_def.winstructs import *
@@ -75,7 +76,7 @@ CERT_E_INVALID_POLICY,
CERT_E_INVALID_NAME,
CRYPT_E_FILE_ERROR,
]
wintrust_return_value_mapper = {x:x for x in wintrust_know_return_value}
wintrust_return_value_mapper = gdef.FlagMapper(*wintrust_know_return_value)
def check_signature(filename):
@@ -109,7 +110,7 @@ def check_signature(filename):
x = winproxy.WinVerifyTrust(None, ctypes.byref(WVTPolicyGUID), ctypes.byref(win_trust_data))
win_trust_data.dwStateAction = WTD_STATEACTION_CLOSE
winproxy.WinVerifyTrust(None, ctypes.byref(WVTPolicyGUID), ctypes.byref(win_trust_data))
return wintrust_return_value_mapper.get(x & 0xffffffff, x & 0xffffffff)
return wintrust_return_value_mapper[x & 0xffffffff]
def get_catalog_for_filename(filename):