mirror of
https://github.com/hakril/PythonForWindows
synced 2026-06-08 14:31:45 +00:00
hakril
397 lines
15 KiB
Python
397 lines
15 KiB
Python
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(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)
|