Scopes, policy, and hooks

def scoped(
    config: ScopeConfig | None = None,
    *,
    catalogue: ProgramCatalogue | None = None,
) -> _ScopedContext

Create a scoped context manager for narrowed execution.

>>> with scoped(ScopeConfig(allowlist=frozenset([ECHO]))) as ctx:
...     assert ctx.is_allowed(ECHO)
>>> catalogue = ProgramCatalogue.from_programs(ECHO)
>>> with scoped(catalogue=catalogue) as ctx:
...     assert ctx.is_allowed(ECHO)

class ScopeConfig(
    allowlist: frozenset[Program] | None = None,
    before_hooks: tuple[BeforeHook, ...] = (),
    after_hooks: tuple[AfterHook, ...] = (),
    observe_hooks: tuple[ExecHook, ...] = (),
    timeout: float | None = None,
    env_overlay: EnvOverlay | None = None,
    env_mode: EnvMode = EnvMode.OVERLAY,
    catalogue: ProgramCatalogue | None = None,
    executable_bindings: ExecutableBindingOverlay | None = None,
)

Bases: _ExecutableBindingSource

Configuration object for scoped execution context updates.

allowlist: frozenset[Program] | None

before_hooks: tuple[BeforeHook, ...]

after_hooks: tuple[AfterHook, ...]

observe_hooks: tuple[ExecHook, ...]

timeout: float | None

env_overlay: EnvOverlay | None

env_mode: EnvMode

catalogue: ProgramCatalogue | None

executable_bindings: ExecutableBindingOverlay | None

def executable_binding(program: Program) -> ExecutableBinding | None

>>> from cuprum.context import CuprumContext
>>> CuprumContext().executable_binding("echo") is None
True

class CuprumContext(
    allowlist: frozenset[Program] = dc.field(default_factory=frozenset),
    before_hooks: tuple[BeforeHook, ...] = (),
    after_hooks: tuple[AfterHook, ...] = (),
    observe_hooks: tuple[ExecHook, ...] = (),
    timeout: float | None = None,
    env_overlay: EnvOverlay | None = None,
    _allowlist_is_restricted: bool = False,
    env_mode: EnvMode = EnvMode.OVERLAY,
    catalogue: ProgramCatalogue | None = dc.field(default=None, compare=False),
    executable_bindings: ExecutableBindingOverlay | None = None,
)

Bases: _HookPolicy, _ExecutableBindingPolicy

Immutable execution context holding allowlist and hooks.

allowlist: frozenset[Program]

before_hooks: tuple[BeforeHook, ...]

after_hooks: tuple[AfterHook, ...]

observe_hooks: tuple[ExecHook, ...]

timeout: float | None

env_overlay: EnvOverlay | None

env_mode: EnvMode

catalogue: ProgramCatalogue | None

executable_bindings: ExecutableBindingOverlay | None

def is_allowed(program: Program) -> bool

def check_allowed(program: Program) -> None

def narrow(config: ScopeConfig) -> CuprumContext

def with_allowlist(allowlist: frozenset[Program]) -> CuprumContext

def with_program(program: Program) -> CuprumContext

def without_program(program: Program) -> CuprumContext

def with_env_overlay(
    overlay: EnvOverlay | None,
    mode: EnvMode = EnvMode.OVERLAY,
) -> CuprumContext

def with_before_hook(hook: BeforeHook) -> typ.Self

def without_before_hook(hook: BeforeHook) -> typ.Self

def with_after_hook(hook: AfterHook) -> typ.Self

def without_after_hook(hook: AfterHook) -> typ.Self

def with_observe_hook(hook: ExecHook) -> typ.Self

def without_observe_hook(hook: ExecHook) -> typ.Self

def with_executable_binding(
    program: Program,
    binding: ExecutableBinding,
) -> typ.Self

>>> from cuprum.context import CuprumContext
>>> from cuprum.executable_binding import executable_binding
>>> ctx = CuprumContext().with_executable_binding(
...     "echo", executable_binding("echo", "/bin/echo")
... )
>>> ctx.resolve_executable("echo", cwd=None)
'/bin/echo'

def resolve_executable(
    program: Program,
    *,
    cwd: str | None,
) -> str | None

>>> from cuprum.context import CuprumContext
>>> CuprumContext().resolve_executable("echo", cwd=None) is None
True

def executable_binding(program: Program) -> ExecutableBinding | None

