Da

✏️ Editing: _winapi.py

"""
Windows API functions implemented as ctypes functions and classes as found
in jaraco.windows (3.4.1).

If you encounter issues with this module, please consider reporting the issues
in jaraco.windows and asking the author to port the fixes back here.
"""

import builtins
import ctypes.wintypes

from paramiko.util import u

######################
# jaraco.windows.error

def format_system_message(errno):
    """
    Call FormatMessage with a system error number to retrieve
    the descriptive error message.
    """
    # first some flags used by FormatMessageW
    ALLOCATE_BUFFER = 0x100
    FROM_SYSTEM = 0x1000

# Let FormatMessageW allocate the buffer (we'll free it below)
    # Also, let it know we want a system error message.
    flags = ALLOCATE_BUFFER | FROM_SYSTEM
    source = None
    message_id = errno
    language_id = 0
    result_buffer = ctypes.wintypes.LPWSTR()
    buffer_size = 0
    arguments = None
    bytes = ctypes.windll.kernel32.FormatMessageW(
        flags,
        source,
        message_id,
        language_id,
        ctypes.byref(result_buffer),
        buffer_size,
        arguments,
    )
    # note the following will cause an infinite loop if GetLastError
    #  repeatedly returns an error that cannot be formatted, although
    #  this should not happen.
    handle_nonzero_success(bytes)
    message = result_buffer.value
    ctypes.windll.kernel32.LocalFree(result_buffer)
    return message

class WindowsError(builtins.WindowsError):
    """more info about errors at
    http://msdn.microsoft.com/en-us/library/ms681381(VS.85).aspx"""

def __init__(self, value=None):
        if value is None:
            value = ctypes.windll.kernel32.GetLastError()
        strerror = format_system_message(value)
        args = 0, strerror, None, value
        super().__init__(*args)

@property
    def message(self):
        return self.strerror

@property
    def code(self):
        return self.winerror

def __str__(self):
        return self.message

def __repr__(self):
        return "{self.__class__.__name__}({self.winerror})".format(**vars())

def handle_nonzero_success(result):
    if result == 0:
        raise WindowsError()

###########################
# jaraco.windows.api.memory

GMEM_MOVEABLE = 0x2

GlobalAlloc = ctypes.windll.kernel32.GlobalAlloc
GlobalAlloc.argtypes = ctypes.wintypes.UINT, ctypes.c_size_t
GlobalAlloc.restype = ctypes.wintypes.HANDLE

GlobalLock = ctypes.windll.kernel32.GlobalLock
GlobalLock.argtypes = (ctypes.wintypes.HGLOBAL,)
GlobalLock.restype = ctypes.wintypes.LPVOID

GlobalUnlock = ctypes.windll.kernel32.GlobalUnlock
GlobalUnlock.argtypes = (ctypes.wintypes.HGLOBAL,)
GlobalUnlock.restype = ctypes.wintypes.BOOL

GlobalSize = ctypes.windll.kernel32.GlobalSize
GlobalSize.argtypes = (ctypes.wintypes.HGLOBAL,)
GlobalSize.restype = ctypes.c_size_t

CreateFileMapping = ctypes.windll.kernel32.CreateFileMappingW
CreateFileMapping.argtypes = [
    ctypes.wintypes.HANDLE,
    ctypes.c_void_p,
    ctypes.wintypes.DWORD,
    ctypes.wintypes.DWORD,
    ctypes.wintypes.DWORD,
    ctypes.wintypes.LPWSTR,
]
CreateFileMapping.restype = ctypes.wintypes.HANDLE

MapViewOfFile = ctypes.windll.kernel32.MapViewOfFile
MapViewOfFile.restype = ctypes.wintypes.HANDLE

UnmapViewOfFile = ctypes.windll.kernel32.UnmapViewOfFile
UnmapViewOfFile.argtypes = (ctypes.wintypes.HANDLE,)

RtlMoveMemory = ctypes.windll.kernel32.RtlMoveMemory
RtlMoveMemory.argtypes = (ctypes.c_void_p, ctypes.c_void_p, ctypes.c_size_t)

ctypes.windll.kernel32.LocalFree.argtypes = (ctypes.wintypes.HLOCAL,)

