hermes-agent/gateway/runtime_bootstrap_windows.py

"""Native local-only bootstrap pipes with cancellable overlapped I/O.

The pipe worker threads do not own session state. Handlers marshal authority
operations to the gateway loop; the ticket store itself is thread-safe.
"""
from __future__ import annotations

import ctypes
from ctypes import wintypes
import logging
import os
from pathlib import Path
import threading
import time

logger = logging.getLogger(__name__)


def _native():
    if os.name != 'nt':
        raise OSError('native Windows bootstrap requires Windows')
    import _winapi
    return _winapi


def _api(dll, name, result, args):
    fn = getattr(dll, name)
    fn.restype, fn.argtypes = result, args
    return fn


def _token_sid(token) -> str:
    """String SID of a token's user (``TokenUser``)."""
    a = ctypes.WinDLL('advapi32', use_last_error=True)
    length = wintypes.DWORD()
    token_info = _api(a, 'GetTokenInformation', wintypes.BOOL,
                      [wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD, ctypes.POINTER(wintypes.DWORD)])
    token_info(token, 1, None, 0, ctypes.byref(length))
    buf = ctypes.create_string_buffer(length.value)
    if not token_info(token, 1, buf, length, ctypes.byref(length)):
        raise PermissionError('cannot read bootstrap peer identity')
    return _sid_text(ctypes.cast(buf, ctypes.POINTER(ctypes.c_void_p))[0])


def _sid_text(sid) -> str:
    """String form (``S-1-...``) of the SID at *sid*."""
    k = ctypes.WinDLL('kernel32', use_last_error=True)
    a = ctypes.WinDLL('advapi32', use_last_error=True)
    text = wintypes.LPWSTR()
    convert = _api(a, 'ConvertSidToStringSidW', wintypes.BOOL,
                   [ctypes.c_void_p, ctypes.POINTER(wintypes.LPWSTR)])
    if not convert(sid, ctypes.byref(text)):
        raise PermissionError('cannot encode bootstrap peer identity')
    try:
        return text.value
    finally:
        _api(k, 'LocalFree', ctypes.c_void_p, [ctypes.c_void_p])(ctypes.cast(text, ctypes.c_void_p))


def _process_sid(pid: int) -> str:
    k = ctypes.WinDLL('kernel32', use_last_error=True)
    a = ctypes.WinDLL('advapi32', use_last_error=True)
    open_process = _api(k, 'OpenProcess', wintypes.HANDLE, [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD])
    close = _api(k, 'CloseHandle', wintypes.BOOL, [wintypes.HANDLE])
    process = open_process(0x1000, False, pid)
    if not process:
        raise PermissionError('cannot verify bootstrap peer process')
    token = wintypes.HANDLE()
    try:
        open_token = _api(a, 'OpenProcessToken', wintypes.BOOL,
                          [wintypes.HANDLE, wintypes.DWORD, ctypes.POINTER(wintypes.HANDLE)])
        if not open_token(process, 8, ctypes.byref(token)):
            raise PermissionError('cannot verify bootstrap peer token')
        return _token_sid(token)
    finally:
        if token:
            close(token)
        close(process)


class _AclHeader(ctypes.Structure):
    _fields_ = [('revision', ctypes.c_ubyte), ('sbz1', ctypes.c_ubyte), ('size', wintypes.WORD),
                ('count', wintypes.WORD), ('sbz2', wintypes.WORD)]