>>> from cuprum.context import CuprumContext
>>> CuprumContext().executable_binding("echo") is None
True

def current_context() -> CuprumContext

Return the current execution context.

def allow(*programs: Program) -> AllowRegistration

Extend the current context's allowlist with additional programs.

>>> with allow(LS):
...     assert current_context().is_allowed(LS)

class AllowRegistration(*programs: Program) -> None

Bases: _TokenRegistration

Registration handle for dynamic allowlist extension.

Supports detach() and context-manager usage for scoped allowing. The token-restoration discipline is documented on _TokenRegistration.

def detach() -> None

def env(
    *overlays: cabc.Mapping[str, EnvOverlayValue],
    **kwvars: EnvOverlayValue | EnvMode,
) -> EnvRegistration

Overlay environment variables on top of the live os.environ.

Mirrors dict in how arguments are combined: positional mappings are merged left-to-right and any keyword arguments win over them. For effective modes other than REPLACE, values are not snapshot against os.environ — the live process environment is read at subprocess spawn time so that variables set after Cuprum is imported (for example by monkeypatch.setenv) remain visible. A REPLACE mode renders from an empty environment instead, so the live process environment stays out of the child.

>>> import os
>>> os.environ["GIT_AUTHOR_NAME"] = "Cuprum"
>>> with env(PATH="/usr/bin"):
...     # Subprocesses spawned here see PATH=/usr/bin overlaid on the
...     # *live* os.environ, including GIT_AUTHOR_NAME.
...     pass

class EnvRegistration(
    overlay: EnvOverlay,
    mode: EnvMode = EnvMode.OVERLAY,
) -> None

Bases: _TokenRegistration

Registration handle for a scoped environment overlay.

The overlay is layered on top of any overlay already present in the current context; nested registrations therefore behave as a stack. The token-restoration discipline is documented on _TokenRegistration.

The overlay itself is overlay-only. For effective modes other than REPLACE, the live os.environ is read at subprocess spawn time, when render_env renders the composed policy, so any updates to the process environment after the registration is created — for example via pytest's monkeypatch.setenv — remain visible to subprocesses spawned inside the scope. This is the behaviour the issue requires. A REPLACE mode is the exception: it renders from an empty environment, so the live process environment stays out of the child entirely.

overlay: EnvOverlay | None

def detach() -> None

def merge_env_overlays(
    parent: EnvOverlay | None,
    child: EnvOverlay | None,
) -> EnvOverlay | None

Layer child over parent; None means *inherit unchanged*.

Both layers are kept overlay-only — they never include a snapshot of os.environ. The live process environment is read at spawn time so that callers can monkey-patch or otherwise mutate os.environ after Cuprum has been imported and still have those updates visible to subprocesses spawned inside the scope.

Exposed as public API so that other cuprum modules — and downstream code that builds custom observation tags — can merge overlay layers without reaching for a private symbol.

def resolve_env(*layers: EnvOverlay | None) -> dict[str, str] | None

Merge os.environ (read live) with the supplied overlay layers.

Layers are applied left-to-right; later values win. None *and empty* layers are skipped — an empty overlay contributes nothing, so treating it as a no-op avoids an unnecessary copy of os.environ and lets the subprocess inherit the parent environment directly. When every layer is skipped the function returns None so callers may pass it through to subprocess APIs to mean *inherit the parent environment unchanged*.

The call to os.environ.copy is deferred until at least one overlay is non-empty, so the result reflects whatever the process environment looks like at the moment of resolution — the exact behaviour the issue requires.

from cuprum import context

Execution context with scoped allowlists and hooks.

CuprumContext provides a ContextVar-backed execution context that scopes allowlists and hooks for command execution. Contexts support narrowing (restricting the allowlist) and hook registration with deterministic ordering.

Example: >>> from cuprum.context import ScopeConfig, scoped, before, current_context >>> from cuprum.catalogue import ECHO >>> def log_hook(cmd): ... print(f"Running: {cmd}") >>> with scoped(ScopeConfig(allowlist=frozenset([ECHO]))): ... with before(log_hook): ... ctx = current_context() ... ctx.is_allowed(ECHO) True

