diff --git a/README.md b/README.md index a292bb2..ea37187 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/tests/test_security.py b/tests/test_security.py new file mode 100644 index 0000000..9c2f9ea --- /dev/null +++ b/tests/test_security.py @@ -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", "") + + diff --git a/windows/security.py b/windows/security.py new file mode 100644 index 0000000..9b18e73 --- /dev/null +++ b/windows/security.py @@ -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 "".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) + diff --git a/windows/winproxy.py b/windows/winproxy.py index 88ca827..f3c3f9e 100644 --- a/windows/winproxy.py +++ b/windows/winproxy.py @@ -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):