#####################
# jaraco.windows.mmap

class MemoryMap:
    """
    A memory map object which can have security attributes overridden.
    """

def __init__(self, name, length, security_attributes=None):
        self.name = name
        self.length = length
        self.security_attributes = security_attributes
        self.pos = 0

def __enter__(self):
        p_SA = (
            ctypes.byref(self.security_attributes)
            if self.security_attributes
            else None
        )
        INVALID_HANDLE_VALUE = -1
        PAGE_READWRITE = 0x4
        FILE_MAP_WRITE = 0x2
        filemap = ctypes.windll.kernel32.CreateFileMappingW(
            INVALID_HANDLE_VALUE,
            p_SA,
            PAGE_READWRITE,
            0,
            self.length,
            u(self.name),
        )
        handle_nonzero_success(filemap)
        if filemap == INVALID_HANDLE_VALUE:
            raise Exception("Failed to create file mapping")
        self.filemap = filemap
        self.view = MapViewOfFile(filemap, FILE_MAP_WRITE, 0, 0, 0)
        return self

def seek(self, pos):
        self.pos = pos

def write(self, msg):
        assert isinstance(msg, bytes)
        n = len(msg)
        if self.pos + n >= self.length:  # A little safety.
            raise ValueError(f"Refusing to write {n} bytes")
        dest = self.view + self.pos
        length = ctypes.c_size_t(n)
        ctypes.windll.kernel32.RtlMoveMemory(dest, msg, length)
        self.pos += n

def read(self, n):
        """
        Read n bytes from mapped view.
        """
        out = ctypes.create_string_buffer(n)
        source = self.view + self.pos
        length = ctypes.c_size_t(n)
        ctypes.windll.kernel32.RtlMoveMemory(out, source, length)
        self.pos += n
        return out.raw

def __exit__(self, exc_type, exc_val, tb):
        ctypes.windll.kernel32.UnmapViewOfFile(self.view)
        ctypes.windll.kernel32.CloseHandle(self.filemap)

#############################
# jaraco.windows.api.security

# from WinNT.h
READ_CONTROL = 0x00020000
STANDARD_RIGHTS_REQUIRED = 0x000F0000
STANDARD_RIGHTS_READ = READ_CONTROL
STANDARD_RIGHTS_WRITE = READ_CONTROL
STANDARD_RIGHTS_EXECUTE = READ_CONTROL
STANDARD_RIGHTS_ALL = 0x001F0000

# from NTSecAPI.h
POLICY_VIEW_LOCAL_INFORMATION = 0x00000001
POLICY_VIEW_AUDIT_INFORMATION = 0x00000002
POLICY_GET_PRIVATE_INFORMATION = 0x00000004
POLICY_TRUST_ADMIN = 0x00000008
POLICY_CREATE_ACCOUNT = 0x00000010
POLICY_CREATE_SECRET = 0x00000020
POLICY_CREATE_PRIVILEGE = 0x00000040
POLICY_SET_DEFAULT_QUOTA_LIMITS = 0x00000080
POLICY_SET_AUDIT_REQUIREMENTS = 0x00000100
POLICY_AUDIT_LOG_ADMIN = 0x00000200
POLICY_SERVER_ADMIN = 0x00000400
POLICY_LOOKUP_NAMES = 0x00000800
POLICY_NOTIFICATION = 0x00001000

POLICY_READ = (
    STANDARD_RIGHTS_READ
    | POLICY_VIEW_AUDIT_INFORMATION
    | POLICY_GET_PRIVATE_INFORMATION
)

POLICY_EXECUTE = (
    STANDARD_RIGHTS_EXECUTE
    | POLICY_VIEW_LOCAL_INFORMATION
    | POLICY_LOOKUP_NAMES
)

class TokenAccess:
    TOKEN_QUERY = 0x8

class TokenInformationClass:
    TokenUser = 1

class TOKEN_USER(ctypes.Structure):
    num = 1
    _fields_ = [
        ("SID", ctypes.c_void_p),
        ("ATTRIBUTES", ctypes.wintypes.DWORD),
    ]