Scoped environment overlays (issue #100): four public symbols layer environment variables on top of the live os.environ for subprocesses spawned inside the scope. env(*overlays, **kwvars) is a factory that returns an EnvRegistration applying an immutable overlay to the active context, resolved live at subprocess spawn time against os.environ. EnvRegistration is the handle returned by env(); it works as a context manager and exposes detach for explicit teardown. merge_env_overlays(parent, child) is the overlay-only merge that returns an immutable MappingProxyType without reading os.environ — it is the helper used to record the effective overlay in observation events. resolve_env(*layers) is an overlay-only helper that merges os.environ with one or more overlay layers and returns a plain dict or None when no layers contribute; it is not on the subprocess spawn path. Spawning composes through _resolve_env_policy and renders through render_env, which is what consults the effective EnvMode. Precedence at spawn time, from lowest to highest, is: live os.environ < scoped env() overlays (innermost wins) < per-call ExecutionContext.env. This deliberately diverges from plumbum's local.env, which snapshots os.environ at module import time and therefore misses variables added to the process environment after import. The CuprumContext and ScopeConfig dataclasses each carry an env_overlay field (Mapping[str, str] | None) that is coerced to an immutable proxy on construction.

>>> from cuprum.context import env, current_context >>> with env(MY_VAR="hello"): ... current_context().env_overlay["MY_VAR"] 'hello'

Scoped executable bindings (issue #440): bind_executable(program, path_or_resolver, *, allow_relative=False) binds a logical catalogued program to the executable it should run inside the scope, and returns an ExecutableBindingRegistration handle that detaches or acts as a context manager. The CuprumContext and ScopeConfig dataclasses each carry an executable_bindings field coerced to an immutable proxy on construction; merge_executable_bindings(parent, child) records the effective layer without resolving it. Bindings are a second, narrower authority than the allowlist: the allowlist decides which logical programs may run at all, and a binding decides only which executable a permitted program runs. Resolution therefore happens strictly after allowlist enforcement, and CuprumContext.resolve_executable never consults is_allowed or check_allowed.

>>> from cuprum.context import bind_executable, current_context >>> from cuprum.catalogue import ECHO >>> with bind_executable(ECHO, "/opt/tools/echo"): ... current_context().resolve_executable(ECHO, cwd=None) '/opt/tools/echo'

UNSET = UnsetType()

ContextError

EnvMode

UnsetType

def bind_executable(
    program: Program,
    path_or_resolver: str | os.PathLike[str] | ExecutableResolver,
    *,
    allow_relative: bool = False,
) -> ExecutableBindingRegistration

Bind a logical program to an executable for the enclosing scope.

The binding applies only inside the scope; leaving it restores the bindings that were in effect before. A nested registration for the same program overrides the outer one for the inner scope alone, and bindings for distinct programs compose.

>>> with bind_executable(ECHO, "/opt/tools/echo"):
...     # Commands for ECHO run /opt/tools/echo inside this scope.
...     pass

class ExecutableBindingRegistration(
    program: Program,
    binding: ExecutableBinding,
) -> None

Bases: _TokenRegistration

Registration handle for a scoped executable binding.

The binding is layered onto any bindings already present in the current context, so nested registrations behave as a stack: the innermost binding for a program is the one an execution resolves. The token-restoration discipline is documented on _TokenRegistration.

A binding supplies the executable a permitted program runs. It is not a permission: the allowlist continues to decide which logical programs may run at all, and resolution happens only after that decision has been made.

program: Program

binding: ExecutableBinding

def detach() -> None

def merge_executable_bindings(
    parent: ExecutableBindingOverlay | None,
    child: ExecutableBindingOverlay | None,
) -> ExecutableBindingOverlay | None

Layer child over parent; None means *inherit unchanged*.

The child layer wins key by key: an inner scope can rebind a program the outer scope already bound, and can add bindings the outer scope never mentioned. Composition never copies a binding — entries move between layers by identity, so a resolver stays the same callable wherever it is read from.

>>> merged = merge_executable_bindings({"echo": "outer"}, {"echo": "inner"})
>>> merged["echo"]
'inner'

from cuprum import executable_binding

Typed bindings from a logical program identity to an executable path.

A Program names an executable for catalogue and policy purposes, and the same value is handed to the operating system as argv[0]. Those two jobs usually coincide, but not always: a tool may live at a version-pinned absolute path, inside a virtual environment, or be replaced by a controlled stub during tests. This module carries the second job as its own value so the first can stay a bounded logical label.

The path vocabulary this module builds on — ExecutablePath, PathBindingRejection, executable_path, and advisory_path_rejection — lives in executable_paths and is re-exported here, so callers reach the whole feature through one import.

Both modules are pure: neither reads or writes CuprumContext, and neither imports the context package at runtime. That keeps the dependency direction acyclic, so the context package and the execution layer can both depend on them.

Resolution and its limits -------------------------

resolve_binding turns a binding plus a working directory into the string to execute. It is pure and total for every binding the constructors accept.

Checking a path and then executing it is not atomic. Between the advisory check and the child's exec, another process with write access to the directory can replace, rename, or re-link the file, and the child will run whatever is there at that moment. Cuprum cannot close that window from inside the process: it would need an O_PATH descriptor plus fexecve, which is not available portably or from asyncio. Operators who need the guarantee that the checked path is the executed binary must supply it from the filesystem, by owning the directory, restricting write permission to a trusted account, and deploying the tree read-only.

class ExecutableBinding(
    path: ExecutablePath | None = None,
    resolver: ExecutableResolver | None = None,
)

ExecutablePath = typ.NewType('ExecutablePath', str)

class ExecutableResolutionError(
    program: Program,
    cause: BaseException,
) -> None

>>> from cuprum.context import CuprumContext
>>> def missing() -> str:
...     raise FileNotFoundError("no venv")
>>> binding = executable_binding("echo", missing)
>>> ctx = CuprumContext().with_executable_binding("echo", binding)
>>> try:
...     ctx.resolve_executable("echo", cwd=None)
... except ExecutableResolutionError as err:
...     print(err)
Program 'echo' cannot resolve its executable: FileNotFoundError

ExecutableResolver = cabc.Callable[[], str]

class InvalidExecutableBindingError(
    program: Program | None,
    path: str,
    reason: PathBindingRejection,
) -> None

PathBindingRejection

def advisory_path_rejection(
    raw_value: str,
) -> PathBindingRejection | None

>>> advisory_path_rejection("tool") is None
True

def classify_executable_path(
    raw_value: str,
    *,
    allow_relative: bool,
) -> PathBindingRejection | None

>>> classify_executable_path("/opt/tools/tool", allow_relative=False) is None
True
>>> classify_executable_path("bin/tool", allow_relative=False).name
'NOT_ABSOLUTE'
>>> classify_executable_path("bin/tool", allow_relative=True) is None
True
>>> classify_executable_path(r"C:\tools\..\evil.exe", allow_relative=False)
<PathBindingRejection.PARENT_SEGMENT: "ExecutablePath cannot contain '..' segments">

def executable_binding(
    program: Program,
    path_or_resolver: str | os.PathLike[str] | ExecutableResolver,
    *,
    allow_relative: bool = False,
) -> ExecutableBinding

>>> binding = executable_binding("tool", "/opt/tools/tool")
>>> binding.path
'/opt/tools/tool'
>>> binding.is_lazy
False

def executable_path(
    value: str | os.PathLike[str],
    *,
    allow_relative: bool = False,
) -> ExecutablePath

>>> executable_path("/opt//tools/./tool")
'/opt/tools/tool'
>>> executable_path("bin/tool", allow_relative=True)
'bin/tool'

def resolve_binding(
    binding: ExecutableBinding,
    *,
    cwd: str | None,
) -> str

>>> resolve_binding(
...     executable_binding("tool", "bin/tool", allow_relative=True),
...     cwd="/srv/project",
... )
'/srv/project/bin/tool'

def before(hook: BeforeHook) -> HookRegistration

Register a before-execution hook in the current context.

>>> def log_cmd(cmd):
...     print(f"Running: {cmd.program}")
>>> with before(log_cmd):
...     # Commands run here will trigger log_cmd
...     pass

def after(hook: AfterHook) -> HookRegistration

Register an after-execution hook in the current context.

>>> def log_result(cmd, result):
...     print(f"Finished: {cmd.program} -> {result.exit_code}")
>>> with after(log_result):
...     # Commands run here will trigger log_result
...     pass

class HookRegistration(
    hook: BeforeHook | AfterHook | ExecHook,
    hook_type: typ.Literal['before', 'after', 'observe'],
) -> None

Bases: _TokenRegistration

Registration handle for hooks with detach and context-manager support.

The token-restoration discipline is documented on _TokenRegistration.

Prefer the before, after, and observe factories: each pairs one hook shape with its slot, so a mismatched hook is rejected at the call site. Constructing the handle directly pairs the hook and the tag by hand, and only the tag itself is validated — at runtime.

def detach() -> None