diff --git a/samples/crypto/encryption_demo.py b/samples/crypto/encryption_demo.py index 14c7eff..ddb7b30 100644 --- a/samples/crypto/encryption_demo.py +++ b/samples/crypto/encryption_demo.py @@ -37,6 +37,24 @@ def decrypt(src, pfxfile, password, outfile=None, **kwargs): f.write(decrypted) return decrypted + +def sign(src, pfxfile, password, outfile=None, **kwargs): + if password is None: + password = getpass.getpass() + pfx = crypto.import_pfx(pfxfile.read(), password) + # Decrypt the content of the file + certs = pfx.certs + if len(certs) != 1: + raise ValueError("PFX containing exactly one certificate expected for signature") + cert = certs[0] + signed = crypto.sign(cert, src.read()) + if outfile is None: + print(u"Result = <{0}>".format(signed)) + else: + with open(outfile, "wb") as f: + f.write(signed) + return signed + PFW_TMP_KEY_CONTAINER = "PythonForWindowsTMPContainer" def genkeys(common_name, pfxpassword, outname, keysize=2048, **kwargs): @@ -74,6 +92,7 @@ def genkeys(common_name, pfxpassword, outname, keysize=2048, **kwargs): # Add the newly created certificate to our TMP cert-store cert_store.add_certificate(certif) # Generate a pfx from the TMP cert-store + print("Password is <{0}>".format(pfxpassword)) pfx = gencrypt.generate_pfx(cert_store, pfxpassword) if outname is None: outname = common_name.lower() @@ -103,8 +122,10 @@ def genkeys(common_name, pfxpassword, outname, keysize=2048, **kwargs): ### openssl asn1parse -inform DER -in {file} -parser = argparse.ArgumentParser(prog=__file__) -subparsers = parser.add_subparsers(description='valid subcommands',) + + +parser = argparse.ArgumentParser(prog=__file__, formatter_class=argparse.ArgumentDefaultsHelpFormatter) +subparsers = parser.add_subparsers(description='valid subcommands') cryptparse = subparsers.add_parser('crypt') cryptparse.set_defaults(func=crypt) @@ -121,7 +142,16 @@ decryptparse.add_argument('pfxfile', type=argparse.FileType('rb'), help='PFX fil decryptparse.add_argument('--password', help='Password of the PFX') decryptparse.add_argument('--outfile', default=None, help='The outputfile default is print') -genkeysparse = subparsers.add_parser('genkey') + +decryptparse = subparsers.add_parser('sign') +decryptparse.set_defaults(func=sign) +decryptparse.add_argument('src', type=argparse.FileType('rb'), help='File to sign') +decryptparse.add_argument('pfxfile', type=argparse.FileType('rb'), help='PFX file to use') +decryptparse.add_argument('--password', help='Password of the PFX') +decryptparse.add_argument('--outfile', default=None, help='The outputfile default is print') + + +genkeysparse = subparsers.add_parser('genkey', formatter_class=argparse.ArgumentDefaultsHelpFormatter) 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') diff --git a/tests/test_token.py b/tests/test_token.py new file mode 100644 index 0000000..4aac8c2 --- /dev/null +++ b/tests/test_token.py @@ -0,0 +1,28 @@ +import windows +import windows.generated_def as gdef + + + +def test_token_info(): + token = windows.current_process.token + assert isinstance(token.computername, basestring) + assert isinstance(token.username, basestring) + assert isinstance(token.integrity, (int, long)) + assert isinstance(token.is_elevated, (bool)) + +def test_lower_integrity(proc32): + # Lowering the integrity in remote process + # Because we don't want to mess with the token of our testing process + + proc32.execute_python("import windows") + # We stock the handle becase lowering the integrity + # will mess with token retrieval + proc32.execute_python("token = windows.current_process.token") + proc32.execute_python("token.integrity = 123") + # execute_python will raise this in our own process :) + proc32.execute_python("assert token.integrity == 123") + + +def test_token_elevation(): + tok = windows.current_process.token + assert tok.TokenElevation \ No newline at end of file diff --git a/windows/native_exec/native_function.py b/windows/native_exec/native_function.py index ae34d5a..b0cf654 100644 --- a/windows/native_exec/native_function.py +++ b/windows/native_exec/native_function.py @@ -63,7 +63,8 @@ class CustomAllocator(object): def __init__(self): self.maps = [] - self.get_new_page(0x1000) + self.cur_offset = 0 + self.cur_page_size = 0 # Force get_new_page on first request self.names = [] @classmethod diff --git a/windows/native_exec/simple_x64.py b/windows/native_exec/simple_x64.py index 10e6e25..e6c64dd 100644 --- a/windows/native_exec/simple_x64.py +++ b/windows/native_exec/simple_x64.py @@ -317,6 +317,12 @@ def accept_as_16immediat(x): except struct.error: raise ImmediatOverflow("16bits signed Immediat overflow") +def accept_as_unsigned_16immediat(x): + try: + return struct.pack("".format(self.AceCount) +PAcl = ctypes.POINTER(Acl) # Security descriptor @@ -416,7 +417,7 @@ class SecurityDescriptor(gdef.PSECURITY_DESCRIPTOR): winproxy.GetSecurityDescriptorDacl(self, dacl_present, pdacl, lpbDaclDefaulted) if not dacl_present or not pdacl: return None - return ctypes.cast(pdacl, ctypes.POINTER(Acl))[0] + return ctypes.cast(pdacl, PAcl)[0] @property def sacl(self): @@ -426,7 +427,7 @@ class SecurityDescriptor(gdef.PSECURITY_DESCRIPTOR): winproxy.GetSecurityDescriptorSacl(self, sacl_present, psacl, lpbSaclDefaulted) if not sacl_present or not psacl: return None - return ctypes.cast(psacl, ctypes.POINTER(Acl))[0] + return ctypes.cast(psacl, PAcl)[0] # Constructors @classmethod diff --git a/windows/winobject/token.py b/windows/winobject/token.py new file mode 100644 index 0000000..4c19289 --- /dev/null +++ b/windows/winobject/token.py @@ -0,0 +1,396 @@ +import ctypes + +import windows +from windows import utils +from windows import winproxy +import windows.generated_def as gdef + +import functools + +# Move to windows.security ? +def lookup_privilege(luid, system_name=None): + pass + +# stolen from windows.utils :D (refactor !) +def lookup_privilege_name(privilege_value): + if isinstance(privilege_value, tuple): + luid = LUID(privilege_value[1], privilege_value[0]) + privilege_value = luid + size = DWORD(0x100) + buff = ctypes.c_buffer(size.value) + winproxy.LookupPrivilegeNameA(None, privilege_value, buff, size) + return buff[:size.value] + +KNOW_INTEGRITY_LEVEL = [ +gdef.SECURITY_MANDATORY_UNTRUSTED_RID, +gdef.SECURITY_MANDATORY_LOW_RID, +gdef.SECURITY_MANDATORY_MEDIUM_RID, +gdef.SECURITY_MANDATORY_MEDIUM_PLUS_RID, +gdef.SECURITY_MANDATORY_HIGH_RID, +gdef.SECURITY_MANDATORY_SYSTEM_RID, +gdef.SECURITY_MANDATORY_PROTECTED_PROCESS_RID +] + +know_integrity_level_mapper = gdef.FlagMapper(*KNOW_INTEGRITY_LEVEL) + + +# Voodoo to fix lookup-strangeness in class declaration +def meta_craft(x): + def partial_applier(infos_class, rtype): + return property(functools.partial(x, infos_class=infos_class, rtype=rtype)) + return partial_applier + + +class TokenGroups(gdef.TOKEN_GROUPS): + @property + def _groups(self): + return windows.utils.resized_array(self.Groups, self.GroupCount) + + @property + def sids_and_attributes(self): + return self._groups # Something else ? + + @property + def sids(self): + return [g.Sid for g in self._groups] + + def __repr__(self): + return "<{0} count={1}>".format(type(self).__name__, self.GroupCount) + +TokenGroupsType = TokenGroups # Prevent confusion with token.TokenGroups + +class TokenPrivileges(gdef.TOKEN_PRIVILEGES): + @property + def _privileges(self): + return windows.utils.resized_array(self.Privileges, self.PrivilegeCount) + + @property + def all(self): + return list(self._privileges) + +TokenPrivilegesType = TokenPrivileges + +class TokenSecurityAttributesInformation(gdef.TOKEN_SECURITY_ATTRIBUTES_INFORMATION): + @property + def attributes(self): + tptr = ctypes.cast(self.Attribute.pAttributeV1, ctypes.POINTER(TokenSecurityAttributeV1)) + # Well look like this cast does NOT keep a ref to self. + # Setup the base object ref ourself + tptr._custom_base_ = self + return tptr[:self.AttributeCount] + + +class TokenSecurityAttributeV1(gdef.TOKEN_SECURITY_ATTRIBUTE_V1): + VALUE_ARRAY_PTR_BY_TYPE = { + gdef.TOKEN_SECURITY_ATTRIBUTE_TYPE_INT64: "pInt64", + gdef.TOKEN_SECURITY_ATTRIBUTE_TYPE_UINT64: "pUint64", + gdef.TOKEN_SECURITY_ATTRIBUTE_TYPE_STRING: "pString", + gdef.TOKEN_SECURITY_ATTRIBUTE_TYPE_FQBN: "pFqbn", + # TOKEN_SECURITY_ATTRIBUTE_TYPE_SID + # TOKEN_SECURITY_ATTRIBUTE_TYPE_BOOLEAN + gdef.TOKEN_SECURITY_ATTRIBUTE_TYPE_OCTET_STRING: "pOctetString", + } + + @property + def name(self): + return self.Name.str + + @property + def values(self): + array_name = self.VALUE_ARRAY_PTR_BY_TYPE[self.ValueType] + return getattr(self.Values, array_name)[:self.ValueCount] + + def __repr__(self): + return """<{0} name="{1}">""".format(type(self).__name__, self.name) + + +# https://docs.microsoft.com/en-us/windows/desktop/SecAuthZ/access-tokens +class Token(utils.AutoHandle): + """The token of a process""" + def __init__(self, handle): + self._handle = handle + + def _get_required_token_information_size(self, infos_class): + cbsize = gdef.DWORD() + try: + winproxy.GetTokenInformation(self.handle, infos_class, None, 0, ctypes.byref(cbsize)) + except winproxy.WinproxyError as e: + if not e.winerror in (gdef.ERROR_INSUFFICIENT_BUFFER, gdef.ERROR_BAD_LENGTH): + raise + return cbsize.value + + def _get_token_infomations(self, infos_class, rtype): + required_size = self._get_required_token_information_size(infos_class) + requested_size = max(required_size, ctypes.sizeof(rtype)) + buffer = utils.buffer_type(rtype, 1)(size=requested_size) + cbsize = gdef.DWORD() + winproxy.GetTokenInformation(self.handle, infos_class, buffer, buffer.real_size, cbsize) + return buffer[0] + + + craft = meta_craft(_get_token_infomations) + # https://docs.microsoft.com/en-us/windows/desktop/api/winnt/ne-winnt-_token_information_class + TokenUser = craft(gdef.TokenUser, gdef.TOKEN_USER) + TokenGroups = craft(gdef.TokenGroups , TokenGroupsType) + TokenPrivileges = craft(gdef.TokenPrivileges , TokenPrivilegesType) + TokenOwner = craft(gdef.TokenOwner, gdef.TOKEN_OWNER ) + TokenPrimaryGroup = craft(gdef.TokenPrimaryGroup, gdef.TOKEN_PRIMARY_GROUP) + TokenDefaultDacl = craft(gdef.TokenDefaultDacl, gdef.TOKEN_DEFAULT_DACL ) + TokenSource = craft(gdef.TokenSource, gdef.TOKEN_SOURCE) + TokenType = craft(gdef.TokenType, gdef.TOKEN_TYPE) + TokenImpersonationLevel = craft(gdef.TokenImpersonationLevel, gdef.SECURITY_IMPERSONATION_LEVEL) + TokenStatistics = craft(gdef.TokenStatistics, gdef.TOKEN_STATISTICS) + TokenRestrictedSids = craft(gdef.TokenRestrictedSids, TokenGroupsType) + TokenSessionId = craft(gdef.TokenSessionId, gdef.DWORD) + TokenGroupsAndPrivileges = craft(gdef.TokenGroupsAndPrivileges, gdef.TOKEN_GROUPS_AND_PRIVILEGES) + # TokenSessionReference = craft(gdef.TokenSessionReference, ???) # Reserved. + TokenSandBoxInert = craft(gdef.TokenSandBoxInert, gdef.DWORD) + # TokenAuditPolicy = craft(gdef.TokenAuditPolicy, ???) # Reserved. + TokenOrigin = craft(gdef.TokenOrigin, gdef.TOKEN_ORIGIN) + TokenElevationType = craft(gdef.TokenElevationType, gdef.TOKEN_ELEVATION_TYPE) + TokenLinkedToken = craft(gdef.TokenLinkedToken, gdef.TOKEN_LINKED_TOKEN) + TokenElevation = craft(gdef.TokenElevation, gdef.TOKEN_ELEVATION) + TokenHasRestrictions = craft(gdef.TokenHasRestrictions, gdef.DWORD) + TokenAccessInformation = craft(gdef.TokenAccessInformation, gdef.TOKEN_ACCESS_INFORMATION ) + TokenVirtualizationAllowed = craft(gdef.TokenVirtualizationAllowed, gdef.DWORD) + TokenVirtualizationEnabled = craft(gdef.TokenVirtualizationEnabled, gdef.DWORD) + TokenIntegrityLevel = craft(gdef.TokenIntegrityLevel, gdef.TOKEN_MANDATORY_LABEL ) + TokenUIAccess = craft(gdef.TokenUIAccess, gdef.DWORD) + TokenMandatoryPolicy = craft(gdef.TokenMandatoryPolicy, gdef.TOKEN_MANDATORY_POLICY) + TokenLogonSid = craft(gdef.TokenLogonSid, TokenGroupsType) + TokenIsAppContainer = craft(gdef.TokenIsAppContainer, gdef.DWORD) + TokenCapabilities = craft(gdef.TokenCapabilities, TokenGroupsType) + TokenAppContainerSid = craft(gdef.TokenAppContainerSid, gdef.TOKEN_APPCONTAINER_INFORMATION) + TokenAppContainerNumber = craft(gdef.TokenAppContainerNumber, gdef.DWORD) + TokenUserClaimAttributes = craft(gdef.TokenUserClaimAttributes, gdef.CLAIM_SECURITY_ATTRIBUTES_INFORMATION) + TokenDeviceClaimAttributes = craft(gdef.TokenDeviceClaimAttributes, gdef.CLAIM_SECURITY_ATTRIBUTES_INFORMATION) + # TokenRestrictedUserClaimAttributes = craft(gdef.TokenRestrictedUserClaimAttributes, ???) # Reserved. + # TokenRestrictedDeviceClaimAttributes = craft(gdef.TokenRestrictedDeviceClaimAttributes, ???) # Reserved. + TokenDeviceGroups = craft(gdef.TokenDeviceGroups, TokenGroups) + TokenRestrictedDeviceGroups = craft(gdef.TokenRestrictedDeviceGroups, gdef.TOKEN_GROUPS) + # Reserved. + # Structure found in ntseapi.h (thx internet) + TokenSecurityAttributes = craft(gdef.TokenSecurityAttributes, TokenSecurityAttributesInformation) + # TokenIsRestricted = craft(gdef.TokenIsRestricted, ???) # Reserved. + # TokenProcessTrustLevel = craft(gdef.TokenProcessTrustLevel, ???) # Reserved. + # TokenPrivateNameSpace = craft(gdef.TokenPrivateNameSpace, ???) # Reserved. + # TokenSingletonAttributes = craft(gdef.TokenSingletonAttributes, ???) # Reserved. + # TokenBnoIsolation = craft(gdef.TokenBnoIsolation, ???) # Reserved. + # TokenChildProcessFlags = craft(gdef.TokenChildProcessFlags, ???) # Reserved. + # TokenIsLessPrivilegedAppContainer = craft(gdef.TokenIsLessPrivilegedAppContainer, ???) # Reserved. + + # property arround raw 'GetTokenInformation' + + @property + def user(self): + return self.TokenUser.User.Sid + + groups = TokenGroups # The property not the ctypes Struct + + @property + def owner(self): + return self.TokenOwner.Owner + + @property + def primary_group(self): + return self.TokenPrimaryGroup.PrimaryGroup + + @property + def default_dacl(self): + return self._get_token_infomations(gdef.TokenDefaultDacl, windows.security.PAcl)[0] + + # def source(self): (tok.TokenSource) ?? + + @property + def type(self): + return self.TokenType.value + + @property + def impersonation_level(self): + return self.TokenImpersonationLevel.value + + statistics = TokenStatistics + restricted_sids = TokenRestrictedSids + session_id = TokenSessionId + + @property + def groups_and_privileges(self): + # Return enhanced 'TOKEN_GROUPS_AND_PRIVILEGES' ? + raise NotImplementedError("Token.groups_and_privileges") + + sandbox_inert = TokenSandBoxInert + + # audit_policy + + @property + def origin(self): + # Make a enhanced LUID ? + origin_logon_session = self.TokenOrigin.OriginatingLogonSession + return gdef.ULONG64.from_buffer(origin_logon_session).value + + @property + def elevation_type(self): + return self.TokenElevationType.value + + @property + def linked_token(self): + return Token(self.TokenLinkedToken.LinkedToken) + + @property + def elevation(self): + return bool(self.TokenElevation.TokenIsElevated) + + is_elevated = elevation # Keep this old name ? + has_restriction = TokenHasRestrictions + + @property + def access_information(self): + # Return enhanced subclass ? + raise NotImplementedError("Token.access_information") + # return self.TokenAccessInformation + + virtualization_allowed = TokenVirtualizationAllowed + virtualization_enabled = TokenVirtualizationEnabled + + @property + def integrity_level(self): + # Return SID_AND_ATTRIBUTES ? Only SID ? + return self.TokenIntegrityLevel.Label # SID_AND_ATTRIBUTES + + ui_access = TokenUIAccess + + VALID_TOKEN_POLICIES = gdef.FlagMapper( + gdef.TOKEN_MANDATORY_POLICY_OFF, + gdef.TOKEN_MANDATORY_POLICY_NO_WRITE_UP, + gdef.TOKEN_MANDATORY_POLICY_NEW_PROCESS_MIN, + gdef.TOKEN_MANDATORY_POLICY_VALID_MASK, + ) + + @property + def mandatory_policy(self): + return self.VALID_TOKEN_POLICIES[self.TokenMandatoryPolicy.Policy] + + @property + def logon_sid(self): + rgroups = self.TokenLogonSid + assert rgroups.GroupCount == 1, "More than 1 TokenLogonSid" + return rgroups.Groups[0] + + is_appcontainer = TokenIsAppContainer + capabilities = TokenCapabilities + + @property + def appcontainer_sid(self): + return self.TokenAppContainerSid.TokenAppContainer + + appcontainer_number = TokenAppContainerNumber + + # def security_attribute: + # see PTOKEN_SECURITY_ATTRIBUTES_INFORMATION + # https://github.com/wj32/Backup/blob/59aa77379f9f7ca57f27265e796dd2fe4dae9fab/include/phnt/ntseapi.h + # blablabla :) + + + def duplicate(self, access_rigth=0, attributes=None, impersonation_level=gdef.SecurityImpersonation, toktype=gdef.TokenPrimary): + newtoken = gdef.HANDLE() + winproxy.DuplicateTokenEx(self.handle, access_rigth, attributes, impersonation_level, toktype, newtoken) + return type(self)(newtoken.value) + + + + ### OLD CODE + + def get_integrity(self): + """Return the integrity level of a token + + :type: :class:`int` + """ + buffer_size = self.get_required_information_size(gdef.TokenIntegrityLevel) + buffer = ctypes.c_buffer(buffer_size) + self.get_informations(gdef.TokenIntegrityLevel, buffer) + sid = ctypes.cast(buffer, ctypes.POINTER(gdef.TOKEN_MANDATORY_LABEL))[0].Label.Sid + count = winproxy.GetSidSubAuthorityCount(sid) + integrity = winproxy.GetSidSubAuthority(sid, count[0] - 1)[0] + return know_integrity_level_mapper[integrity] + + def set_integrity(self, integrity): + """Set the integrity level of a token + + :param type: :class:`int` + """ + mandatory_label = gdef.TOKEN_MANDATORY_LABEL() + mandatory_label.Label.Attributes = 0x60 + mandatory_label.Label.Sid = gdef.PSID.from_string("S-1-16-{0}".format(integrity)) + self.set_informations(gdef.TokenIntegrityLevel, mandatory_label) + + integrity = property(get_integrity, set_integrity) + + @property + def is_elevated(self): + """``True`` if process is Admin""" + elevation = gdef.TOKEN_ELEVATION() + self.get_informations(gdef.TokenElevation, elevation) + return bool(elevation.TokenIsElevated) + + @property + def token_user(self): + buffer_size = self.get_required_information_size(gdef.TokenUser) + buffer = ctypes.c_buffer(buffer_size) + self.get_informations(gdef.TokenUser, buffer) + return ctypes.cast(buffer, ctypes.POINTER(gdef.TOKEN_USER))[0] + + @property + def computername(self): + """The computername of the token""" + return self._user_and_computer_name()[1] + + @property + def username(self): + """The username of the token""" + return self._user_and_computer_name()[0] + + + + def _user_and_computer_name(self): + tok_usr = self.token_user + sid = tok_usr.User.Sid + 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, sid, username, usernamesize, computername, computernamesize, peUse) + return username[:usernamesize.value], computername[:computernamesize.value] + + def get_informations(self, info_type, data): + cbsize = gdef.DWORD() + winproxy.GetTokenInformation(self.handle, info_type, ctypes.byref(data), ctypes.sizeof(data), ctypes.byref(cbsize)) + return cbsize.value + + def get_required_information_size(self, info_type): + cbsize = gdef.DWORD() + try: + winproxy.GetTokenInformation(self.handle, info_type, None, 0, ctypes.byref(cbsize)) + except WindowsError as e: + if not e.winerror == gdef.ERROR_INSUFFICIENT_BUFFER: + raise + return cbsize.value + + #TODO: TEST + DOC + def set_informations(self, info_type, infos): + return winproxy.SetTokenInformation(self.handle, info_type, ctypes.byref(infos), ctypes.sizeof(infos)) + + # TEST + def dacl(self): + """TEST CODE""" + buffer_size = self.get_required_information_size(TokenDefaultDacl) + buffer = ctypes.c_buffer(buffer_size) + self.get_informations(gdef.TokenDefaultDacl, buffer) + # TODO: use windows.security.Acl + return ctypes.cast(buffer, POINTER(TOKEN_DEFAULT_DACL))[0] + + + + def __repr__(self): + tid_int = gdef.ULONG64.from_buffer(self.TokenStatistics.TokenId).value + return "<{0} TokenId={1:#x}>".format(type(self).__name__, tid_int)