First version of windows.security based on clmntb work

This commit is contained in:
hakril
2018-09-26 09:48:34 +02:00
parent 0e52272ddc
commit bbde38ffe4
4 changed files with 772 additions and 4 deletions
+4
View File
@@ -486,6 +486,10 @@ The local debugger handles
- Network
- COM
## Acknowledgments
* clmntb for his initial work on ``windows.security``
[LKD_GITHUB]: https://github.com/sogeti-esec-lab/LKD/
[SAMPLE_DIR]: https://github.com/hakril/PythonForWindows/tree/master/samples
+195
View File
@@ -0,0 +1,195 @@
import windows.security
from windows.security import SecurityDescriptor
from pfwtest import *
import ctypes
# CC -> Create-Child -> 1
# GR -> Generic read -> 0x80000000L
# AN -> Anonymous -> S-1-5-7
TEST_SDDL = [
"O:ANG:AND:(A;;RPWPCCDCLCSWRCWDWOGA;;;S-1-0-0)(D;;RPWPCCDCLCSWRCWDWOGA;;;S-1-0-0)",
"O:ANG:AND:(A;;GR;;;S-1-0-0)",
"O:ANG:AND:(OA;;CC;;00000042-0043-0044-0045-000000000000;S-1-0-0)",
"O:ANG:AND:(OA;;CCGR;00004242-0043-0044-0045-000000000000;00000042-0043-0044-0045-000000000000;S-1-0-0)",
]
@pytest.mark.parametrize("sddl", TEST_SDDL)
def test_security_descriptor_from_string(sddl):
sd = SecurityDescriptor.from_string(sddl)
def test_pacl_object():
SDDL = "O:ANG:S-1-2-3D:(A;;;;;S-1-42-42)(A;;;;;S-1-42-43)(A;;;;;S-1-42-44)"
dacl = SecurityDescriptor.from_string(SDDL).dacl
assert dacl is not None
assert len(dacl) == 3 # __len__
assert len(list(dacl)) == 3 # __iter__
assert len(dacl.aces) == 3
assert ctypes.addressof(dacl[0]) == ctypes.addressof(dacl[0]) # __getitem__
assert len([ctypes.addressof(dacl[i])for i in range(3)]) == 3
with pytest.raises(IndexError):
x = dacl[3]
def test_sec_descrip_owner_group():
SDDL = "O:ANG:S-1-2-3"
sd = SecurityDescriptor.from_string(SDDL)
assert sd.owner.to_string() == "S-1-5-7"
assert sd.group.to_string() == "S-1-2-3"
assert sd.dacl is None
assert sd.sacl is None
def test_mask_sid_ace():
SDDL = "D:(A;CIOI;CCGR;;;S-1-42-42)"
# OBJECT_INHERIT_ACE(0x1L) | CONTAINER_INHERIT_ACE(0x2L)
# Create-Child | GENERIC_READ(0x80000000L)
sd = SecurityDescriptor.from_string(SDDL)
dacl = sd.dacl
assert dacl is not None
ace = dacl[0]
# Test the ACE
assert ace.Header.AceType == gdef.ACCESS_ALLOWED_ACE_TYPE
# flags + flags split
assert ace.Header.AceFlags == gdef.OBJECT_INHERIT_ACE | gdef.CONTAINER_INHERIT_ACE
assert set(ace.Header.flags) == {gdef.OBJECT_INHERIT_ACE, gdef.CONTAINER_INHERIT_ACE}
# mask + mask split
assert ace.Mask == 1 | gdef.GENERIC_READ
assert set(ace.mask) == {1, gdef.GENERIC_READ}
# SID
assert ace.sid.to_string() == "S-1-42-42"
SGUID = gdef.GUID.from_string
COMPLEXE_SDDL_GUID = [
("D:(OA;;;00000042-0043-0044-0045-000000000001;;S-1-0-0)",
SGUID("00000042-0043-0044-0045-000000000001"),
None),
("D:(OA;;;;00000042-0043-0044-0045-000000000000;S-1-0-0)",
None,
SGUID("00000042-0043-0044-0045-000000000000")),
("D:(OA;;;00000042-0043-0044-0045-000000000002;00000042-0043-0044-0045-000000000003;S-1-0-0)",
SGUID("00000042-0043-0044-0045-000000000002"),
SGUID("00000042-0043-0044-0045-000000000003")),
("D:(OA;;;;;S-1-0-0)",
None,
None),
]
@pytest.mark.parametrize("sddl, obj_guid, inherited_object_guid", COMPLEXE_SDDL_GUID)
def test_complex_ace_guid_sid(sddl, obj_guid, inherited_object_guid):
print(sddl)
sd = SecurityDescriptor.from_string(sddl)
assert sd.dacl is not None
ace = sd.dacl[0]
assert ace.sid.to_string() == "S-1-0-0"
if obj_guid is None and inherited_object_guid is None:
# No GUID -> transformed in ACCESS_ALLOWED_ACE_TYPE
assert ace.Header.AceType == gdef.ACCESS_ALLOWED_ACE_TYPE
return
assert ace.object_type == obj_guid
assert ace.inherited_object_type == inherited_object_guid
ALL_DACL_ACE_TYPES = [
("D:(A;;;;;S-1-2-3)", gdef.ACCESS_ALLOWED_ACE_TYPE),
("D:(D;;;;;S-1-2-3)", gdef.ACCESS_DENIED_ACE_TYPE),
("D:(OA;;;;00000042-0043-0044-0045-000000000000;S-1-0-0)",
gdef.ACCESS_ALLOWED_OBJECT_ACE_TYPE),
("D:(OD;;;;00000042-0043-0044-0045-000000000001;S-1-0-0)",
gdef.ACCESS_DENIED_OBJECT_ACE_TYPE),
("D:AI(XA;;GR;;;WD;(YOLO))", gdef.ACCESS_ALLOWED_CALLBACK_ACE_TYPE),
("D:AI(XD;;GR;;;WD;(YOLO))", gdef.ACCESS_DENIED_CALLBACK_ACE_TYPE),
("D:AI(ZA;;GR;;00000042-0043-0044-0045-000000000001;WD;(YOLO))", gdef.ACCESS_ALLOWED_CALLBACK_OBJECT_ACE_TYPE),
# NO SDDL DEFINE FOR : gdef.ACCESS_DENIED_CALLBACK_OBJECT_ACE_TYPE)
]
@pytest.mark.parametrize("sddl, ace_type", ALL_DACL_ACE_TYPES)
def test_ace_dacl_subclass(sddl, ace_type):
sd = SecurityDescriptor.from_string(sddl)
dacl = sd.dacl
assert len(dacl) == 1
ace = dacl[0] # Will raise if AceHeader is not handled
assert ace.Header.AceType == ace_type
# SACL STUFF
ALL_SACL_ACE_TYPES = [
("S:(AU;;;;;AN)", gdef.SYSTEM_AUDIT_ACE_TYPE),
("S:(ML;;;;;S-1-16-4000)", gdef.SYSTEM_MANDATORY_LABEL_ACE_TYPE),
# S-1-19-512-4096 what retrieved in a ACE from a directory in C:\Program Files\WindowsApps\
("S:(TL;;;;;S-1-19-512-4096)", gdef.SYSTEM_PROCESS_TRUST_LABEL_ACE_TYPE),
("S:(SP;;;;;S-1-17-1)", gdef.SYSTEM_SCOPED_POLICY_ID_ACE_TYPE),
("S:(OU;;;;00000042-0043-0044-0045-000000000000;AN)", gdef.SYSTEM_AUDIT_OBJECT_ACE_TYPE),
("S:(XU;;;;;S-1-2-3;(YOLO))", gdef.SYSTEM_AUDIT_CALLBACK_ACE_TYPE),
## Reserved for futur use (RFS): not handled by ADVAPI.dll
# ("S:(AL;;;;;S-1-2-3)", gdef.SYSTEM_ALARM_OBJECT_ACE_TYPE),
#("S:(OL;;;;00000042-0043-0044-0045-000000000000;AN)", gdef.SYSTEM_ALARM_OBJECT_ACE_TYPE),
## NO SDDL FOR:
# SYSTEM_ALARM_CALLBACK_ACE_TYPE
# SYSTEM_AUDIT_CALLBACK_OBJECT_ACE_TYPE
# SYSTEM_ALARM_CALLBACK_OBJECT_ACE_TYPE
]
@pytest.mark.parametrize("sddl, ace_type", ALL_SACL_ACE_TYPES)
def test_ace_sacl_subclass(sddl, ace_type):
sd = SecurityDescriptor.from_string(sddl)
sacl = sd.sacl
assert len(sacl) == 1
ace = sacl[0] # Will raise if AceHeader is not handled
assert ace.Header.AceType == ace_type
RESOURCE_ATTRIBUTES_SDDLS = [
("""S:(RA;;;;;WD; ("TestName",TI,0,-2, -1, 0, 1, 2))""",
(-2, -1, 0, 1, 2 )),
("""S:(RA;;;;;WD; ("TestName",TU,0,3,4,42))""",
(3, 4, 42)),
("""S:(RA;;;;;WD; ("TestName",TS,0,"Windows","SQL", ""))""",
("Windows", "SQL", "")),
("""S:(RA;;;;;WD; ("TestName",TD,0, AN, S-1-2-3-4-5-6-7-8-9))""",
(gdef.PSID.from_string("S-1-5-7"),
gdef.PSID.from_string("S-1-2-3-4-5-6-7-8-9"))),
("""S:(RA;;;;;WD; ("TestName",TX,0, 42000042, 0123456789abcdef))""",
("B\x00\x00B", "\x01\x23\x45\x67\x89\xab\xcd\xef")),
("""S:(RA;;;;;WD; ("TestName",TB,0, 0, 1, 0, 0, 1))""",
(False, True, False, False, True)),
]
@pytest.mark.parametrize("sddl, expected_values", RESOURCE_ATTRIBUTES_SDDLS)
def test_ace_resource_attribute(sddl, expected_values):
sd = SecurityDescriptor.from_string(sddl)
ra = sd.sacl[0]
assert ra.Header.AceType == gdef.SYSTEM_RESOURCE_ATTRIBUTE_ACE_TYPE
attr = ra.attribute
assert attr.name == "TestName"
assert attr.values == expected_values
CONDITIONAL_SDDLS = [
("D:AI(XA;;GR;;;WD;(ATTR1))", "ATTR1"),
("D:AI(XD;;GR;;;WD;(ATTR2))", "ATTR2"),
("S:AI(XU;;GR;;;WD;(ATTR3))", "ATTR3")
]
@pytest.mark.parametrize("sddl, expected_value", CONDITIONAL_SDDLS)
def test_conditional_ace_applicationdata(sddl, expected_value):
sd = SecurityDescriptor.from_string(sddl)
acl = sd.dacl
if acl is None:
acl = sd.sacl
ace = acl[0]
appdata = ace.application_data
# https://msdn.microsoft.com/en-us/library/hh877860.aspx
assert appdata.startswith("artx")
assert expected_value in appdata.replace("\x00", "")
+492
View File
@@ -0,0 +1,492 @@
import ctypes
import sys
import windows
import windows.generated_def as gdef
from windows import winproxy
# Temporary ? real API ?
def lookup_sid(psid):
usernamesize = gdef.DWORD(0x1000)
computernamesize = gdef.DWORD(0x1000)
username = ctypes.c_buffer(usernamesize.value)
computername = ctypes.c_buffer(computernamesize.value)
peUse = gdef.SID_NAME_USE()
winproxy.LookupAccountSidA(None, psid, username, usernamesize, computername, computernamesize, peUse)
return computername[:computernamesize.value], username[:usernamesize.value]
# ACE
ACE_FLAGS = gdef.FlagMapper(
gdef.OBJECT_INHERIT_ACE ,
gdef.CONTAINER_INHERIT_ACE ,
gdef.NO_PROPAGATE_INHERIT_ACE ,
gdef.INHERIT_ONLY_ACE ,
gdef.INHERITED_ACE ,
gdef.VALID_INHERIT_FLAGS ,
gdef.SUCCESSFUL_ACCESS_ACE_FLAG,
gdef.FAILED_ACCESS_ACE_FLAG
)
ACE_MASKS = gdef.FlagMapper(
gdef.GENERIC_READ ,
gdef.GENERIC_WRITE ,
gdef.GENERIC_EXECUTE ,
gdef.GENERIC_ALL ,
gdef.READ_CONTROL ,
gdef.DELETE ,
gdef.WRITE_DAC ,
gdef.WRITE_OWNER ,
)
class AceHeader(gdef.ACE_HEADER):
"""Improved ACE_HEADER"""
def _to_ace_type(self, ace_type):
return ctypes.cast(ctypes.byref(self), ctypes.POINTER(ace_type))[0]
@property
def AceType(self):
raw_type = super(AceHeader, self).AceType
return ACE_CLASS_TYPE_MAPPER[raw_type]
@property
def flags(self):
return list(self._flags_generator())
def _flags_generator(self):
flags = self.AceFlags
for i in range(8): # Sizeof(AceFlags) * 8
v = flags & (1 << i)
if v:
yield ACE_FLAGS[v]
def subclass(self):
# ACE_CLASS_BY_ACE_TYPE is defined later in this file
subcls = ACE_CLASS_BY_ACE_TYPE[self.AceType]
return self._to_ace_type(subcls)
def __repr__(self):
return "<{0} type={1}>".format(type(self).__name__, self.AceType)
class AceBase(object): # Ca ou mettre flags extraction dans le ctypes generated
@property
def Header(self): # Override the ctypes Header for the struct -> return extended header
addr = ctypes.addressof(self)
sheader = super(AceBase, type(self)).Header
return AceHeader.from_address(addr + sheader.offset)
class MaskAndSidACE(AceBase):
# "Virtual" ACE for ACE struct with
# ACE_HEADER Header;
# ACCESS_MASK Mask;
# DWORD SidStart;
def _sid_offset(self):
return type(self).SidStart.offset
@property
def sid(self):
return gdef.PSID(ctypes.addressof(self) + self._sid_offset())
@property
def mask(self):
return list(self._mask_generator())
def _mask_generator(self):
mask = self.Mask
for i in range(32): # sizeof ACCESS_MASK * 8
v = mask & (1 << i)
if v:
yield ACE_MASKS[v]
def __repr__(self):
return "<{0} mask={1}>".format(type(self).__name__, self.Mask)
class CallbackACE(MaskAndSidACE):
@property
def application_data(self):
"""FROM : https://msdn.microsoft.com/en-us/library/hh877860.aspx"""
selfptr = ctypes.cast(ctypes.addressof(self), gdef.PUCHAR)
datastart = ctypes.sizeof(self) + self.sid.size - 4
dataend = self.Header.AceSize
return selfptr[datastart: dataend]
class ObjectRelatedACE(MaskAndSidACE):
FLAGS_VALUES = (gdef.ACE_OBJECT_TYPE_PRESENT,
gdef.ACE_INHERITED_OBJECT_TYPE_PRESENT)
@property
def flags(self):
flags = self.Flags
return [x for x in self.FLAGS_VALUES if flags & x]
@property
def object_type(self):
if not self.Flags & gdef.ACE_OBJECT_TYPE_PRESENT:
return None
return self.ObjectType
@property
def inherited_object_type(self):
if not self.Flags & gdef.ACE_INHERITED_OBJECT_TYPE_PRESENT:
return None
if self.Flags & gdef.ACE_OBJECT_TYPE_PRESENT:
# There is an ObjectType so our offset is the good one
return self.InheritedObjectType
# No ObjectType -> InheritedObjectType is at ObjectType offset
# Those are the same type so we can directly use ObjectType
return self.ObjectType
def _sid_offset(self):
base_offset = type(self).SidStart.offset
if not self.Flags & gdef.ACE_OBJECT_TYPE_PRESENT:
base_offset -= ctypes.sizeof(gdef.GUID)
if not self.Flags & gdef.ACE_INHERITED_OBJECT_TYPE_PRESENT:
base_offset -= ctypes.sizeof(gdef.GUID)
return base_offset
# DACL related ACE
# Allow the resolution of Header first
class AccessAllowedACE(MaskAndSidACE, gdef.ACCESS_ALLOWED_ACE):
ACE_TYPE = gdef.ACCESS_ALLOWED_ACE_TYPE
class AccessDeniedACE(MaskAndSidACE, gdef.ACCESS_DENIED_ACE):
ACE_TYPE = gdef.ACCESS_DENIED_ACE_TYPE
class AccessAllowedCallbackACE(CallbackACE, gdef.ACCESS_ALLOWED_CALLBACK_ACE):
ACE_TYPE = gdef.ACCESS_ALLOWED_CALLBACK_ACE_TYPE
class AccessDeniedCallbackACE(CallbackACE, gdef.ACCESS_DENIED_CALLBACK_ACE):
ACE_TYPE = gdef.ACCESS_DENIED_CALLBACK_ACE_TYPE
class AccessAllowedObjectACE(ObjectRelatedACE, gdef.ACCESS_ALLOWED_OBJECT_ACE):
ACE_TYPE = gdef.ACCESS_ALLOWED_OBJECT_ACE_TYPE
class AccessDeniedObjectACE(ObjectRelatedACE, gdef.ACCESS_DENIED_OBJECT_ACE):
ACE_TYPE = gdef.ACCESS_DENIED_OBJECT_ACE_TYPE
class AccessAllowedCallbackObjectACE(CallbackACE, gdef.ACCESS_ALLOWED_CALLBACK_OBJECT_ACE):
ACE_TYPE = gdef.ACCESS_ALLOWED_CALLBACK_OBJECT_ACE_TYPE
# Strangly -> no SDDL for this one
class AccessDeniedCallbackObjectACE(CallbackACE, gdef.ACCESS_DENIED_CALLBACK_OBJECT_ACE):
ACE_TYPE = gdef.ACCESS_DENIED_CALLBACK_OBJECT_ACE_TYPE
# SACL related ACE
class SystemAuditACE(MaskAndSidACE, gdef.SYSTEM_AUDIT_ACE):
ACE_TYPE = gdef.SYSTEM_AUDIT_ACE_TYPE
class SystemAlarmACE(MaskAndSidACE, gdef.SYSTEM_ALARM_ACE):
"""reserved for future use."""
ACE_TYPE = gdef.SYSTEM_ALARM_ACE_TYPE
class SystemAuditObjectACE(ObjectRelatedACE, gdef.SYSTEM_AUDIT_OBJECT_ACE):
ACE_TYPE = gdef.SYSTEM_AUDIT_OBJECT_ACE_TYPE
class SystemAlarmObjectACE(ObjectRelatedACE, gdef.SYSTEM_ALARM_OBJECT_ACE):
"""reserved for future use."""
ACE_TYPE = gdef.SYSTEM_ALARM_OBJECT_ACE_TYPE
class SystemAuditCallbackACE(CallbackACE, gdef.SYSTEM_AUDIT_CALLBACK_ACE):
ACE_TYPE = gdef.SYSTEM_AUDIT_CALLBACK_ACE_TYPE
class SystemAlarmCallbackACE(CallbackACE, gdef.SYSTEM_ALARM_CALLBACK_ACE):
"""reserved for future use."""
ACE_TYPE = gdef.SYSTEM_ALARM_CALLBACK_ACE_TYPE
class SystemAuditCallbackObjectACE(CallbackACE, gdef.SYSTEM_AUDIT_CALLBACK_OBJECT_ACE):
ACE_TYPE = gdef.SYSTEM_AUDIT_CALLBACK_OBJECT_ACE_TYPE
class SystemAlarmCallbackObjectACE(CallbackACE, gdef.SYSTEM_ALARM_CALLBACK_OBJECT_ACE):
"""Reserved for future use"""
ACE_TYPE = gdef.SYSTEM_ALARM_CALLBACK_OBJECT_ACE_TYPE
class SystemMandatoryLabelACE(MaskAndSidACE, gdef.SYSTEM_MANDATORY_LABEL_ACE):
ACE_TYPE = gdef.SYSTEM_MANDATORY_LABEL_ACE_TYPE
class SystemResourceAttributeACE(MaskAndSidACE, gdef.SYSTEM_RESOURCE_ATTRIBUTE_ACE):
ACE_TYPE = gdef.SYSTEM_RESOURCE_ATTRIBUTE_ACE_TYPE
@property
def attribute(self):
# Sid-size not in the initial struct
sid_size_over = self.sid.size - type(self).SidStart.size
sec_attr_addr = ctypes.addressof(self) + ctypes.sizeof(self) + sid_size_over
return ClaimSecurityAttributeRelativeV1.from_address(sec_attr_addr)
class SystemScopedPolicyIDACE(MaskAndSidACE, gdef.SYSTEM_SCOPED_POLICY_ID_ACE):
ACE_TYPE = gdef.SYSTEM_SCOPED_POLICY_ID_ACE_TYPE
class SystemProcessTrustLabelACE(MaskAndSidACE, gdef.SYSTEM_PROCESS_TRUST_LABEL_ACE):
"""Reserved. (from MSDC)"""
ACE_TYPE = gdef.SYSTEM_PROCESS_TRUST_LABEL_ACE_TYPE
ACE_CLASS_BY_ACE_TYPE = {cls.ACE_TYPE: cls for cls in (
# DACL
AccessAllowedACE,
AccessDeniedACE,
AccessAllowedCallbackACE,
AccessDeniedCallbackACE,
AccessAllowedObjectACE,
AccessDeniedObjectACE,
AccessAllowedCallbackObjectACE,
# SACL
SystemAuditACE,
SystemAlarmACE, # reserved for future use.
SystemAuditObjectACE,
SystemAlarmObjectACE, # reserved for future use.
SystemAuditCallbackACE,
SystemAlarmCallbackACE, # reserved for future use.
SystemAuditCallbackObjectACE,
SystemAlarmCallbackObjectACE, # reserved for future use.
SystemMandatoryLabelACE,
SystemResourceAttributeACE,
SystemScopedPolicyIDACE,
SystemProcessTrustLabelACE,
)}
ACE_CLASS_TYPE_MAPPER = gdef.FlagMapper(*ACE_CLASS_BY_ACE_TYPE.keys())
# CLAIM_SECURITY_ATTRIBUTE_RELATIVE_V1 follow the SYSTEM_RESOURCE_ATTRIBUTE_ACE
# For ACE of type SYSTEM_RESOURCE_ATTRIBUTE_ACE_TYPE
def retrieve_long64_from_addr(addr):
return gdef.LONG64.from_address(addr).value
def retrieve_ulong64_from_addr(addr):
return gdef.ULONG64.from_address(addr).value
def retrieve_wstr_from_addr(addr):
return gdef.LPWSTR(addr).value
# https://msdn.microsoft.com/en-us/library/hh877847.aspx
def retrieve_psid_from_addr(addr):
psid_addr = addr + gdef.CLAIM_SECURITY_ATTRIBUTE_OCTET_STRING_RELATIVE.OctetString.offset
return gdef.PSID(psid_addr)
def retrieve_bool_from_addr(addr):
return bool(gdef.ULONG64.from_address(addr).value)
def retrieve_octet_string_from_addr(addr):
# Good doc: https://msdn.microsoft.com/en-us/library/hh877833.aspx
# Doc broken in: https://docs.microsoft.com/en-us/windows/desktop/api/winnt/ns-winnt-_claim_security_attribute_relative_v1
ostring = gdef.CLAIM_SECURITY_ATTRIBUTE_OCTET_STRING_RELATIVE.from_address(addr)
# Bypass the array limit
return ctypes.cast(ostring.OctetString, gdef.PUCHAR)[:ostring.Length]
class ClaimSecurityAttributeRelativeV1(gdef.CLAIM_SECURITY_ATTRIBUTE_RELATIVE_V1):
VALUE_ARRAY_PTR_BY_TYPE = {
gdef.CLAIM_SECURITY_ATTRIBUTE_TYPE_INT64:
("pInt64", retrieve_long64_from_addr),
gdef.CLAIM_SECURITY_ATTRIBUTE_TYPE_UINT64:
("pUint64", retrieve_ulong64_from_addr),
gdef.CLAIM_SECURITY_ATTRIBUTE_TYPE_STRING:
("ppString", retrieve_wstr_from_addr),
gdef.CLAIM_SECURITY_ATTRIBUTE_TYPE_SID:
# ppString is not the good one
# But none is doc for PSID
("ppString", retrieve_psid_from_addr),
gdef.CLAIM_SECURITY_ATTRIBUTE_TYPE_BOOLEAN:
("pUint64", retrieve_bool_from_addr),
gdef.CLAIM_SECURITY_ATTRIBUTE_TYPE_OCTET_STRING:
("pOctetString", retrieve_octet_string_from_addr),
}
@property
def name(self):
return gdef.LPWSTR(ctypes.addressof(self) + self.Name).value
@property
def values(self):
array_name, get_value = self.VALUE_ARRAY_PTR_BY_TYPE[self.ValueType]
base = ctypes.addressof(self)
array = getattr(self.Values, array_name)
# The pointer allow us to bypass the array _length_ of 1
array_ptr = ctypes.cast(array, ctypes.POINTER(array._type_))
offsets = array_ptr[:self.ValueCount]
# Cast values
return tuple(get_value(base + off) for off in offsets)
# ACL
class Acl(gdef.ACL):
@property
def size_info(self):
size_info = gdef.ACL_SIZE_INFORMATION()
winproxy.GetAclInformation(self, ctypes.byref(size_info), ctypes.sizeof(size_info), gdef.AclSizeInformation)
return size_info
def get_ace(self, i):
ace = gdef.PVOID()
winproxy.GetAce(self, i, ace)
# TODO: subclass ACL
return AceHeader.from_address(ace.value).subclass()
@property
def aces(self):
return list(self)
def __len__(self):
return self.AceCount
def __getitem__(self, i):
try:
return self.get_ace(i)
except WindowsError as e:
if e.winerror == gdef.ERROR_INVALID_PARAMETER:
raise IndexError("Invalid ACL index {0}".format(i))
raise
def __iter__(self):
for i in range(self.AceCount):
yield self.get_ace(i)
def __repr__(self):
return "<Acl count={0}>".format(self.AceCount)
# Security descriptor
class SecurityDescriptor(gdef.PSECURITY_DESCRIPTOR):
"""TODO: free the underliying buffer when not needed anymore
for now the underliying memory is never free
"""
DEFAULT_SECURITY_INFORMATION = (
gdef.OWNER_SECURITY_INFORMATION |
gdef.GROUP_SECURITY_INFORMATION |
gdef.DACL_SECURITY_INFORMATION |
gdef.ATTRIBUTE_SECURITY_INFORMATION |
# gdef.SACL_SECURITY_INFORMATION | # Need special rights
gdef.SCOPE_SECURITY_INFORMATION |
gdef.PROCESS_TRUST_LABEL_SECURITY_INFORMATION
)
_close_function = winproxy.LocalFree
# def __init__(self, needs_free=True):
# self._needs_free = needs_free
@property
def control(self):
lpdwRevision = gdef.DWORD()
control = gdef.SECURITY_DESCRIPTOR_CONTROL()
winproxy.GetSecurityDescriptorControl(self, control, lpdwRevision)
return control.value
@property
def revision(self):
lpdwRevision = gdef.DWORD()
control = gdef.SECURITY_DESCRIPTOR_CONTROL()
winproxy.GetSecurityDescriptorControl(self, control, lpdwRevision)
return lpdwRevision.value
@property
def owner(self):
owner = gdef.PSID()
lpbOwnerDefaulted = gdef.BOOL()
winproxy.GetSecurityDescriptorOwner(self, owner, lpbOwnerDefaulted)
return owner
@property
def group(self):
group = gdef.PSID()
lpbGroupDefaulted = gdef.BOOL()
winproxy.GetSecurityDescriptorGroup(self, group, lpbGroupDefaulted)
return group
@property
def dacl(self):
dacl_present = gdef.BOOL()
pdacl = gdef.PACL()
lpbDaclDefaulted = gdef.BOOL()
winproxy.GetSecurityDescriptorDacl(self, dacl_present, pdacl, lpbDaclDefaulted)
if not dacl_present or not pdacl:
return None
return ctypes.cast(pdacl, ctypes.POINTER(Acl))[0]
@property
def sacl(self):
sacl_present = gdef.BOOL()
psacl = gdef.PACL()
lpbSaclDefaulted = gdef.BOOL()
winproxy.GetSecurityDescriptorSacl(self, sacl_present, psacl, lpbSaclDefaulted)
if not sacl_present or not psacl:
return None
return ctypes.cast(psacl, ctypes.POINTER(Acl))[0]
# Constructors
@classmethod
def from_string(cls, sddl):
self = cls()
winproxy.ConvertStringSecurityDescriptorToSecurityDescriptorA(
sddl,
gdef.SDDL_REVISION_1,
self,
None)
# TODO: we need to free this buffer..
# Keep track of Security Descritor state ?
return self
@classmethod
def _from_name_and_type(cls, objname, objtype, query_sacl=False, security_infos=DEFAULT_SECURITY_INFORMATION):
self = cls()
if query_sacl:
security_infos |= gdef.SACL_SECURITY_INFORMATION
winproxy.GetNamedSecurityInfoA(
objname,
objtype,
security_infos,
None,
None,
None,
None,
self
)
return self
@classmethod
def from_filename(cls, filename, query_sacl=False):
return cls._from_name_and_type(filename, gdef.SE_FILE_OBJECT, query_sacl=query_sacl)
def to_string(self, security_information=DEFAULT_SECURITY_INFORMATION):
result_cstr = gdef.LPSTR()
winproxy.ConvertSecurityDescriptorToStringSecurityDescriptorA(
self,
gdef.SDDL_REVISION_1,
security_information,
result_cstr,
None)
result = result_cstr.value # Retrieve a python-str copy
winproxy.LocalFree(result_cstr)
return result
# TST
# def relative(self):
# return bool(self.control & gdef.SE_SELF_RELATIVE)
# If we want auto-free we need to handle relf-relative SD
# We need to keep-track of sub-object of the SD
# Just a ref to SD from SACL / DACL ?
# def __del__(self):
# if self._needs_free and sys.path is not None:
# print("FREE SELF")
# self._close_function(self)
+81 -4
View File
@@ -40,8 +40,8 @@ def resolve(winfunc):
class Kernel32Error(WindowsError):
def __new__(cls, func_name):
win_error = ctypes.WinError()
def __new__(cls, func_name, error_code=None):
win_error = ctypes.WinError(error_code) #GetLastError by default
api_error = super(Kernel32Error, cls).__new__(cls)
api_error.api_name = func_name
api_error.winerror = win_error.winerror & 0xffffffff
@@ -49,6 +49,9 @@ class Kernel32Error(WindowsError):
api_error.args = (func_name, win_error.winerror, win_error.strerror)
return api_error
def __init__(self, func_name, error_code=None):
super(Kernel32Error, self).__init__(func_name)
def __repr__(self):
return "{0}: {1}".format(self.api_name, super(Kernel32Error, self).__repr__())
@@ -99,6 +102,12 @@ def should_return_zero_check(func_name, result, func, args):
raise Kernel32Error(func_name)
return args
def result_error_code_check(func_name, result, func, args):
"""TODO: DOC"""
if result:
raise Kernel32Error(func_name, error_code=result)
return args
def iphlpapi_error_check(func_name, result, func, args):
"""raise IphlpapiError if result is NOT 0"""
@@ -1282,6 +1291,7 @@ def CreateWellKnownSid(WellKnownSidType, DomainSid=None, pSid=None, cbSid=Needed
GetSidSubAuthorityCount = TransparentAdvapi32Proxy("GetSidSubAuthorityCount")
GetSidSubAuthority = TransparentAdvapi32Proxy("GetSidSubAuthority")
GetLengthSid = TransparentAdvapi32Proxy("GetLengthSid")
EqualSid = TransparentAdvapi32Proxy("EqualSid")
@Advapi32Proxy('GetTokenInformation')
def GetTokenInformation(TokenHandle=NeededParameter, TokenInformationClass=NeededParameter, TokenInformation=None, TokenInformationLength=0, ReturnLength=None):
@@ -1300,12 +1310,12 @@ def RegOpenKeyExA(hKey, lpSubKey, ulOptions, samDesired, phkResult):
# Security stuff
@Advapi32Proxy('GetNamedSecurityInfoA', should_return_zero_check)
@Advapi32Proxy('GetNamedSecurityInfoA', result_error_code_check)
def GetNamedSecurityInfoA(pObjectName, ObjectType, SecurityInfo, ppsidOwner=None, ppsidGroup=None, ppDacl=None, ppSacl=None, ppSecurityDescriptor=None):
return GetNamedSecurityInfoA.ctypes_function(pObjectName, ObjectType, SecurityInfo, ppsidOwner, ppsidGroup, ppDacl, ppSacl, ppSecurityDescriptor)
@Advapi32Proxy('GetNamedSecurityInfoW', should_return_zero_check)
@Advapi32Proxy('GetNamedSecurityInfoW', result_error_code_check)
def GetNamedSecurityInfoW(pObjectName, ObjectType, SecurityInfo, ppsidOwner=None, ppsidGroup=None, ppDacl=None, ppSacl=None, ppSecurityDescriptor=None):
return GetNamedSecurityInfoW.ctypes_function(pObjectName, ObjectType, SecurityInfo, ppsidOwner, ppsidGroup, ppDacl, ppSacl, ppSecurityDescriptor)
@@ -1459,6 +1469,73 @@ def GetNumberOfEventLogRecords(hEventLog, NumberOfRecords):
def CloseEventLog(hEventLog):
return CloseEventLog.ctypes_function(hEventLog)
## Security stuff
## Security stuff
@Advapi32Proxy("IsValidSecurityDescriptor")
def IsValidSecurityDescriptor(pSecurityDescriptor):
return IsValidSecurityDescriptor.ctypes_function(pSecurityDescriptor)
@Advapi32Proxy("ConvertStringSecurityDescriptorToSecurityDescriptorA")
def ConvertStringSecurityDescriptorToSecurityDescriptorA(StringSecurityDescriptor, StringSDRevision, SecurityDescriptor, SecurityDescriptorSize):
return ConvertStringSecurityDescriptorToSecurityDescriptorA.ctypes_function(StringSecurityDescriptor, StringSDRevision, SecurityDescriptor, SecurityDescriptorSize)
@Advapi32Proxy("ConvertStringSecurityDescriptorToSecurityDescriptorW")
def ConvertStringSecurityDescriptorToSecurityDescriptorW(StringSecurityDescriptor, StringSDRevision, SecurityDescriptor, SecurityDescriptorSize):
return ConvertStringSecurityDescriptorToSecurityDescriptorW.ctypes_function(StringSecurityDescriptor, StringSDRevision, SecurityDescriptor, SecurityDescriptorSize)
@Advapi32Proxy("ConvertSecurityDescriptorToStringSecurityDescriptorA")
def ConvertSecurityDescriptorToStringSecurityDescriptorA(SecurityDescriptor, RequestedStringSDRevision, SecurityInformation, StringSecurityDescriptor, StringSecurityDescriptorLen):
return ConvertSecurityDescriptorToStringSecurityDescriptorA.ctypes_function(SecurityDescriptor, RequestedStringSDRevision, SecurityInformation, StringSecurityDescriptor, StringSecurityDescriptorLen)
@Advapi32Proxy("ConvertSecurityDescriptorToStringSecurityDescriptorW")
def ConvertSecurityDescriptorToStringSecurityDescriptorW(SecurityDescriptor, RequestedStringSDRevision, SecurityInformation, StringSecurityDescriptor, StringSecurityDescriptorLen):
return ConvertSecurityDescriptorToStringSecurityDescriptorW.ctypes_function(SecurityDescriptor, RequestedStringSDRevision, SecurityInformation, StringSecurityDescriptor, StringSecurityDescriptorLen)
@Advapi32Proxy("GetSecurityDescriptorDacl")
def GetSecurityDescriptorDacl(pSecurityDescriptor, lpbDaclPresent, pDacl, lpbDaclDefaulted):
return GetSecurityDescriptorDacl.ctypes_function(pSecurityDescriptor, lpbDaclPresent, pDacl, lpbDaclDefaulted)
@Advapi32Proxy("GetAclInformation")
def GetAclInformation(pAcl, pAclInformation, nAclInformationLength, dwAclInformationClass):
return GetAclInformation.ctypes_function(pAcl, pAclInformation, nAclInformationLength, dwAclInformationClass)
@Advapi32Proxy("GetSecurityDescriptorLength")
def GetSecurityDescriptorLength(pSecurityDescriptor):
return GetSecurityDescriptorLength.ctypes_function(pSecurityDescriptor)
@Advapi32Proxy("IsValidSecurityDescriptor")
def IsValidSecurityDescriptor(pSecurityDescriptor):
return IsValidSecurityDescriptor.ctypes_function(pSecurityDescriptor)
@Advapi32Proxy("GetSecurityDescriptorControl")
def GetSecurityDescriptorControl(pSecurityDescriptor, pControl, lpdwRevision):
return GetSecurityDescriptorControl.ctypes_function(pSecurityDescriptor, pControl, lpdwRevision)
@Advapi32Proxy("GetSecurityDescriptorOwner")
def GetSecurityDescriptorOwner(pSecurityDescriptor, pOwner, lpbOwnerDefaulted):
return GetSecurityDescriptorOwner.ctypes_function(pSecurityDescriptor, pOwner, lpbOwnerDefaulted)
@Advapi32Proxy("GetSecurityDescriptorGroup")
def GetSecurityDescriptorGroup(pSecurityDescriptor, pGroup, lpbGroupDefaulted):
return GetSecurityDescriptorGroup.ctypes_function(pSecurityDescriptor, pGroup, lpbGroupDefaulted)
@Advapi32Proxy("GetSecurityDescriptorDacl")
def GetSecurityDescriptorDacl(pSecurityDescriptor, lpbDaclPresent, pDacl, lpbDaclDefaulted):
return GetSecurityDescriptorDacl.ctypes_function(pSecurityDescriptor, lpbDaclPresent, pDacl, lpbDaclDefaulted)
@Advapi32Proxy("GetSecurityDescriptorSacl")
def GetSecurityDescriptorSacl(pSecurityDescriptor, lpbSaclPresent, pSacl, lpbSaclDefaulted):
return GetSecurityDescriptorSacl.ctypes_function(pSecurityDescriptor, lpbSaclPresent, pSacl, lpbSaclDefaulted)
@Advapi32Proxy("GetAce")
def GetAce(pAcl, dwAceIndex, pAce):
return GetAce.ctypes_function(pAcl, dwAceIndex, pAce)
# ##### Iphlpapi (network list and stuff) ###### #
def set_tcp_entry_error_check(func_name, result, func, args):