def _pipe_security(handle) -> tuple[str, tuple[tuple[int, str], ...]]:
    """The pipe object's owner SID and its DACL as ``(ace_type, sid)`` pairs, read from *handle*
    (``GetSecurityInfo``; the client's GENERIC_READ carries READ_CONTROL). Unlike ``OpenProcess`` on
    the server's PID this needs no access to the server process, so it still answers when the
    server is elevated and this client is not (#140706)."""
    k = ctypes.WinDLL('kernel32', use_last_error=True)
    a = ctypes.WinDLL('advapi32', use_last_error=True)
    get_info = _api(a, 'GetSecurityInfo', wintypes.DWORD,
                    [wintypes.HANDLE, ctypes.c_int, wintypes.DWORD] + [ctypes.POINTER(ctypes.c_void_p)] * 5)
    get_ace = _api(a, 'GetAce', wintypes.BOOL, [ctypes.c_void_p, wintypes.DWORD, ctypes.POINTER(ctypes.c_void_p)])
    owner, dacl, descriptor = ctypes.c_void_p(), ctypes.c_void_p(), ctypes.c_void_p()
    # SE_KERNEL_OBJECT; OWNER_SECURITY_INFORMATION | DACL_SECURITY_INFORMATION
    if get_info(handle, 6, 0x1 | 0x4, ctypes.byref(owner), None, ctypes.byref(dacl), None,
                ctypes.byref(descriptor)):
        raise PermissionError('cannot read named-pipe owner')
    try:
        if not owner:  # inside the try: the descriptor is freed on this refusal too
            raise PermissionError('cannot read named-pipe owner')
        aces = []
        # A NULL DACL (no entries to read) grants everyone: it is reported as no ACEs at all.
        for index in range(_AclHeader.from_address(dacl.value).count if dacl else 0):
            ace = ctypes.c_void_p()
            if not get_ace(dacl, index, ctypes.byref(ace)):
                raise PermissionError('cannot read named-pipe DACL')
            kind = ctypes.c_ubyte.from_address(ace.value).value
            # ACCESS_ALLOWED_ACE / ACCESS_DENIED_ACE: 4-byte ACE_HEADER, 4-byte mask, then the SID.
            aces.append((kind, _sid_text(ace.value + 8) if kind in (0, 1) else ''))
        return _sid_text(owner.value), tuple(aces)
    finally:
        _api(k, 'LocalFree', ctypes.c_void_p, [ctypes.c_void_p])(descriptor)


def _pipe_server_session(handle) -> int | None:
    """The pipe server's logon session, from the pipe (no access to the server process needed)."""
    k = ctypes.WinDLL('kernel32', use_last_error=True)
    session = wintypes.ULONG()
    get = _api(k, 'GetNamedPipeServerSessionId', wintypes.BOOL, [wintypes.HANDLE, ctypes.POINTER(wintypes.ULONG)])
    return session.value if get(handle, ctypes.byref(session)) else None


class _ImpersonationLeak(RuntimeError):
    """RevertToSelf failed: this thread may still carry the client's identity. Deliberately not an
    OSError, so the serving loop does not swallow it and the worker stops instead."""


def _pipe_client_sid(handle) -> str:
    """The client's SID from the identification token the kernel captured when it opened the pipe
    (the client dials with ``SECURITY_SQOS_PRESENT``). Unlike ``OpenProcess`` on the client's PID,
    this still answers after a short-lived client (a one-shot ``gateway ensure``) has exited.
    Windows only allows it once data has been read from the pipe."""
    k = ctypes.WinDLL('kernel32', use_last_error=True)
    a = ctypes.WinDLL('advapi32', use_last_error=True)
    impersonate = _api(a, 'ImpersonateNamedPipeClient', wintypes.BOOL, [wintypes.HANDLE])
    revert = _api(a, 'RevertToSelf', wintypes.BOOL, [])
    open_thread_token = _api(a, 'OpenThreadToken', wintypes.BOOL,
                             [wintypes.HANDLE, wintypes.DWORD, wintypes.BOOL, ctypes.POINTER(wintypes.HANDLE)])
    current_thread = _api(k, 'GetCurrentThread', wintypes.HANDLE, [])
    close = _api(k, 'CloseHandle', wintypes.BOOL, [wintypes.HANDLE])
    if not impersonate(handle):
        raise PermissionError('cannot impersonate named-pipe peer')
    token = wintypes.HANDLE()
    try:
        try:
            # OpenAsSelf: the access check runs against the gateway's own token, not the client's.
            if not open_thread_token(current_thread(), 8, True, ctypes.byref(token)):
                raise PermissionError('cannot read named-pipe peer token')
        finally:
            if not revert():
                raise _ImpersonationLeak('RevertToSelf failed after reading a named-pipe peer')
        return _token_sid(token)
    finally:
        if token:
            close(token)


