diff --git a/emrg/sandbox/fence.py b/emrg/sandbox/fence.py index 4e8a4822..2894fb7d 100644 --- a/emrg/sandbox/fence.py +++ b/emrg/sandbox/fence.py @@ -52,7 +52,7 @@ from emrg.sandbox.grants import GrantKindNotEnforced from emrg.sandbox.policy import DANGER_FULL_ACCESS, SandboxPolicy -from emrg.sandbox.roots import canonical_path, writable_roots +from emrg.sandbox.roots import canonical_path, named_grants, writable_roots from emrg.tools.file_policy import protected_paths @@ -91,7 +91,13 @@ def file_refusal(file_path: str, policy: SandboxPolicy) -> str | None: # ``emrg/tools/file_policy.py`` because the legacy scanner reads it too. return f"{policy.mode} sandbox: blocked write to protected daemon file {file_path!r}" try: - roots = writable_roots(policy) + # The enforcement question is asked here rather than left to + # ``writable_roots``, which projects paths and so has no answer for a + # kind that contributes none. Asking it through the projection made it + # refuse an *enforceability* problem, and the two became inseparable the + # moment Windows began enforcing a key (PR #2498): the raise stopped, and + # with it both the refusal and the projection's own guarantee. + named_grants(policy) except GrantKindNotEnforced as refused: # A grant this platform cannot enforce is refused rather than dropped # (issue #2492, requirement 5). The fence answers in text because its @@ -99,6 +105,7 @@ def file_refusal(file_path: str, policy: SandboxPolicy) -> str | None: # same exception propagate instead, so a confined command fails closed # rather than running with a boundary nobody applied. return f"{policy.mode} sandbox: {refused.reason}" + roots = writable_roots(policy) if any(_is_within_root(target, root) for root in roots): return None granted = ( diff --git a/emrg/sandbox/grants.py b/emrg/sandbox/grants.py index 1944b62c..86831a6d 100644 --- a/emrg/sandbox/grants.py +++ b/emrg/sandbox/grants.py @@ -69,15 +69,18 @@ #: The kinds **this platform's dialects implement**, per platform family. #: -#: ``registry`` is deliberately in none of them: the kind is modelled and -#: :func:`enforcement_refusal` names the reason, but the code that applies a DACL -#: to a key does not exist yet (issue #2492 Stage 2). A set rather than a comment -#: because two callers read it — the refusal sentence, and the guard that keeps -#: :data:`DEFAULT_MANIFEST` from claiming a kind its platform cannot honour. +#: ``registry`` is enforced on ``win32`` alone: a key's DACL is read and rewritten +#: through ``RegGetKeySecurity`` / ``RegSetKeySecurity`` +#: (:mod:`emrg.sandbox.win32.registry`). The other three families name no +#: registry API at all, so for them the kind stays modelled and unenforced, and +#: :func:`enforcement_refusal` says so by name (issue #2492 Stage 2). A set +#: rather than a comment because two callers read it — the refusal sentence, and +#: the guard that keeps :data:`DEFAULT_MANIFEST` from claiming a kind its +#: platform cannot honour. ENFORCED_KINDS: dict[str, tuple[str, ...]] = { "darwin": (GRANT_DIRECTORY, GRANT_FILE), "linux": (GRANT_DIRECTORY, GRANT_FILE), - "win32": (GRANT_DIRECTORY, GRANT_FILE), + "win32": (GRANT_DIRECTORY, GRANT_FILE, GRANT_REGISTRY), "other": (GRANT_DIRECTORY, GRANT_FILE), } @@ -90,13 +93,12 @@ #: Two properties this table owes, both asserted by tests rather than trusted: #: #: * **a row's kind is one its platform can enforce** — :data:`ENFORCED_KINDS` is -#: the reader, so the Windows registry keys the rant also names are absent -#: here until Stage 2 can enforce them. A manifest promising a grant the run -#: does not have would be the lying boundary this module refuses elsewhere; -#: * **the table is data, not a code path** — an entry whose directory does not -#: exist on this host is simply absent from the derivation -#: (:func:`default_grants`), so a deployment that has never run ``lark-cli`` -#: is not an error and not a grant. +#: the reader, so a row is only written for a kind that platform's dialect +#: implements. That is what the Windows registry rows below are held to, and +#: why no ``darwin``/``linux`` row names one; +#: * **the table is data, not a code path** — an entry whose object does not exist +#: on this host is simply absent from the derivation (:func:`default_grants`), so +#: a deployment that has never run ``lark-cli`` is not an error and not a grant. #: #: ``%APPDATA%`` is expanded with :func:`os.path.expandvars`, which understands #: the ``%NAME%`` form on Windows only; on POSIX the two darwin/linux rows carry @@ -113,6 +115,16 @@ "win32": ( (GRANT_DIRECTORY, "~/.lark-cli"), (GRANT_DIRECTORY, "%APPDATA%/Microsoft/Protect"), + # Two rows for one tool, and both are load-bearing: the **parent** is what + # lets the confined child create ``keychain`` in the first place (the + # capability ACE is inheritable), while the **key** itself is what reaches + # a ``keychain`` that already exists — inheritance applies when a key is + # created, and ``RegSetKeySecurity`` on the parent does not walk the keys + # already beneath it. A deployment where ``keychain`` is absent loses only + # the second row (:func:`default_grants` is total over a key this host does + # not have) and still gets the first. + (GRANT_REGISTRY, "HKCU\\Software\\LarkCli"), + (GRANT_REGISTRY, "HKCU\\Software\\LarkCli\\keychain"), ), } @@ -292,10 +304,18 @@ def default_grants(platform: str | None = None) -> list[RootGrant]: out: list[RootGrant] = [] for kind, spelling in DEFAULT_MANIFEST.get(family, ()): target = os.path.expandvars(os.path.expanduser(spelling)) - try: - exists = os.path.isdir(target) if kind == GRANT_DIRECTORY else os.path.exists(target) - except OSError: # pragma: no cover - both calls swallow their own errors - exists = False + if kind == GRANT_REGISTRY: + # Imported here rather than at module scope: the registry predicate is + # Windows-only machinery, and this derivation runs on every platform + # for every family. Off Windows it answers False without touching it. + from emrg.sandbox.win32.registry import key_exists # noqa: PLC0415 + + exists = key_exists(target) + else: + try: + exists = os.path.isdir(target) if kind == GRANT_DIRECTORY else os.path.exists(target) + except OSError: # pragma: no cover - both calls swallow their own errors + exists = False if not exists: continue out.append(RootGrant(kind=kind, target=target, source=SOURCE_DEFAULT)) diff --git a/emrg/sandbox/providers/win32.py b/emrg/sandbox/providers/win32.py index 99471558..27adbf2a 100644 --- a/emrg/sandbox/providers/win32.py +++ b/emrg/sandbox/providers/win32.py @@ -44,6 +44,7 @@ from dataclasses import dataclass from emrg.sandbox.contract import Runner, RunnerFailureRule +from emrg.sandbox.grants import GRANT_REGISTRY from emrg.sandbox.policy import SandboxPolicy from emrg.sandbox.roots import canonical_path, named_grants from emrg.sandbox.win32.ffi import Win32Bindings, win32 @@ -387,10 +388,12 @@ def runner_argv(policy: SandboxPolicy) -> list[str]: "--temp", tempfile.gettempdir(), "--mode", policy.mode, ] - extra_roots = _extra_root_spellings(policy) + path_spellings, key_spellings = _extra_grant_spellings(policy) extra_flags: list[str] = [] - for root in extra_roots: + for root in path_spellings: extra_flags += ["--extra-root", root] + for key in key_spellings: + extra_flags += ["--extra-key", key] if policy.session_id is None: return [ *runner_invocation(), @@ -399,7 +402,7 @@ def runner_argv(policy: SandboxPolicy) -> list[str]: "--mode", policy.mode, *extra_flags, ] - capability = STORE.materialize(policy.session_id, workspace, extra_roots=extra_roots) + capability = STORE.materialize(policy.session_id, workspace, extra_roots=path_spellings) return [ *runner_invocation(), "--workspace", workspace, @@ -411,8 +414,8 @@ def runner_argv(policy: SandboxPolicy) -> list[str]: ] -def _extra_root_spellings(policy: SandboxPolicy) -> tuple[str, ...]: - """The enforceable extra grants' paths — the manifest's and the session's. +def _extra_grant_spellings(policy: SandboxPolicy) -> tuple[tuple[str, ...], tuple[str, ...]]: + """The enforceable extra grants' spellings, split by object kind. Read from the shared derivation (:func:`~emrg.sandbox.roots.named_grants`) rather than off ``policy.extra_grants``, because a grant the host never typed @@ -420,16 +423,42 @@ def _extra_root_spellings(policy: SandboxPolicy) -> tuple[str, ...]: manifest is for (issue #2492). Canonicalization and de-duplication happen in that derivation, once, for every dialect. - A **registry** grant never reaches this function: no platform enforces that - kind yet, and the derivation raises - :class:`~emrg.sandbox.grants.GrantKindNotEnforced` rather than handing a key - spelling to a path consumer. On Windows that is the loop Stage 2 closes — the - key needs a DACL of its own, not an ``--extra-root`` flag. + The split is the whole point of the pair: a path grant becomes an + ``--extra-root`` and a key grant becomes an ``--extra-key``, because this + backend applies a DACL to a *path* with ``SetNamedSecurityInfoW`` and to a + *key* with ``RegSetKeySecurity``. Handing a key spelling to the path half + would grant a filesystem object nobody named. + + :param policy: the resolved per-call policy. + :returns: the path spellings and the key spellings. + """ + named = named_grants(policy) + return ( + tuple(grant.target for grant in named if grant.kind != GRANT_REGISTRY), + tuple(grant.target for grant in named if grant.kind == GRANT_REGISTRY), + ) + + +def _extra_root_spellings(policy: SandboxPolicy) -> tuple[str, ...]: + """The enforceable extra grants' paths — the manifest's and the session's. + + :param policy: the resolved per-call policy. + :returns: the canonical path spellings. + """ + return _extra_grant_spellings(policy)[0] + + +def _extra_key_spellings(policy: SandboxPolicy) -> tuple[str, ...]: + """The enforceable extra grants' registry keys — the manifest's and the session's. + + One producer for the key half too, for the reason the path half has one: the + ACE the seam applies and the SID the runner puts in the token are derived from + this same list, so a key cannot end up granted on one side only. :param policy: the resolved per-call policy. - :returns: the canonical spellings. + :returns: the canonical key spellings. """ - return tuple(grant.target for grant in named_grants(policy)) + return _extra_grant_spellings(policy)[1] def release_session(session_id: str) -> None: diff --git a/emrg/sandbox/roots.py b/emrg/sandbox/roots.py index 233e1ade..0105a4bd 100644 --- a/emrg/sandbox/roots.py +++ b/emrg/sandbox/roots.py @@ -14,7 +14,7 @@ import os import tempfile -from typing import TYPE_CHECKING, NamedTuple +from typing import TYPE_CHECKING, Callable, NamedTuple from emrg.sandbox.grants import ( GRANT_DIRECTORY, @@ -192,19 +192,58 @@ def granted_grants(policy: SandboxPolicy) -> list[RootGrant]: """ if policy.mode != "workspace-write": return [] - return _dedupe([*_mode_grants(policy), *named_grants(policy)]) + return _composed(policy, named_grants) + + +def _composed( + policy: SandboxPolicy, named: Callable[[SandboxPolicy], list[RootGrant]] +) -> list[RootGrant]: + """One composition of a policy's grants: the mode's sources, then the named ones. + + The two named-list policies differ in exactly one respect — whether the + enforcement table is consulted (:func:`named_grants` asks it, + :func:`stored_grants` does not) — and both reach this same composition, so + the readers of either cannot disagree about which grants exist: "the shell + can write where ``write`` cannot" needs one derivation, not one call site. + + ``read-only`` (and ``danger-full-access``, which never asks) grant nothing: + the early return is what makes requirement 4 of issue #2492 true by + construction rather than by a check — a default manifest cannot widen a tier + that grants none, because this function never consults one for that tier. + + :param policy: the file-effect policy to derive the grant list from. + :param named: which named-list policy to append — the enforcement-checked + one, or the display one the path projection reads. + :returns: the canonical, deduplicated grants; empty for every mode but + ``workspace-write``. + """ + if policy.mode != "workspace-write": + return [] + return _dedupe([*_mode_grants(policy), *named(policy)]) def writable_roots(policy: SandboxPolicy) -> list[str]: - """The **paths** among :func:`granted_grants` — the projection path-readers use. + """The **paths** a policy grants — the projection path-readers use. - A derived view, not a second allow-list: it is built from - :func:`granted_grants` and can therefore never disagree with it. It exists + A derived view, not a second allow-list: it is built by the same + :func:`_composed` composition :func:`granted_grants` reads, and applies one + filter to it, so it can never disagree with that derivation. It exists because several readers need nothing but paths — the fence's containment test, the confined cache relocation, and the providers' own spellings — and because a registry grant has no path to contribute, so a path-only reader must not be handed one. + **It does not ask the enforcement table**, and that is the point of the + split. Whether this platform can enforce a registry key is a question about + enforcement; a key contributes no path either way. Asking it here is what + once made this projection's own guarantee unmeasurable: while no platform + enforced a key the projection refused before it could be asked, and the day + one did — Windows, PR #2498 — the refusal vanished and so did the "no key + among the paths" claim, leaving a green reading about nothing. The loud + answer to a grant this platform cannot enforce is :func:`named_grants`'s, + and every reader that *acts* on grants reads that one: the profile builders + directly, and the fence by asking for the refusal itself. + ``read-only`` still answers ``[]``, which is the property the host's rule rests on: at a tier that grants nothing, nothing is writable however the session or the manifest is configured. @@ -212,7 +251,11 @@ def writable_roots(policy: SandboxPolicy) -> list[str]: :param policy: the file-effect policy to derive the allow-list from. :returns: the canonical writable paths. """ - return [grant.target for grant in granted_grants(policy) if grant.kind != GRANT_REGISTRY] + return [ + grant.target + for grant in _composed(policy, stored_grants) + if grant.kind != GRANT_REGISTRY + ] #: The shared POSIX temp root — one spelling, named once, read through diff --git a/emrg/sandbox/win32/abi.py b/emrg/sandbox/win32/abi.py index be25b2cb..c25b3d11 100644 --- a/emrg/sandbox/win32/abi.py +++ b/emrg/sandbox/win32/abi.py @@ -96,6 +96,51 @@ #: Maximum SID allocation size in bytes. SECURITY_MAX_SID_SIZE = 68 +# --- registry keys (issue #2492 Stage 2) ----------------------------------- + +#: Predefined ``HKEY_CURRENT_USER`` handle. +HKEY_CURRENT_USER = 0x80000001 +#: Predefined ``HKEY_LOCAL_MACHINE`` handle. +HKEY_LOCAL_MACHINE = 0x80000002 +#: Standard right: read an object's DACL. +READ_CONTROL = 0x00020000 +#: Standard right: replace an object's DACL. +WRITE_DAC = 0x00040000 +#: Key access needed to read a key's DACL and rewrite it — and nothing else. +#: +#: ``RegOpenKeyExW`` wants the rights to be asked for up front, so the applier +#: asks for exactly the two the edit needs: with these, a key whose DACL the +#: caller may not rewrite fails at the open, before any merge is attempted. +KEY_SECURITY_ACCESS = READ_CONTROL | WRITE_DAC +#: Key right: read a key's values. +KEY_QUERY_VALUE = 0x0001 +#: Key right: write a key's values. +KEY_SET_VALUE = 0x0002 +#: Key right: create a subkey. +KEY_CREATE_SUB_KEY = 0x0004 +#: Key right: enumerate a key's subkeys. +KEY_ENUMERATE_SUB_KEYS = 0x0008 +#: Key right: receive change notifications for a key. +KEY_NOTIFY = 0x0010 +#: Capability-SID access mask granting read, value writes and subkey creation. +#: +#: The registry's counterpart of :data:`GRANT_MASK`, and it excludes the standard +#: rights for the same reason: ``WRITE_DAC`` and ``WRITE_OWNER`` would let a +#: confined child rewrite the key's DACL or take its ownership and walk out of the +#: allowlist it was handed. ``READ_CONTROL`` is excluded with them — the mask +#: grants what a tool does to a key's *contents*, never what it does to the key's +#: own security descriptor. +REGISTRY_GRANT_MASK = ( + KEY_QUERY_VALUE | KEY_SET_VALUE | KEY_CREATE_SUB_KEY | KEY_ENUMERATE_SUB_KEYS | KEY_NOTIFY +) +#: ``SECURITY_DESCRIPTOR_REVISION``: the only revision ``InitializeSecurityDescriptor`` accepts. +SECURITY_DESCRIPTOR_REVISION = 1 +#: The bytes an absolute ``SECURITY_DESCRIPTOR`` occupies. +#: +#: Named by the SDK for exactly this use — the buffer an absolute descriptor is +#: built in before ``MakeSelfRelativeSD`` measures the form the API wants. +SECURITY_DESCRIPTOR_MIN_LENGTH = 20 + # --- process creation (Win32 process owner, win32-process/src/abi.ts) ------- #: STARTUPINFO.dwFlags: the std handles in the struct are meaningful. diff --git a/emrg/sandbox/win32/acl.py b/emrg/sandbox/win32/acl.py index 89319be9..d56401ff 100644 --- a/emrg/sandbox/win32/acl.py +++ b/emrg/sandbox/win32/acl.py @@ -278,12 +278,14 @@ def merge_and_apply( throw_last_error(api, "LocalFree", f"{label}({path}) new ACL") -def has_exact_grant(api: Win32Bindings, old_acl: int, sid_ptr: int) -> bool: +def has_exact_grant( + api: Win32Bindings, old_acl: int, sid_ptr: int, mask: int = abi.GRANT_MASK +) -> bool: """Whether the DACL already carries the exact write grant this module adds. Every field is read through bounded offset reads (Allow ACE, ``OI|CI`` - inheritance, :data:`~emrg.sandbox.win32.abi.GRANT_MASK`, the capability - SID) — no pointer arithmetic, no ``memcpy``. The ACE's SID is **inline** + inheritance, the access mask, the capability SID) — no pointer arithmetic, + no ``memcpy``. The ACE's SID is **inline** (embedded after the 4-byte mask), so there is no pointer to follow: it is read byte-wise against the ACE's own size. A malformed header reads as "no exact grant", which sends the caller down the merge path — the path that @@ -292,6 +294,10 @@ def has_exact_grant(api: Win32Bindings, old_acl: int, sid_ptr: int) -> bool: :param api: the binding table. :param old_acl: the current explicit DACL pointer. :param sid_ptr: the capability SID to match. + :param mask: the access mask the grant's ACE carries — the file rights by + default, and :data:`~emrg.sandbox.win32.abi.REGISTRY_GRANT_MASK` for a + registry key. A second checker was the alternative and would have been + this one with one name changed. :returns: whether the exact grant ACE is already present. """ header = ACLStructure.from_address(old_acl) @@ -307,11 +313,11 @@ def has_exact_grant(api: Win32Bindings, old_acl: int, sid_ptr: int) -> bool: ace_size = int(ctypes.c_uint16.from_address(old_acl + offset + 2).value) if ace_size < 8 or offset + ace_size > acl_size: return False # implausible: fall back to the merge path - mask = int(ctypes.c_uint32.from_address(old_acl + offset + 4).value) + ace_mask = int(ctypes.c_uint32.from_address(old_acl + offset + 4).value) exact = ( ace_type == abi.ACCESS_ALLOWED_ACE_TYPE and ace_flags == abi.SUB_CONTAINERS_AND_OBJECTS_INHERIT - and mask == abi.GRANT_MASK + and ace_mask == mask ) if exact: inline = _inline_sid_bytes(old_acl + offset + 8, ace_size - 8) diff --git a/emrg/sandbox/win32/ffi.py b/emrg/sandbox/win32/ffi.py index 6f7d0636..45ab8882 100644 --- a/emrg/sandbox/win32/ffi.py +++ b/emrg/sandbox/win32/ffi.py @@ -353,6 +353,49 @@ def _declare(self) -> None: c.POINTER(c.c_void_p), c.POINTER(c.c_void_p), ] + # -- a registry key's DACL (issue #2492 Stage 2) --------------------- + # + # ``LSTATUS`` is a 32-bit signed ``LONG``, not a ``DWORD``: a negative + # status declared unsigned would read as a huge success code and the + # caller's ``!= ERROR_SUCCESS`` check would still fire, but the code it + # printed would be a number the system's message table has no text for. + + a32.RegOpenKeyExW.restype = wintypes.LONG + a32.RegOpenKeyExW.argtypes = [ + c.c_void_p, wintypes.LPCWSTR, wintypes.DWORD, c.c_int, + c.POINTER(c.c_void_p), + ] + + a32.RegCloseKey.restype = wintypes.LONG + a32.RegCloseKey.argtypes = [c.c_void_p] + + a32.RegGetKeySecurity.restype = wintypes.LONG + a32.RegGetKeySecurity.argtypes = [ + c.c_void_p, wintypes.DWORD, c.c_void_p, c.POINTER(wintypes.DWORD), + ] + + a32.RegSetKeySecurity.restype = wintypes.LONG + a32.RegSetKeySecurity.argtypes = [c.c_void_p, wintypes.DWORD, c.c_void_p] + + a32.GetSecurityDescriptorDacl.restype = wintypes.BOOL + a32.GetSecurityDescriptorDacl.argtypes = [ + c.c_void_p, c.POINTER(wintypes.BOOL), c.POINTER(c.c_void_p), + c.POINTER(wintypes.BOOL), + ] + + a32.InitializeSecurityDescriptor.restype = wintypes.BOOL + a32.InitializeSecurityDescriptor.argtypes = [c.c_void_p, wintypes.DWORD] + + a32.SetSecurityDescriptorDacl.restype = wintypes.BOOL + a32.SetSecurityDescriptorDacl.argtypes = [ + c.c_void_p, wintypes.BOOL, c.c_void_p, wintypes.BOOL, + ] + + a32.MakeSelfRelativeSD.restype = wintypes.BOOL + a32.MakeSelfRelativeSD.argtypes = [ + c.c_void_p, c.c_void_p, c.POINTER(wintypes.DWORD), + ] + # -- the primitive helpers every call site shares ------------------------ def get_last_error(self) -> int: diff --git a/emrg/sandbox/win32/registry.py b/emrg/sandbox/win32/registry.py new file mode 100644 index 00000000..8105df0b --- /dev/null +++ b/emrg/sandbox/win32/registry.py @@ -0,0 +1,471 @@ +"""DACL editing for a registry key: grant and revoke a capability SID on a key. + +The registry counterpart of :mod:`emrg.sandbox.win32.acl`, and deliberately a +second module rather than a second mode of that one: the two edit one kind of +thing — a security descriptor's DACL — through APIs that have almost nothing in +common. A path is addressed by name, and its DACL is read and written by +``GetNamedSecurityInfoW`` / ``SetNamedSecurityInfoW``, which hand back a +self-relative descriptor and take one back. A key is addressed by an opened +handle, and its descriptor is read and written by ``RegGetKeySecurity`` / +``RegSetKeySecurity`` — so the merged ACL has to be assembled into a self-relative +descriptor here, through ``InitializeSecurityDescriptor`` → +``SetSecurityDescriptorDacl`` → ``MakeSelfRelativeSD``. One module with two +bodies would be one function with two of everything; the parts that genuinely +are shared — the ``EXPLICIT_ACCESS_W`` packing, the ACE merge, the exact-ACE +check, the per-object lock — are imported from ``acl.py``, not copied. + +Three contracts, the first two carried over from that module: + +* the ACL pointer returned by ``SetEntriesInAclW`` is its **own** allocation and + must be freed, while a descriptor's DACL points *into* the descriptor and must + not be; +* the exact-ACE check keeps provisioning idempotent: without it every provision + re-merges an identical ACE, which re-propagates it across the key's subkeys; +* **the merge preserves what is already there.** The DACL a key carries is where + the user's own access comes from. The ACE added here is an addition, never a + replacement — a rebuild built from the capability SID alone would lock the host + out of their own key. So the current DACL is read, merged into, and written + back; and a key whose DACL is **absent** (a NULL DACL, which grants everyone + full control) is refused rather than replaced, because narrowing that is not + this module's decision to make. + +The key's spelling is parsed once (:func:`split_key`) and its capability SID is +derived from that same spelling (:func:`~emrg.sandbox.win32.sid.registry_write_sid`) +by both halves of the grant — the ACE applied here and the SID the runner puts in +the restricted token — so the two cannot disagree about which key was granted. +""" + +from __future__ import annotations + +import ctypes +from ctypes import wintypes + +from emrg.sandbox.win32 import abi +from emrg.sandbox.win32.acl import ( + build_explicit_access, + has_exact_grant, + with_path_lock, +) +from emrg.sandbox.win32.ffi import ( + Win32Bindings, + alloc_ptr_slot, + decode_ptr, + is_null_ptr, + local_free, + throw_last_error, + throw_win32, +) + +#: The predefined roots a grant may name, by their accepted spellings. +#: +#: Only the two hives a per-session grant can mean anything under: a grant names +#: what one confined process may write, which is the current user's own state +#: (``HKCU``) or a machine-wide key that is already writable (``HKLM``). ``HKU``, +#: ``HKCR`` and ``HKCC`` are views onto the same two, and accepting a third +#: spelling of one place would let two grants of one key derive two SIDs. +KEY_ROOTS: dict[str, int] = { + "HKCU": abi.HKEY_CURRENT_USER, + "HKEY_CURRENT_USER": abi.HKEY_CURRENT_USER, + "HKLM": abi.HKEY_LOCAL_MACHINE, + "HKEY_LOCAL_MACHINE": abi.HKEY_LOCAL_MACHINE, +} + + +class RegistryKeyError(ValueError): + """A grant's target is not a registry key this module can address.""" + + +class KeyDaclRefused(RuntimeError): + """A key's DACL is not one this module will edit.""" + + +class KeyDacl: + """A key's current DACL pointer and the buffer that owns it. + + ``RegGetKeySecurity`` writes the descriptor into a buffer the caller supplies, + and the DACL is a pointer *into* that buffer — so the two travel together or + the pointer is meaningless. It is meaningless in a way that does not raise: + the pointer's next consumers are :func:`~emrg.sandbox.win32.acl.has_exact_grant`, + which walks it as an ACL header, and ``SetEntriesInAclW``, which copies every + ACE it is told is there. On a collected buffer those read an allocator's + bookkeeping as ``AclSize``/``AceCount``, so the walk runs off the end of a + freed block and the copy can overrun the new ACL: the heap is corrupted and + some later call is the one that dies. + + This is :class:`~emrg.sandbox.win32.acl.CurrentDacl` in the registry's idiom, + and the two files carry one each because nothing else keeps a descriptor + alive: the answer holds the buffer, so a caller that holds the answer holds + the memory its pointer addresses. + """ + + __slots__ = ("_buffer", "pointer") + + def __init__(self, pointer: int, buffer: ctypes.Array) -> None: + self._buffer = buffer + self.pointer = pointer + + @property + def buffer(self) -> ctypes.Array: + """The descriptor buffer :attr:`pointer` addresses. + + Exposed so a test can assert the pointer lies inside it, and for nothing + else: the buffer is owned here and is not the caller's to free or resize. + + :returns: the buffer this answer keeps alive. + """ + return self._buffer + + +def split_key(key: str) -> tuple[int, str]: + """Split a key's full spelling into its predefined root and its subkey path. + + Pure, and measured on every platform: the parsing is the half of this module + that says which key a grant named, so it is the half a POSIX leg can hold to + its rule. + + A **bare hive** is refused. ``HKCU`` alone names the whole user hive, and a + grant that reaches every key the user owns is a widening nobody asked for — + the same reason a path grant names the directory rather than the volume. + + :param key: the full spelling, e.g. ``HKCU\\Software\\LarkCli\\keychain``. + :returns: the predefined ``HKEY`` value and the subkey path. + :raises RegistryKeyError: when the spelling names no key this module can + address, with the reason in the message. + """ + if not isinstance(key, str): + raise RegistryKeyError(f"a registry grant needs a key spelling, not {type(key).__name__}") + spelling = key.strip() + if not spelling: + raise RegistryKeyError("a registry grant needs a key spelling, not an empty string") + head, separator, tail = spelling.partition("\\") + if not separator: + raise RegistryKeyError( + f"registry key {spelling!r} names a whole hive; a grant names the key it needs, " + "e.g. HKCU\\Software\\Vendor\\Component" + ) + root = KEY_ROOTS.get(head.upper()) + if root is None: + raise RegistryKeyError( + f"registry key {spelling!r} starts with {head!r}, which is not one of " + f"{', '.join(sorted(KEY_ROOTS))}" + ) + subkey = tail.strip("\\") + if not subkey: + raise RegistryKeyError( + f"registry key {spelling!r} names a whole hive; a grant names the key it needs, " + "e.g. HKCU\\Software\\Vendor\\Component" + ) + return root, subkey + + +def key_exists(key: str) -> bool: + """Whether this host has the key. + + The registry counterpart of ``os.path.isdir`` in the manifest's totality rule: + a default grant naming a key nobody created is **absent**, never an error, so + a host that has never run the tool that owns the key is not a broken boundary. + + Answers ``False`` off Windows, where ``winreg`` does not exist — which is what + lets the one derivation read every platform's manifest row on every platform. + + :param key: the key's full spelling. + :returns: whether the key is present and readable. + """ + try: + root, subkey = split_key(key) + except RegistryKeyError: + return False + try: + import winreg # noqa: PLC0415 - Windows-only stdlib, imported where it exists + except ImportError: # pragma: no cover - the POSIX legs + return False + hive = winreg.HKEY_CURRENT_USER if root == abi.HKEY_CURRENT_USER else winreg.HKEY_LOCAL_MACHINE + try: + with winreg.OpenKey(hive, subkey, 0, winreg.KEY_READ): + return True + except OSError: + return False + + +def open_key(api: Win32Bindings, key: str) -> int: + """Open a key for the two rights a DACL edit needs, and nothing else. + + Asking for exactly ``READ_CONTROL | WRITE_DAC`` means a key the caller may not + re-DACL fails here, at the open, instead of half way through a merge. + + :param api: the binding table. + :param key: the key's full spelling. + :returns: the key handle. + :raises RegistryKeyError: when the spelling is not one this module addresses. + :raises Win32Error: when the open failed. + """ + root, subkey = split_key(key) + handle_slot = alloc_ptr_slot() + status = int(api.advapi32.RegOpenKeyExW( + ctypes.c_void_p(root), + subkey, + 0, + abi.KEY_SECURITY_ACCESS, + ctypes.byref(handle_slot), + )) + if status != abi.ERROR_SUCCESS: + throw_win32(api, "RegOpenKeyExW", status, key) + handle = decode_ptr(handle_slot) + if is_null_ptr(handle): + throw_win32(api, "RegOpenKeyExW", api.get_last_error(), f"{key}: null key handle") + return int(handle) + + +def close_key(api: Win32Bindings, handle: int, key: str) -> None: + """Close one key handle, reporting a failure. + + :param api: the binding table. + :param handle: the handle :func:`open_key` returned. + :param key: the key's full spelling, for the error message. + :raises Win32Error: when ``RegCloseKey`` reported a failure. + """ + status = int(api.advapi32.RegCloseKey(ctypes.c_void_p(handle))) + if status != abi.ERROR_SUCCESS: + throw_win32(api, "RegCloseKey", status, key) + + +def _close_key_best_effort(api: Win32Bindings, handle: int) -> None: + """Close a handle on a path that is already failing. + + A cleanup failure must not mask the action's own error — the rule + ``acl.py``'s lock states — so this call's status is dropped and only the + original exception travels. + + :param api: the binding table. + :param handle: the handle to close. + """ + try: + api.advapi32.RegCloseKey(ctypes.c_void_p(handle)) + except BaseException: # noqa: BLE001 - the caller is already raising + pass + + +def read_current_key_dacl(api: Win32Bindings, key: str, handle: int) -> KeyDacl | None: + """Read the key's current explicit DACL, with the buffer that owns it. + + The descriptor is read into a buffer this module owns, so nothing here is + freed with ``LocalFree``: the DACL pointer refers *into* that buffer, and the + two are therefore handed back together — as a :class:`KeyDacl`, never as a + bare pointer, because a pointer alone outlives nothing and its consumers + would read freed memory instead of raising. + + :param api: the binding table. + :param key: the key's full spelling, for error messages. + :param handle: the open key handle. + :returns: the DACL pointer with the buffer that owns it, or ``None`` when the + key carries no DACL. + :raises Win32Error: when a call failed. + :raises KeyDaclRefused: when the key carries a NULL DACL, which grants + everyone full control; replacing it would narrow access the host has + today, and that is not this module's decision. + """ + size = wintypes.DWORD(0) + probe = int(api.advapi32.RegGetKeySecurity( + ctypes.c_void_p(handle), abi.DACL_SECURITY_INFORMATION, None, ctypes.byref(size), + )) + if probe != abi.ERROR_INSUFFICIENT_BUFFER: + throw_win32(api, "RegGetKeySecurity", probe, f"{key} (size probe)") + if size.value == 0: # pragma: no cover - the API never answers zero here + throw_win32( + api, "RegGetKeySecurity", abi.ERROR_INSUFFICIENT_BUFFER, f"{key}: empty descriptor" + ) + buffer = ctypes.create_string_buffer(size.value) + read = int(api.advapi32.RegGetKeySecurity( + ctypes.c_void_p(handle), + abi.DACL_SECURITY_INFORMATION, + ctypes.cast(buffer, ctypes.c_void_p), + ctypes.byref(size), + )) + if read != abi.ERROR_SUCCESS: + throw_win32(api, "RegGetKeySecurity", read, key) + present = wintypes.BOOL() + dacl_slot = alloc_ptr_slot() + defaulted = wintypes.BOOL() + if int(api.advapi32.GetSecurityDescriptorDacl( + ctypes.cast(buffer, ctypes.c_void_p), + ctypes.byref(present), + ctypes.byref(dacl_slot), + ctypes.byref(defaulted), + )) == 0: + throw_last_error(api, "GetSecurityDescriptorDacl", key) + if not present.value: + raise KeyDaclRefused( + f"registry key {key} carries no DACL, which grants every principal full control; " + "writing one here would narrow access this module did not grant and cannot restore" + ) + dacl = decode_ptr(dacl_slot) + if is_null_ptr(dacl): + return None + return KeyDacl(int(dacl), buffer) + + +def merge_entry(api: Win32Bindings, entry, old_acl: int | None, label: str) -> int: + """Merge one ``EXPLICIT_ACCESS_W`` into a DACL and return the new ACL. + + The registry half of ``acl.py``'s ``merge_and_apply``: ``old_acl`` ``None`` + builds a DACL from scratch, and the result is the caller's to free. + + :param api: the binding table. + :param entry: the packed entry to merge. + :param old_acl: the current DACL pointer, or ``None``. + :param label: the caller's name for error details. + :returns: the merged ACL pointer. + :raises Win32Error: when the merge failed. + """ + new_acl_slot = alloc_ptr_slot() + merged = int(api.advapi32.SetEntriesInAclW( + 1, + ctypes.byref(entry), + ctypes.c_void_p(old_acl or 0), + ctypes.byref(new_acl_slot), + )) + if merged != abi.ERROR_SUCCESS: + throw_win32(api, "SetEntriesInAclW", merged, label) + new_acl = decode_ptr(new_acl_slot) + if is_null_ptr(new_acl): + throw_win32(api, "SetEntriesInAclW", api.get_last_error(), f"{label}: null new ACL") + return int(new_acl) + + +def apply_dacl(api: Win32Bindings, key: str, handle: int, new_acl: int) -> None: + """Install ``new_acl`` as the key's DACL. + + ``RegSetKeySecurity`` takes a **self-relative** descriptor, while an ACL is + installed on an *absolute* one — so the merged ACL is put into a fresh + absolute descriptor, and ``MakeSelfRelativeSD`` produces the form the API + wants. Building a fresh descriptor rather than editing the one read earlier is + what keeps this call honest: the only field carried across is the DACL this + module just merged, and the previous descriptor's own memory is never mutated. + + :param api: the binding table. + :param key: the key's full spelling, for error messages. + :param handle: the open key handle. + :param new_acl: the merged ACL pointer. + :raises Win32Error: when any call failed. + """ + descriptor = ctypes.create_string_buffer(abi.SECURITY_DESCRIPTOR_MIN_LENGTH) + absolute = ctypes.cast(descriptor, ctypes.c_void_p) + if int(api.advapi32.InitializeSecurityDescriptor( + absolute, abi.SECURITY_DESCRIPTOR_REVISION + )) == 0: + throw_last_error(api, "InitializeSecurityDescriptor", key) + if int(api.advapi32.SetSecurityDescriptorDacl( + absolute, True, ctypes.c_void_p(new_acl), False + )) == 0: + throw_last_error(api, "SetSecurityDescriptorDacl", key) + + # The measuring call fails on purpose: a NULL buffer is answered with + # ERROR_INSUFFICIENT_BUFFER and the size the relative form needs. + size = wintypes.DWORD(0) + if int(api.advapi32.MakeSelfRelativeSD(absolute, None, ctypes.byref(size))) == 0: + code = api.get_last_error() + if code != abi.ERROR_INSUFFICIENT_BUFFER: + throw_win32(api, "MakeSelfRelativeSD", code, f"{key} (size probe)") + if size.value == 0: # pragma: no cover - the API never answers zero here + throw_win32( + api, "MakeSelfRelativeSD", abi.ERROR_INSUFFICIENT_BUFFER, f"{key}: empty measurement" + ) + relative = ctypes.create_string_buffer(size.value) + if int(api.advapi32.MakeSelfRelativeSD( + absolute, ctypes.cast(relative, ctypes.c_void_p), ctypes.byref(size) + )) == 0: + throw_last_error(api, "MakeSelfRelativeSD", key) + + status = int(api.advapi32.RegSetKeySecurity( + ctypes.c_void_p(handle), abi.DACL_SECURITY_INFORMATION, ctypes.cast(relative, ctypes.c_void_p) + )) + if status != abi.ERROR_SUCCESS: + throw_win32(api, "RegSetKeySecurity", status, key) + + +def _free_new_acl(api: Win32Bindings, new_acl: int, label: str) -> None: + """Free the ACL ``SetEntriesInAclW`` allocated. + + :param api: the binding table. + :param new_acl: the ACL pointer. + :param label: the caller's name for error details. + :raises Win32Error: when ``LocalFree`` reported a failure. + """ + freed = local_free(api, new_acl) + if not is_null_ptr(freed): + throw_last_error(api, "LocalFree", f"{label} new ACL") + + +def grant_key_write(api: Win32Bindings, key: str, sid_ptr: int) -> None: + """Grant :data:`~emrg.sandbox.win32.abi.REGISTRY_GRANT_MASK` to a capability SID on a key. + + Idempotent: a key whose DACL already carries the exact ACE (one surviving from + an earlier provision) skips the write entirely — otherwise every provision + re-propagates an identical inheritable ACE across the key's subkeys. + Otherwise read-merge-write, so the key's pre-existing ACEs survive. Runs under + the per-object lock, keyed by the key's own spelling. + + :param api: the binding table. + :param key: the key whose DACL gains the grant. + :param sid_ptr: the capability SID the ACE names. + :raises RegistryKeyError: when the spelling is not one this module addresses. + :raises KeyDaclRefused: when the key carries a NULL DACL. + :raises Win32Error: when any call failed. + """ + with with_path_lock(api, key): + handle = open_key(api, key) + try: + # ``dacl`` is bound, not just read: its buffer owns the pointer that + # ``has_exact_grant`` walks and ``merge_entry`` copies ACEs out of, + # so letting it go out of scope here would free the memory under both. + dacl = read_current_key_dacl(api, key, handle) + old_acl = dacl.pointer if dacl is not None else None + if old_acl is not None and has_exact_grant( + api, old_acl, sid_ptr, mask=abi.REGISTRY_GRANT_MASK + ): + close_key(api, handle, key) + return + entry = build_explicit_access(sid_ptr, abi.GRANT_ACCESS, abi.REGISTRY_GRANT_MASK) + new_acl = merge_entry(api, entry, old_acl, f"grantKeyWrite({key})") + except BaseException: + _close_key_best_effort(api, handle) + raise + try: + apply_dacl(api, key, handle, new_acl) + finally: + _free_new_acl(api, new_acl, f"grantKeyWrite({key})") + close_key(api, handle, key) + + +def revoke_key_write(api: Win32Bindings, key: str, sid_ptr: int) -> bool: + """Remove every ACE naming a capability SID from a key's DACL. + + Other entries are preserved: the merge is a ``REVOKE_ACCESS`` of one trustee. + + :param api: the binding table. + :param key: the key whose DACL loses the capability SID's ACEs. + :param sid_ptr: the capability SID whose ACEs are removed. + :returns: whether a removal was attempted (``False`` when the key carries no + explicit DACL — the NULL-DACL case is refused, not treated as empty). + :raises RegistryKeyError: when the spelling is not one this module addresses. + :raises KeyDaclRefused: when the key carries a NULL DACL. + :raises Win32Error: when any call failed. + """ + with with_path_lock(api, key): + handle = open_key(api, key) + try: + dacl = read_current_key_dacl(api, key, handle) + if dacl is None: + close_key(api, handle, key) + return False + entry = build_explicit_access(sid_ptr, abi.REVOKE_ACCESS, 0) + new_acl = merge_entry(api, entry, dacl.pointer, f"revokeKeyWrite({key})") + except BaseException: + _close_key_best_effort(api, handle) + raise + try: + apply_dacl(api, key, handle, new_acl) + finally: + _free_new_acl(api, new_acl, f"revokeKeyWrite({key})") + close_key(api, handle, key) + return True diff --git a/emrg/sandbox/win32/runner.py b/emrg/sandbox/win32/runner.py index 5b78b948..e99bba93 100644 --- a/emrg/sandbox/win32/runner.py +++ b/emrg/sandbox/win32/runner.py @@ -14,7 +14,7 @@ python -m emrg.sandbox.win32.runner --workspace