mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
Added Retn to x86/64 assemblers + encryption_demo sample can now sign a file + [WIP] token improvement
This commit is contained in:
@@ -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')
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
@@ -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("<H", x)
|
||||
except struct.error:
|
||||
raise ImmediatOverflow("16bits unsigned Immediat overflow")
|
||||
|
||||
|
||||
def accept_as_32immediat(x):
|
||||
try:
|
||||
@@ -385,6 +391,18 @@ class Imm16(object):
|
||||
return None, None
|
||||
return (1, BitArray.from_string(imm16), None)
|
||||
|
||||
class UImm16(object):
|
||||
def accept_arg(self, args, instr_state):
|
||||
try:
|
||||
x = int(args[0])
|
||||
except (ValueError, TypeError):
|
||||
return (None, None)
|
||||
try:
|
||||
imm16 = accept_as_unsigned_16immediat(x)
|
||||
except ImmediatOverflow:
|
||||
return None, None
|
||||
return (1, BitArray.from_string(imm16))
|
||||
|
||||
|
||||
class Imm32(object):
|
||||
def accept_arg(self, args, instr_state):
|
||||
@@ -810,7 +828,8 @@ class Xchg(Instruction):
|
||||
|
||||
|
||||
class Ret(Instruction):
|
||||
encoding = [(RawBits.from_int(8, 0xc3),)]
|
||||
encoding = [(RawBits.from_int(8, 0xc3),),
|
||||
(RawBits.from_int(8, 0xc2), UImm16())]
|
||||
|
||||
|
||||
class Int(Instruction):
|
||||
|
||||
@@ -785,7 +785,8 @@ class Cpuid(Instruction):
|
||||
|
||||
|
||||
class Ret(Instruction):
|
||||
encoding = [(RawBits.from_int(8, 0xc3),)]
|
||||
encoding = [(RawBits.from_int(8, 0xc3),),
|
||||
(RawBits.from_int(8, 0xc2), UImm16())]
|
||||
|
||||
|
||||
class ScasB(Instruction):
|
||||
|
||||
+3
-2
@@ -356,6 +356,7 @@ class Acl(gdef.ACL):
|
||||
def __repr__(self):
|
||||
return "<Acl count={0}>".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
|
||||
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user