def _peer_subject(handle, *, server: bool) -> str:
    """``sid:<owner>`` for a same-user peer; PermissionError for another user or an unidentified one.

    The serving side (``server=False``) reads the identity from the pipe's client token first and
    falls back to the client's process; an unreadable peer is refused, never served as anonymous.
    The client side (``server=True``) is judged by :func:`judge_pipe_server`."""
    sid = None
    if not server:
        try:
            sid = _pipe_client_sid(handle)
        except PermissionError:
            sid = None
    if sid is None:
        k = ctypes.WinDLL('kernel32', use_last_error=True)
        name = 'GetNamedPipeServerProcessId' if server else 'GetNamedPipeClientProcessId'
        get_pid = _api(k, name, wintypes.BOOL, [wintypes.HANDLE, ctypes.POINTER(wintypes.ULONG)])
        pid = wintypes.ULONG()
        if not get_pid(handle, ctypes.byref(pid)):
            raise PermissionError('cannot identify named-pipe peer')
        if server:
            return _server_subject(handle, pid.value)
        sid = _process_sid(pid.value)
    if sid != _process_sid(os.getpid()):
        raise PermissionError('named-pipe peer belongs to another user')
    return 'sid:' + sid


def _disconnect_pipe(handle) -> None:
    """``DisconnectNamedPipe`` via kernel32: CPython's ``_winapi`` does not export it, and the
    server thread must outlive its first client (a dead pipe leaves the gateway stuck in
    ``starting`` for every later ``identify``)."""
    k = ctypes.WinDLL('kernel32', use_last_error=True)
    disconnect = _api(k, 'DisconnectNamedPipe', wintypes.BOOL, [wintypes.HANDLE])
    if not disconnect(handle):
        error = ctypes.get_last_error()
        raise OSError(None, 'DisconnectNamedPipe failed', None, error)


def _remaining_ms(deadline):
    remaining = deadline - time.monotonic()
    if remaining <= 0:
        raise TimeoutError('runtime control deadline exceeded')
    return max(1, int(remaining * 1000))


def _complete(ov, deadline):
    win = _native()
    try:
        if win.WaitForSingleObject(ov.event, _remaining_ms(deadline)) != win.WAIT_OBJECT_0:
            raise TimeoutError('runtime control deadline exceeded')
        transferred, error = ov.GetOverlappedResult(False)
        if error:
            raise OSError(error, 'runtime pipe I/O failed')
        return transferred
    except BaseException:
        # Cancel the native operation, not merely its caller's wait. Completion is
        # reaped before the buffer/OVERLAPPED can be released.
        ov.cancel()
        ov.GetOverlappedResult(True)
        raise


def _accept(handle, budget) -> bool:
    """Wait up to ``budget`` seconds for a client; True once one is connected.

    A client can connect between the wait timing out and the cancel reaching the kernel; the
    reaped result then reports success and that client is served. ERROR_OPERATION_ABORTED (995)
    is the only idle outcome. A client that is already connected completes the accept at once.
    """
    win = _native()
    ov = win.ConnectNamedPipe(handle, overlapped=True)
    if win.WaitForSingleObject(ov.event, max(1, int(budget * 1000))) != win.WAIT_OBJECT_0:
        ov.cancel()
    _, error = ov.GetOverlappedResult(True)
    if error == 995:
        return False
    if error:
        raise OSError(None, 'runtime pipe accept failed', None, error)
    return True


def _read_line(handle, deadline, maximum):
    win = _native()
    chunks = bytearray()
    while b'\n' not in chunks:
        ov, _ = win.ReadFile(handle, min(4096, maximum + 1 - len(chunks)), overlapped=True)
        count = _complete(ov, deadline)
        if not count:
            raise ConnectionError('runtime pipe closed before response')
        chunks.extend(ov.getbuffer()[:count])
        if len(chunks) > maximum:
            raise PermissionError('runtime control message too large')
    return bytes(chunks).split(b'\n', 1)[0]


def _write(handle, data, deadline):
    win = _native()
    while data:
        ov, _ = win.WriteFile(handle, data, overlapped=True)
        sent = _complete(ov, deadline)
        if not sent:
            raise ConnectionError('runtime pipe closed during write')
        data = data[sent:]


_PIPE_CLIENT_SQOS = 0x00100000 | 0x00010000