class SECURITY_DESCRIPTOR(ctypes.Structure):
    """
    typedef struct _SECURITY_DESCRIPTOR
        {
        UCHAR Revision;
        UCHAR Sbz1;
        SECURITY_DESCRIPTOR_CONTROL Control;
        PSID Owner;
        PSID Group;
        PACL Sacl;
        PACL Dacl;
        }   SECURITY_DESCRIPTOR;
    """

SECURITY_DESCRIPTOR_CONTROL = ctypes.wintypes.USHORT
    REVISION = 1

_fields_ = [
        ("Revision", ctypes.c_ubyte),
        ("Sbz1", ctypes.c_ubyte),
        ("Control", SECURITY_DESCRIPTOR_CONTROL),
        ("Owner", ctypes.c_void_p),
        ("Group", ctypes.c_void_p),
        ("Sacl", ctypes.c_void_p),
        ("Dacl", ctypes.c_void_p),
    ]

class SECURITY_ATTRIBUTES(ctypes.Structure):
    """
    typedef struct _SECURITY_ATTRIBUTES {
        DWORD  nLength;
        LPVOID lpSecurityDescriptor;
        BOOL   bInheritHandle;
    } SECURITY_ATTRIBUTES;
    """

_fields_ = [
        ("nLength", ctypes.wintypes.DWORD),
        ("lpSecurityDescriptor", ctypes.c_void_p),
        ("bInheritHandle", ctypes.wintypes.BOOL),
    ]

def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.nLength = ctypes.sizeof(SECURITY_ATTRIBUTES)

@property
    def descriptor(self):
        return self._descriptor

@descriptor.setter
    def descriptor(self, value):
        self._descriptor = value
        self.lpSecurityDescriptor = ctypes.addressof(value)

ctypes.windll.advapi32.SetSecurityDescriptorOwner.argtypes = (
    ctypes.POINTER(SECURITY_DESCRIPTOR),
    ctypes.c_void_p,
    ctypes.wintypes.BOOL,
)

#########################
# jaraco.windows.security

def GetTokenInformation(token, information_class):
    """
    Given a token, get the token information for it.
    """
    data_size = ctypes.wintypes.DWORD()
    ctypes.windll.advapi32.GetTokenInformation(
        token, information_class.num, 0, 0, ctypes.byref(data_size)
    )
    data = ctypes.create_string_buffer(data_size.value)
    handle_nonzero_success(
        ctypes.windll.advapi32.GetTokenInformation(
            token,
            information_class.num,
            ctypes.byref(data),
            ctypes.sizeof(data),
            ctypes.byref(data_size),
        )
    )
    return ctypes.cast(data, ctypes.POINTER(TOKEN_USER)).contents

def OpenProcessToken(proc_handle, access):
    result = ctypes.wintypes.HANDLE()
    proc_handle = ctypes.wintypes.HANDLE(proc_handle)
    handle_nonzero_success(
        ctypes.windll.advapi32.OpenProcessToken(
            proc_handle, access, ctypes.byref(result)
        )
    )
    return result

def get_current_user():
    """
    Return a TOKEN_USER for the owner of this process.
    """
    process = OpenProcessToken(
        ctypes.windll.kernel32.GetCurrentProcess(), TokenAccess.TOKEN_QUERY
    )
    return GetTokenInformation(process, TOKEN_USER)

def get_security_attributes_for_user(user=None):
    """
    Return a SECURITY_ATTRIBUTES structure with the SID set to the
    specified user (uses current user if none is specified).
    """
    if user is None:
        user = get_current_user()

assert isinstance(user, TOKEN_USER), "user must be TOKEN_USER instance"

SD = SECURITY_DESCRIPTOR()
    SA = SECURITY_ATTRIBUTES()
    # by attaching the actual security descriptor, it will be garbage-
    # collected with the security attributes
    SA.descriptor = SD
    SA.bInheritHandle = 1

ctypes.windll.advapi32.InitializeSecurityDescriptor(
        ctypes.byref(SD), SECURITY_DESCRIPTOR.REVISION
    )
    ctypes.windll.advapi32.SetSecurityDescriptorOwner(
        ctypes.byref(SD), user.SID, 0
    )
    return SA

📁 Sid Gifari File Manager