df12 Productions — Execution context and hooks

Cuprum provides CuprumContext to scope allowlists and execution hooks across your application. Contexts are backed by a ContextVar, giving you automatic isolation across threads and async tasks.

When you call SafeCmd.run() or run_sync(), Cuprum automatically:

Checks the current context's allowlist and raises ForbiddenProgramError if the program is not permitted.
Invokes all registered before hooks (in FIFO order) before process execution.
Invokes all registered after hooks (in LIFO order) after the process completes.

Empty allowlist behaviour: When no context is established (or the context has an empty allowlist), all programs are permitted. This permissive default is intentional to ease adoption but weakens safety; establish an explicit allowlist via scoped() to enforce policy once onboarded.

Scoped contexts

Use scoped() to establish a narrowed execution context within a code block:

from cuprum import ECHO, LS, scoped

# Start with a base allowlist
with scoped(allowlist=frozenset([ECHO, LS])) as ctx:
    assert ctx.is_allowed(ECHO)  # True
    assert ctx.is_allowed(LS)  # True

    # Narrow further in nested scope
    with scoped(allowlist=frozenset([ECHO])) as inner:
        assert inner.is_allowed(ECHO)  # True
        assert inner.is_allowed(LS)  # False (narrowed out)

Key properties of scoped():

When the parent allowlist is empty, the provided allowlist becomes the new base.
When the parent has programs, the new allowlist is intersected (can only narrow, never widen).
Context is automatically restored when the block exits, even on exception.

Accessing the current context

Use current_context() or get_context() to access the current execution context:

from cuprum import ECHO, current_context, scoped

with scoped(allowlist=frozenset([ECHO])):
    ctx = current_context()
    if ctx.is_allowed(ECHO):
        print("ECHO is allowed")

Dynamic allowlist extension

Use allow() to temporarily add programs to the current context:

from cuprum import ECHO, LS, allow, current_context, scoped

with scoped(allowlist=frozenset([ECHO])):
    # LS is not currently allowed
    assert not current_context().is_allowed(LS)

    # Temporarily allow LS
    with allow(LS):
        assert current_context().is_allowed(LS)

    # LS is no longer allowed after the block
    assert not current_context().is_allowed(LS)

For manual control, use the AllowRegistration handle directly:

from cuprum import LS, allow, current_context, scoped

with scoped():
    reg = allow(LS)
    assert current_context().is_allowed(LS)
    reg.detach()  # Remove LS from allowlist
    assert not current_context().is_allowed(LS)

Before and after hooks

Register hooks to run before or after command execution:

from cuprum import ECHO, before, after, scoped, sh


def log_before(cmd):
    print(f"About to run: {cmd.program}")


def log_after(cmd, result):
    print(f"Finished {cmd.program} with exit code {result.exit_code}")


with scoped(allowlist=frozenset([ECHO])):
    with before(log_before), after(log_after):
        cmd = sh.make(ECHO)("hello")
        # Hooks will be invoked when cmd.run() is called

Hook ordering:

Before hooks execute in registration order (FIFO): parent hooks run before child hooks.
After hooks execute in reverse order (LIFO): child hooks run before parent hooks, enabling cleanup patterns.

Like allow(), hook registrations can be detached manually:

from cuprum import before, current_context, scoped


def my_hook(cmd):
    pass


with scoped():
    reg = before(my_hook)
    assert my_hook in current_context().before_hooks
    reg.detach()
    assert my_hook not in current_context().before_hooks

Logging hook

Use logging_hook() to register paired hooks that emit structured start and exit events through the standard library logging module. The helper wires a before hook (start) and after hook (exit) into the current context and returns a registration handle that can be used as a context manager:

import logging

from cuprum import ECHO, logging_hook, scoped, sh

logger = logging.getLogger("myapp.commands")

with scoped(allowlist=frozenset([ECHO])):
    with logging_hook(logger=logger):
        sh.make(ECHO)("-n", "hello logging").run_sync()

By default, the hook logs to logging.getLogger("cuprum") at INFO level. The logger or the log levels can be overridden via start_level and exit_level. Start events include the program and argv; exit events include the program, pid, exit code, duration, and lengths of captured stdout/stderr (zero when capture is disabled).

Structured execution events

For richer observability, register an observe hook with sh.observe(). Observe hooks receive ExecEvent values describing:

plan — intent to execute the program (argv/cwd/env resolved).
start — subprocess spawned (pid available).
stdout / stderr — decoded output emitted as lines.
exit — subprocess finished (exit code and duration).

Hooks can be used for structured logging, metrics, or tracing without coupling Cuprum to a specific telemetry library.

from cuprum import ECHO, ExecEvent, scoped, sh
from cuprum.sh import ExecutionContext


events: list[ExecEvent] = []


def capture(ev: ExecEvent) -> None:
    events.append(ev)


with scoped(allowlist=frozenset([ECHO])), sh.observe(capture):
    ctx = ExecutionContext(tags={"run_id": "demo"})
    sh.make(ECHO)("-n", "hello events").run_sync(context=ctx)

stdout_lines = [ev.line for ev in events if ev.phase == "stdout"]
exit_events = [ev for ev in events if ev.phase == "exit"]
assert "hello events" in stdout_lines
assert exit_events[0].tags["run_id"] == "demo"

ExecutionContext.tags is merged into each event's tags mapping. Cuprum also adds default tags such as the project name and pipeline stage metadata.

Thread and async task isolation

CuprumContext uses Python's ContextVar mechanism, which provides automatic isolation:

Each thread gets its own context value.
Each async task inherits the context from its creator and can modify it independently.

This isolation allows scoped() to be used in concurrent code without context leaking between threads or tasks:

import asyncio

from cuprum import ECHO, LS, current_context, scoped


async def worker(name: str, programs):
    with scoped(allowlist=programs):
        await asyncio.sleep(0.1)  # Simulate work
        ctx = current_context()
        print(f"{name}: ECHO allowed = {ctx.is_allowed(ECHO)}")


async def main():
    await asyncio.gather(
        worker("task1", frozenset([ECHO])),
        worker("task2", frozenset([LS])),
    )
    # task1 sees ECHO allowed, task2 does not


asyncio.run(main())

Checking allowlist membership

Use is_allowed() to check if a program is permitted:

from cuprum import ECHO, CuprumContext

ctx = CuprumContext(allowlist=frozenset([ECHO]))
if ctx.is_allowed(ECHO):
    print("ECHO is allowed")

Use check_allowed() to raise ForbiddenProgramError if a program is not allowed:

from cuprum import ECHO, LS, CuprumContext, ForbiddenProgramError

ctx = CuprumContext(allowlist=frozenset([ECHO]))
try:
    ctx.check_allowed(LS)
except ForbiddenProgramError as e:
    print(f"Access denied: {e}")