def query_runtime_control(home: Path, request: bytes, timeout: float) -> bytes:
    from gateway.control_socket import windows_pipe_name
    win = _native()
    deadline = time.monotonic() + timeout
    name = windows_pipe_name(home)
    _remaining_ms(deadline)
    while True:
        try:
            # SECURITY_SQOS_PRESENT | SECURITY_IDENTIFICATION: the server may read (never act as)
            # this client's token, captured now, so it can identify us even after we exit.
            handle = win.CreateFile(name, 0xC0000000, 0, win.NULL, win.OPEN_EXISTING,
                                    win.FILE_FLAG_OVERLAPPED | _PIPE_CLIENT_SQOS, win.NULL)
            break
        except OSError as exc:
            if exc.winerror == 2:
                raise FileNotFoundError('runtime control pipe absent') from None
            if exc.winerror == 5:
                raise PermissionError('runtime control pipe inaccessible') from None
            if exc.winerror != 231:
                raise
            try:
                win.WaitNamedPipe(name, _remaining_ms(deadline))
            except OSError as wait_error:
                if wait_error.winerror == 121:
                    raise TimeoutError('runtime control deadline exceeded') from None
                raise
    try:
        _peer_subject(handle, server=True)
        _write(handle, request.rstrip(b'\n') + b'\n', deadline)
        return _read_line(handle, deadline, 512 * 1024) + b'\n'
    finally:
        win.CloseHandle(handle)


# Listening pipe instances, one worker each. Requests are served concurrently, as the POSIX
# socket does through its executor: a handler may dial this same pipe (a hosted room whose
# source and member share one multiplexer re-enters it via hosted-producer -> hosted-attest).
_PIPE_INSTANCES = 8


class NativeControlServer:
    """Bounded native workers, independent of asyncio event-loop policy."""
    def __init__(self, home, handler):
        self.home, self.handler = home, handler
        self._stop = threading.Event()
        self._ready = threading.Event()
        self._error = None
        self._threads = [threading.Thread(target=self._run, args=(index == 0,), daemon=True,
                                          name=f'runtime-control-pipe-{index}')
                         for index in range(_PIPE_INSTANCES)]
        self._thread = self._threads[0]

    def start(self):
        # The first instance claims the name (FILE_FLAG_FIRST_PIPE_INSTANCE) before any other.
        self._thread.start()
        if not self._ready.wait(5):
            self.close()
            raise TimeoutError('runtime control pipe startup deadline exceeded')
        if self._error:
            raise self._error
        for thread in self._threads[1:]:
            thread.start()

    def close(self):
        self._stop.set()
        for thread in self._threads:
            if thread.ident is not None:
                thread.join(timeout=5)
        if any(thread.is_alive() for thread in self._threads):
            raise TimeoutError('runtime control pipe did not stop')

    def _create_pipe(self, first=True):
        from gateway.control_socket import windows_pipe_name
        win = _native()
        class SecurityAttributes(ctypes.Structure):
            _fields_ = [('length', wintypes.DWORD), ('descriptor', ctypes.c_void_p), ('inherit', wintypes.BOOL)]
        a = ctypes.WinDLL('advapi32', use_last_error=True)
        descriptor = ctypes.c_void_p()
        convert = _api(a, 'ConvertStringSecurityDescriptorToSecurityDescriptorW', wintypes.BOOL,
                       [wintypes.LPCWSTR, wintypes.DWORD, ctypes.POINTER(ctypes.c_void_p), ctypes.c_void_p])
        # Protected DACL: exactly the current user, no inherited Everyone/Users ACEs.
        if not convert('D:P(A;;GA;;;' + _process_sid(os.getpid()) + ')', 1, ctypes.byref(descriptor), None):
            raise PermissionError('cannot create private pipe DACL')
        attrs = SecurityAttributes(ctypes.sizeof(SecurityAttributes), descriptor, False)
        try:
            return win.CreateNamedPipe(windows_pipe_name(self.home),
                                       3 | win.FILE_FLAG_OVERLAPPED | (0x00080000 if first else 0),
                                       8, _PIPE_INSTANCES, 65536, 65536, 2000, ctypes.addressof(attrs))
        finally:
            k = ctypes.WinDLL('kernel32', use_last_error=True)
            _api(k, 'LocalFree', ctypes.c_void_p, [ctypes.c_void_p])(descriptor)

    def _run(self, first=True):
        win = _native()
        handle = None
        try:
            handle = self._create_pipe(first)
            self._ready.set()
            while not self._stop.is_set():
                try:
                    connected = _accept(handle, 0.5)
                except OSError:
                    # e.g. ERROR_NO_DATA: a client came and went. Fall through so the serve path
                    # fails fast and the instance is recycled by DisconnectNamedPipe below.
                    connected = True
                if not connected:
                    # Idle: the instance is still listening and a client may have opened it after
                    # the cancel. DisconnectNamedPipe here dropped that client mid-request (233).
                    continue
                try:
                    # Read first: ImpersonateNamedPipeClient needs data read from the pipe, and the
                    # handler runs only once the peer is identified (bounded: 64 KiB, 2 s).
                    raw = _read_line(handle, time.monotonic() + 2, 64 * 1024)
                    subject = _peer_subject(handle, server=False)
                    response = self.handler(raw, subject)
                    # The I/O budget bounds the peer, not the handler's own (client-bounded) work.
                    deadline = time.monotonic() + 2
                    _write(handle, response, deadline)
                    # DisconnectNamedPipe discards unread output. Let the client
                    # consume and close, bounded by the same I/O deadline.
                    _read_line(handle, deadline, 64 * 1024)
                except (OSError, TimeoutError, ConnectionError):
                    pass
                finally:
                    try:
                        _disconnect_pipe(handle)
                    except OSError as exc:
                        if exc.winerror != 233:  # ERROR_PIPE_NOT_CONNECTED: the client already left
                            raise
        except BaseException as exc:
            # Hand-off to the owning thread: start() re-raises the stored error.
            logger.debug('runtime control pipe stopped', exc_info=True)
            self._error = exc
            self._ready.set()
        finally:
            if handle is not None:
                win.CloseHandle(handle)


