Handle PermissionError when importing adapter.py

In Windows, if the running process has restricted permissions the process may not have permission to read the file "C:\Users\User\Documents\syskeylog.txt". Sounds silly, but this error can propagate through dependencies and prevent them from complete importing, even if the functionality you expect from them do not depend on this project.

Expected Result

Can import adapter.py

Actual Result

    from pypac.parser import PACFile
  File "C:\Users\User\python_venvs\companion_app312\Lib\site-packages\pypac\__init__.py", line 20, in <module>
    from pypac.api import (
  File "C:\Users\User\python_venvs\companion_app312\Lib\site-packages\pypac\api.py", line 7, in <module>
    import requests
  File "C:\Users\User\python_venvs\companion_app312\Lib\site-packages\requests\__init__.py", line 164, in <module>
    from .api import delete, get, head, options, patch, post, put, request
  File "C:\Users\User\python_venvs\companion_app312\Lib\site-packages\requests\api.py", line 11, in <module>
    from . import sessions
  File "C:\Users\User\python_venvs\companion_app312\Lib\site-packages\requests\sessions.py", line 15, in <module>
    from .adapters import HTTPAdapter
  File "C:\Users\User\python_venvs\companion_app312\Lib\site-packages\requests\adapters.py", line 80, in <module>
    _preloaded_ssl_context = create_urllib3_context()
                             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\User\python_venvs\companion_app312\Lib\site-packages\urllib3\util\ssl_.py", line 349, in create_urllib3_context
    context.keylog_filename = sslkeylogfile
    ^^^^^^^^^^^^^^^^^^^^^^^
PermissionError: [Errno 13] Permission denied: 'C:\\Users\\User\\Documents\\syskeylog.txt'

Reproduction Steps

Run the following code in Windows. You gotta remove a line and un-indent a block. See comment within the code.

import _winapi
import asyncio
import ctypes
import msvcrt
import os
import socket
import subprocess
import sys
import time
import traceback
import winreg
from ctypes import wintypes
from multiprocessing import spawn
from subprocess import Popen

import win32api
import win32con
import win32security
from _ctypes import byref, sizeof

def check_privilege():
    # Get the current process token
    token = win32security.OpenProcessToken(win32api.GetCurrentProcess(), win32con.TOKEN_QUERY)

    # Retrieve the privileges
    privileges = win32security.GetTokenInformation(token, win32security.TokenPrivileges)
    elevation = win32security.GetTokenInformation(token, win32security.TokenElevation)
    print(elevation)
    # Check if the required privilege is present
    for privilege in privileges:
        print(f"{win32security.LookupPrivilegeName(None, privilege[0])} - {privilege[1]}", file=sys.stderr)
    return False

def check_groups():
    token = win32security.OpenProcessToken(win32api.GetCurrentProcess(), win32con.TOKEN_QUERY)

    groups = win32security.GetTokenInformation(token, win32security.TokenGroups)

    for group, enabled in groups:
        name = win32security.LookupAccountSid(None, group)
        print(f"{group} - {name} - {hex(enabled)}", file=sys.stderr)

def _get_token():
    # h_token = win32security.LogonUser(
    #     "nopriv",
    #     "VAN-917843-PC3",
    #     "Stillcreekdr123!",
    #     win32con.LOGON32_LOGON_INTERACTIVE,
    #     win32con.LOGON32_PROVIDER_DEFAULT
    # )

    # gid = _get_fnbi_group()
    # _add_fnbi_group_user()

    system_token = win32security.OpenProcessToken(win32api.GetCurrentProcess(),
                                                  win32con.TOKEN_DUPLICATE
                                                  | win32con.TOKEN_ADJUST_DEFAULT
                                                  | win32con.TOKEN_ADJUST_PRIVILEGES
                                                  | win32con.TOKEN_QUERY
                                                  | win32con.TOKEN_ASSIGN_PRIMARY
                                                  | win32con.TOKEN_ADJUST_GROUPS)
    low_integrity_SID = "S-1-16-4096"
    # https://learn.microsoft.com/en-ca/windows/win32/api/securitybaseapi/nf-securitybaseapi-createrestrictedtoken
    # BOOL CreateRestrictedToken(
    #   [in]           HANDLE               ExistingTokenHandle,
    #   [in]           DWORD                Flags,
    #   [in]           DWORD                DisableSidCount,
    #   [in, optional] PSID_AND_ATTRIBUTES  SidsToDisable,
    #   [in]           DWORD                DeletePrivilegeCount,
    #   [in, optional] PLUID_AND_ATTRIBUTES PrivilegesToDelete,
    #   [in]           DWORD                RestrictedSidCount,
    #   [in, optional] PSID_AND_ATTRIBUTES  SidsToRestrict,
    #   [out]          PHANDLE              NewTokenHandle
    # );

    old_groups = win32security.GetTokenInformation(system_token, win32security.TokenGroups)
    disabled_sids = []
    restricted_sids = []
    enabled_masks = {
        "S-1-1-0", # everyone
        "S-1-5-32-545", # Users - needed for dev only?
        "S-1-5-5", # logon session - dev only
    }
    restricted_masks = {

    }

    def mask_match(entry: str, mask_set):
        for m in mask_set:
            if entry.startswith(m):
                return True
        return False

    for g, state in old_groups:
        sid_str = str(g).split(":")[1]

        if mask_match(sid_str, restricted_masks):
            restricted_sids.append((g, 0))
            continue

        if mask_match(sid_str, enabled_masks):
            continue

        disabled_sids.append((g, 0))

    restricted_token = win32security.CreateRestrictedToken(
        system_token,
        win32security.DISABLE_MAX_PRIVILEGE,  # flags
        # None,
        disabled_sids,  # denied SIDS
        None,  # privs to delete
        restricted_sids  # restricted SIDS
    )

    til = (win32security.CreateWellKnownSid(win32security.WinLowLabelSid), win32security.SE_GROUP_INTEGRITY)
    win32security.SetTokenInformation(restricted_token, win32security.TokenIntegrityLevel, til)

    return restricted_token

class STARTUPINFO(ctypes.Structure):
    _fields_ = [
        ("cb", wintypes.DWORD),  # Size of the structure
        ("lpReserved", wintypes.LPWSTR),  # Reserved; must be NULL
        ("lpDesktop", wintypes.LPWSTR),  # Name of the desktop
        ("lpTitle", wintypes.LPWSTR),  # Title of the window
        ("dwX", wintypes.DWORD),  # X position
        ("dwY", wintypes.DWORD),  # Y position
        ("dwXSize", wintypes.DWORD),  # Width of the window
        ("dwYSize", wintypes.DWORD),  # Height of the window
        ("dwXCountChars", wintypes.DWORD),  # Height of the window
        ("dwYCountChars", wintypes.DWORD),  # Height of the window
        ("dwFillAttribute", wintypes.DWORD),  # Height of the window
        ("dwFlags", wintypes.DWORD),  # Flags
        ("wShowWindow", wintypes.WORD),  # Show window command
        ("cbReserved2", wintypes.WORD),  # Size of the reserved area
        ("lpReserved2", wintypes.LPBYTE),  # Pointer to the reserved area
        ("hStdInput", wintypes.HANDLE),  # Handle to standard input
        ("hStdOutput", wintypes.HANDLE),  # Handle to standard output
        ("hStdError", wintypes.HANDLE)  # Handle to standard error
    ]

class STARTUPINFOEX(ctypes.Structure):
    _fields_ = [
        ("StartupInfo", STARTUPINFO),
        ("lpAttributeList", wintypes.LPVOID)
    ]

# Define PROCESS_INFORMATION structure
class PROCESS_INFORMATION(ctypes.Structure):
    _fields_ = [
        ("hProcess", wintypes.HANDLE),
        ("hThread", wintypes.HANDLE),
        ("dwProcessId", wintypes.DWORD),
        ("dwThreadId", wintypes.DWORD)
    ]

# Load the function
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
advapi32 = ctypes.WinDLL('advapi32', use_last_error=True)
libc = ctypes.CDLL("msvcrt.dll")

# Define SetHandleInformation function
SetHandleInformation = kernel32.SetHandleInformation
SetHandleInformation.argtypes = [
    ctypes.wintypes.HANDLE,  # hObject
    ctypes.wintypes.DWORD,  # dwMask
    ctypes.wintypes.DWORD  # dwFlags
]
SetHandleInformation.restype = ctypes.wintypes.BOOL

CreateProcessW = kernel32.CreateProcessW
CreateProcessAsUser = advapi32.CreateProcessAsUserW
CREATE_UNICODE_ENVIRONMENT = 0x00000400
PROC_THREAD_ATTRIBUTE_LIST = 0x00020000
PROC_THREAD_ATTRIBUTE_HANDLE_LIST = 0x00020004
EXTENDED_STARTUPINFO_PRESENT = 0x00080000
HANDLE_FLAG_INHERIT = 0x00000001
HANDLE_FLAG_PROTECT_FROM_CLOSE = 0x00000002

def get_last_error_message():
    error_code = ctypes.get_last_error()
    if error_code == 0:
        return "No error."

    message_buffer = ctypes.create_string_buffer(256)
    message_length = kernel32.FormatMessageA(
        0x00000001 | 0x00001000,  # FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_ALLOCATE_BUFFER
        None,
        error_code,
        0,
        byref(message_buffer),
        ctypes.sizeof(message_buffer),
        None
    )

    if message_length == 0:
        return f"Unknown error code: {error_code}"
    msg = message_buffer.value.decode('ansi').replace('\r', '').replace('\n', '')
    return msg

foo = None

# Function to get Windows handle from file descriptor
def get_windows_handle(file):
    # Get the file descriptor
    fd = file.fileno()

    # Use _get_osfhandle to get the Windows handle
    handle = ctypes.windll.kernel32._get_osfhandle(fd)

    if handle == -1:
        raise ValueError("Could not get the Windows handle.")

    return handle

def make_handle_inheritable(handle):
    if not SetHandleInformation(handle, HANDLE_FLAG_INHERIT, HANDLE_FLAG_INHERIT):
        raise ctypes.WinError(ctypes.get_last_error())

def writer_main():
    def CreateProcess(cmd, cmdline, proc_attr, thread_attr, inherit_handles, creation_flags, current_directory,
                      environment, startup_info):
        # change_privilege(win32con.SE_PRIVILEGE_ENABLED)
        # check_privilege()

        token = _get_token()
        # CreateProcessAsUser
        #   0 [in, optional]      HANDLE                hToken,
        #   1 [in, optional]      LPCSTR                lpApplicationName,
        #   2 [in, out, optional] LPSTR                 lpCommandLine,
        #   3 [in, optional]      LPSECURITY_ATTRIBUTES lpProcessAttributes,
        #   4 [in, optional]      LPSECURITY_ATTRIBUTES lpThreadAttributes,
        #   5 [in]                BOOL                  bInheritHandles,
        #   6 [in]                DWORD                 dwCreationFlags,
        #   7 [in, optional]      LPVOID                lpEnvironment,
        #   8 [in, optional]      LPCSTR                lpCurrentDirectory,
        #   9 [in]                LPSTARTUPINFOA        lpStartupInfo,

        # CreateProcess
        #   0 [in, optional]      LPCSTR                lpApplicationName,
        #   1 [in, out, optional] LPSTR                 lpCommandLine,
        #   2 [in, optional]      LPSECURITY_ATTRIBUTES lpProcessAttributes,
        #   3 [in, optional]      LPSECURITY_ATTRIBUTES lpThreadAttributes,
        #   4 [in]                BOOL                  bInheritHandles,
        #   5 [in]                DWORD                 dwCreationFlags,
        #   6 [in, optional]      LPVOID                lpEnvironment,
        #   7 [in, optional]      LPCSTR                lpCurrentDirectory,
        #   8 [in]                LPSTARTUPINFOA        lpStartupInfo,
        six = STARTUPINFOEX()
        si = six.StartupInfo
        si.cb = ctypes.sizeof(STARTUPINFOEX)
        attribute_list_size = wintypes.DWORD()

        kernel32.InitializeProcThreadAttributeList(None, 1, 0, byref(attribute_list_size))

        buffer = (wintypes.BYTE * (attribute_list_size.value + 10))()
        six.lpAttributeList = ctypes.cast(buffer, wintypes.LPVOID)
        if not six.lpAttributeList:
            print(f"HeapAlloc failed with error: {get_last_error_message()}")
            raise RuntimeError()

        if not kernel32.InitializeProcThreadAttributeList(
                byref(buffer),
                1,
                0,
                byref(attribute_list_size)
        ):
            print(f"InitializeProcThreadAttributeList failed with error: {get_last_error_message()}")
            raise RuntimeError()
        if startup_info is not None:
            si.dwFlags |= win32con.STARTF_USESTDHANDLES  # Enable standard handle redirection
            si.hStdOutput = getattr(startup_info, "hStdOutput")  # Redirect standard output
            si.hStdInput = getattr(startup_info, "hStdInput")  # Redirect standard error
            si.hStdError = getattr(startup_info, "hStdError")  # Redirect standard error
            handle_list = [wintypes.HANDLE(si.hStdOutput), wintypes.HANDLE(si.hStdInput), wintypes.HANDLE(si.hStdError)]
            handle_count = len(handle_list)
            handles_array = (wintypes.HANDLE * handle_count)(*handle_list)  # Create an array of handles
            ctypes.windll.kernel32.UpdateProcThreadAttribute(
                buffer,
                0,
                PROC_THREAD_ATTRIBUTE_HANDLE_LIST,
                handles_array,
                handle_count * sizeof(wintypes.HANDLE),
                None,
                None
            )

        pi = PROCESS_INFORMATION()
        success = advapi32.CreateProcessAsUserW(
            token.handle,
            python_exe, cmdline, proc_attr, thread_attr, inherit_handles,
            creation_flags | CREATE_UNICODE_ENVIRONMENT,
            environment, current_directory, byref(six), byref(pi)
        )

        # return TemporaryLibPatch.execute_unpatched(CreateProcess,
        #     python_exe, cmdline, proc_attr, thread_attr, inherit_handles,
        #     creation_flags,
        #     environment, current_directory, startup_info
        # )

        if not success:
            print(f"CreateProcessAsUser failed with error: {get_last_error_message()}")
            raise RuntimeError()

        return pi.hProcess, pi.hThread, pi.dwProcessId, pi.dwThreadId,

    file = open(__file__, 'r')
    handle = msvcrt.get_osfhandle(file.fileno())
    make_handle_inheritable(handle)

    _winapi.CreateProcess = CreateProcess
    # You can remove the context manager line below. It just wraps around the line above. The Github editor doesn't let me easily remove the indentation of the block....
    with TemporaryLibPatch(_winapi, CreateProcess): 
        def _path_eq(p1, p2):
            return p1 == p2 or os.path.normcase(p1) == os.path.normcase(p2)

        WINENV = _path_eq(sys.executable, sys._base_executable)

        python_exe = spawn.get_executable()
        set_pylauncher = False

        # FROM multiprocessing module:
        # # bpo-35797: When running in a venv, we bypass the redirect

        if WINENV and _path_eq(python_exe, sys.executable):
            python_exe = sys._base_executable
            set_pylauncher = True
            os.environ["__PYVENV_LAUNCHER__"] = sys.executable

        proc = Popen([
            python_exe,
            __file__,
            "--read",
        ], stdout=subprocess.PIPE,
            stdin=subprocess.PIPE,
            shell=False,
            close_fds=False
        )

        if set_pylauncher:
            os.unsetenv("__PYVENV_LAUNCHER__")     
        proc.wait()

def reader_main():
     import requests

if __name__ == '__main__':
    if len(sys.argv) > 1 and sys.argv[1] == "--read":
        reader_main()
    else:
        writer_main()

System Information

$ python -m requests.help

{
  "chardet": {
    "version": null
  },
  "charset_normalizer": {
    "version": "3.3.2"
  },
  "cryptography": {
    "version": ""
  },
  "idna": {
    "version": "3.8"
  },
  "implementation": {
    "name": "CPython",
    "version": "3.12.5"
  },
  "platform": {
    "release": "10",
    "system": "Windows"
  },
  "pyOpenSSL": {
    "openssl_version": "",
    "version": null
  },
  "requests": {
    "version": "2.32.3"
  },
  "system_ssl": {
    "version": "300000d0"
  },
  "urllib3": {
    "version": "2.2.2"
  },
  "using_charset_normalizer": true,
  "using_pyopenssl": false
}

psf / requests

Handle PermissionError when importing adapter.py #6815

Expected Result

Actual Result

Reproduction Steps

System Information