# Why a same-user gateway's control pipe or process refused this client (#139845, #140252). The
# pipe DACL names the user's SID, so a refusal between two processes of one user means a boundary
# the SID does not cross: an elevated owner (High integrity) seen from a filtered token, or an
# owner in another logon session (a session-0 S4U boot task). ``authorization`` reads as a
# credential problem and names neither.
GATEWAY_ELEVATED = "gateway_elevated"
GATEWAY_OTHER_SESSION = "gateway_other_session"


def classify_refused_owner(*, own_session: int | None, owner_session: int | None,
                           own_elevated: bool, owner_elevated: bool | None) -> str | None:
    """Pure decision over probed facts; ``None`` keeps the generic ``authorization`` code.

    ``owner_elevated`` is None when the owner's token could not be opened; from a non-elevated
    client in the same session that refusal is itself the elevation boundary."""
    if own_session is not None and owner_session is not None and own_session != owner_session:
        return GATEWAY_OTHER_SESSION
    if own_elevated:
        return None
    if owner_elevated or (owner_elevated is None and owner_session is not None):
        return GATEWAY_ELEVATED
    return None


def refused_owner_remedy(reason: str, pid: int, owner_session: int | None) -> str:
    if reason == GATEWAY_ELEVATED:
        return (f"The running gateway (PID {pid}) was started as administrator, so this non-elevated "
                "app cannot reach it. Stop it from an administrator prompt (hermes gateway stop), then "
                "start Hermes without elevation")
    where = f"session {owner_session}" if owner_session is not None else "another session"
    return (f"The running gateway (PID {pid}) runs in another Windows logon session ({where}, e.g. a "
            "boot-time scheduled task), so this app cannot reach it. Stop it (hermes gateway stop from an "
            "administrator prompt) and let only one auto-start task launch the gateway")


_ADMINISTRATORS_SID = 'S-1-5-32-544'  # BUILTIN\Administrators: default owner of what an elevated admin creates
_ACCESS_ALLOWED_ACE = 0


class RefusedOwner(PermissionError):
    """This user's gateway, behind a boundary this client must not cross. ``str()`` is the remedy
    (the refused-owner wording above); ``reason_code`` is the bounded code discovery reports."""

    def __init__(self, reason_code: str, pid: int, remedy: str):
        super().__init__(remedy)
        self.reason_code, self.pid = reason_code, pid


def judge_pipe_server(*, own_sid: str, process_sid: str | None, pipe: tuple[str, tuple] | None, pid: int,
                      own_elevated: bool, own_session: int | None, server_session: int | None,
                      server_elevated: bool | None = None) -> str:
    """``sid:<own>`` when this client may talk to the pipe's server; raises otherwise. Pure: the
    Win32 facts are passed in (#140706).

    *process_sid* is the server process's token user, None when this client cannot open the
    process; *pipe* is the pipe object's ``(owner, dacl)`` (:func:`_pipe_security`), None when
    unreadable; *server_elevated* is the readable server token's ``TokenElevation``, None when
    that probe failed.

    * Server process readable: its token user must be this user, and a non-elevated client never
      hands its requests to an elevated server (its tools would run with the full administrator
      token). The server token's elevation says so whatever owns the pipe (the 'Object creator'
      owner policy gives an elevated process's pipe to the user); only when that probe fails does
      the owner decide: BUILTIN\\Administrators means an elevated process made the pipe.
    * Server process unreadable (an elevated server seen from a filtered token): never attached.
      The pipe object says who made it: owner this user or BUILTIN\\Administrators, and a DACL
      granting exactly this user, the gateway's own private shape (``_create_pipe``). Another
      non-administrator account can set neither owner, so that is this user's elevated gateway
      (or an administrator, who already controls this account) and the refusal names it. Any
      other pipe stays the generic refusal."""
    owner, aces = pipe if pipe is not None else (None, ())
    if process_sid is not None:
        if process_sid != own_sid:
            raise PermissionError('named-pipe peer belongs to another user')
        if server_elevated is None:
            server_elevated = owner == _ADMINISTRATORS_SID
        if own_elevated or not server_elevated:
            return 'sid:' + own_sid
        owner_elevated = True
    else:
        if pipe is None:
            raise PermissionError('cannot verify bootstrap peer process')
        # The owner check carries the trust (no other non-administrator account can set either
        # owner). The ACE shape ignores AceFlags and the access mask: it only filters out pipes
        # that are not the gateway's private shape, it proves nothing on its own.
        if owner not in (own_sid, _ADMINISTRATORS_SID) or tuple(aces) != ((_ACCESS_ALLOWED_ACE, own_sid),):
            raise PermissionError('named-pipe peer belongs to another user')
        owner_elevated = True if owner == _ADMINISTRATORS_SID else None
    reason = classify_refused_owner(own_session=own_session, owner_session=server_session,
                                    own_elevated=own_elevated, owner_elevated=owner_elevated)
    if reason is None:
        raise PermissionError('cannot verify bootstrap peer process')
    logger.info('control pipe server PID %s refused as %s: pipe owner %s, server process %s, '
                'server elevated %s, server session %s, own session %s', pid, reason, owner,
                'readable' if process_sid else 'unreadable', server_elevated, server_session, own_session)
    raise RefusedOwner(reason, pid, refused_owner_remedy(reason, pid, server_session))


def _server_subject(handle, pid: int) -> str:
    """The client half of the pipe identity check: the Win32 facts for :func:`judge_pipe_server`."""
    from hermes_dir_acl import process_elevation, process_is_elevated
    try:
        process_sid = _process_sid(pid)
    except PermissionError:
        process_sid = None
    # A readable process's token answers TokenElevation too (the same OpenProcess + TOKEN_QUERY);
    # None, the probe failing, leaves the pipe owner to decide.
    server_elevated = process_elevation(pid) if process_sid is not None else None
    try:
        pipe = _pipe_security(handle)
    except PermissionError:
        pipe = None
    return judge_pipe_server(own_sid=_process_sid(os.getpid()), process_sid=process_sid, pipe=pipe, pid=pid,
                             own_elevated=process_is_elevated(), own_session=_session_of(os.getpid()),
                             server_session=_pipe_server_session(handle), server_elevated=server_elevated)


def _session_of(pid: int) -> int | None:
    k = ctypes.WinDLL('kernel32', use_last_error=True)
    session = wintypes.DWORD()
    to_session = _api(k, 'ProcessIdToSessionId', wintypes.BOOL, [wintypes.DWORD, ctypes.POINTER(wintypes.DWORD)])
    return session.value if to_session(pid, ctypes.byref(session)) else None


def probe_refused_owner(pid: int) -> tuple[str, str] | None:
    """``(reason_code, remedy)`` naming why *pid*'s gateway refused this client, or None."""
    from hermes_dir_acl import process_elevation, process_is_elevated
    own_session, owner_session = _session_of(os.getpid()), _session_of(pid)
    reason = classify_refused_owner(own_session=own_session, owner_session=owner_session,
                                    own_elevated=process_is_elevated(), owner_elevated=process_elevation(pid))
    return (reason, refused_owner_remedy(reason, pid, owner_session)) if reason else None