terok_util

terok_util

terok-util — shared utility library for the terok-* sibling packages.

terok-util sits at the bottom of the terok dependency chain. Every sibling package depends on it; it depends on nothing else in the ecosystem (only stdlib + platformdirs + ruamel.yaml). It collects the small set of cross-cutting helpers that would otherwise be duplicated — or, worse, quietly diverge — across terok-shield, terok-clearance, terok-sandbox, terok-executor, and terok.

What lives here, by module:

• cli_types — argparse-driven CLI registry vocabulary: CommandDef, ArgDef, CommandTree, KeyRow.
• fs — small filesystem helpers (ensure_dir, ensure_dir_writable, write_sensitive_file).
• logging — soft-fail file logger (BestEffortLogger).
• yaml — round-trip YAML facade over ruamel.yaml (load, dump).
• paths — XDG-aware namespace path resolution (namespace_state_dir, namespace_config_dir, namespace_runtime_dir).
• config_stack — layered config merge engine (ConfigStack, deep_merge).
• security — untrusted-string TTY sanitiser (sanitize_tty).
• podman — rootless --userns=keep-id builder (podman_userns_args).

The rule for what belongs here: if two or more terok-* packages need it, it lives in terok-util. Single-package helpers stay in the package that owns them. The __all__ declaration below is the contract — symbols listed are stable across minor releases; anything underscore-prefixed or absent from __all__ is internal and may change without notice.

__all__ = ['ArgDef', 'BestEffortLogger', 'CommandDef', 'CommandTree', 'ConfigStack', 'KeyRow', 'deep_merge', 'ensure_dir', 'ensure_dir_writable', 'config_file_paths', 'host_uid', 'namespace_config_dir', 'namespace_runtime_dir', 'namespace_state_dir', 'podman_userns_args', 'read_config_section', 'read_config_top_level', 'sanitize_tty', 'write_sensitive_file'] module-attribute

ArgDef(name, help='', type=None, default=None, action=None, dest=None, nargs=None, required=False) dataclass

Definition of a single CLI argument.

name instance-attribute

help = '' class-attribute instance-attribute

type = None class-attribute instance-attribute

default = None class-attribute instance-attribute

action = None class-attribute instance-attribute

dest = None class-attribute instance-attribute

nargs = None class-attribute instance-attribute

required = False class-attribute instance-attribute

CommandDef(name, help='', handler=None, args=(), children=(), group='', epilog='', extras=dict()) dataclass

One node in a command tree — a leaf verb or a group of verbs.

Attributes:

Name	Type	Description
name	str	Verb name as it appears on the CLI.
help	str	One-line help string.
handler	Callable[..., Any] | None	Callable implementing the verb. None for groups.
args	tuple[ArgDef, ...]	Argument definitions parsed by argparse.
children	tuple[CommandDef, ...]	Sub-verbs. Non-empty makes this node a group.
group	str	Free-form tag used by per-subsystem grouping (unrelated to the children structural nesting).
epilog	str	Optional long-form text rendered after the argparse argument list in --help output.
extras	Mapping[str, Any]	Bag of package-specific metadata downstream consumers ignore (shield's needs_container / standalone_only would live here on a unified shape).

A frozen-dataclass + structural sharing is the load-bearing part of the wrap-once-share-everywhere story: when a consumer overlays a handler at one path, the modified CommandDef is referenced from every shortcut that also points at that path. Identity is what makes the overlay propagate.

name instance-attribute

help = '' class-attribute instance-attribute

handler = None class-attribute instance-attribute

args = () class-attribute instance-attribute

children = () class-attribute instance-attribute

group = '' class-attribute instance-attribute

epilog = '' class-attribute instance-attribute

extras = field(default_factory=dict) class-attribute instance-attribute

is_group property

Whether this node carries children (i.e. is a verb group).

with_handler(handler)

Return a copy with handler replaced — pure leaf-rewrap.

Source code in src/terok_util/cli_types.py

def with_handler(self, handler: Callable[..., Any]) -> CommandDef:
    """Return a copy with ``handler`` replaced — pure leaf-rewrap."""
    return replace(self, handler=handler)

with_children(children)

Return a copy with children replaced.

Source code in src/terok_util/cli_types.py

def with_children(self, children: tuple[CommandDef, ...]) -> CommandDef:
    """Return a copy with ``children`` replaced."""
    return replace(self, children=children)

CommandTree(roots)

A forest of CommandDef nodes.

The unit of composition for CLI registries: each package exposes its own CommandTree; consumers walk it structurally, overlay handlers where they wrap a concept, extend with their own verbs, and wire the result into argparse.

Composition is identity-preserving — nodes the consumer doesn't touch share object identity with their pre-overlay counterparts, so a shortcut that splices the same subtree at the consumer's top level reaches the same modified handler. terok shield install and terok executor sandbox shield install resolving to the same wrap is a direct consequence.

Build a tree from an iterable of top-level verbs/groups.

Source code in src/terok_util/cli_types.py

def __init__(self, roots: Iterable[CommandDef]) -> None:
    """Build a tree from an iterable of top-level verbs/groups."""
    self._roots: tuple[CommandDef, ...] = tuple(roots)

roots property

The top-level verbs in this tree, in declaration order.

__iter__()

Yield each root verb.

Source code in src/terok_util/cli_types.py

def __iter__(self) -> Iterator[CommandDef]:
    """Yield each root verb."""
    return iter(self._roots)

__len__()

Number of root verbs.

Source code in src/terok_util/cli_types.py

def __len__(self) -> int:
    """Number of root verbs."""
    return len(self._roots)

__add__(other)

Concatenate forests — other's roots appended to this one's.

Source code in src/terok_util/cli_types.py

def __add__(self, other: CommandTree | Iterable[CommandDef]) -> CommandTree:
    """Concatenate forests — *other*'s roots appended to this one's."""
    other_roots = other.roots if isinstance(other, CommandTree) else tuple(other)
    return CommandTree(self._roots + other_roots)

find_at(path)

Return the CommandDef at path.

path is a sequence of verb names from the root. An empty path is rejected (no synthetic root). KeyError if any segment doesn't match a child name.

Source code in src/terok_util/cli_types.py

def find_at(self, path: Sequence[str]) -> CommandDef:
    """Return the [`CommandDef`][terok_util.cli_types.CommandDef] at *path*.

    *path* is a sequence of verb names from the root.  An empty
    path is rejected (no synthetic root).  ``KeyError`` if any
    segment doesn't match a child name.
    """
    if not path:
        raise KeyError("empty path; specify at least one verb name")
    first, *rest = path
    for root in self._roots:
        if root.name == first:
            return _descend(root, tuple(rest))
    raise KeyError(f"no top-level verb {first!r}")

overlay(overrides)

Return a new tree with handlers replaced at the named paths.

overrides maps verb-name tuples (e.g. ("vault", "status")) to replacement handlers. Each match produces one new CommandDef via replace; ancestors are likewise replaced because their children tuples now hold a new node, but unrelated siblings share identity with the input tree.

Sandbox-vocab paths use the operator-facing verb names — same names you'd type on the CLI — so the override map reads like a routing table.

Source code in src/terok_util/cli_types.py

def overlay(self, overrides: Mapping[tuple[str, ...], Callable[..., Any]]) -> CommandTree:
    """Return a new tree with handlers replaced at the named paths.

    *overrides* maps verb-name tuples (e.g. ``("vault", "status")``)
    to replacement handlers.  Each match produces one new
    [`CommandDef`][terok_util.cli_types.CommandDef] via ``replace``;
    ancestors are likewise replaced because their ``children``
    tuples now hold a new node, but unrelated siblings share
    identity with the input tree.

    Sandbox-vocab paths use the operator-facing verb names — same
    names you'd type on the CLI — so the override map reads like a
    routing table.
    """
    return CommandTree(_overlay_forest(self._roots, dict(overrides), ()))

extend_at(path, additions)

Return a new tree with additions appended at the path's children.

Empty path extends the top-level forest. Otherwise the CommandDef at path must be a group — a leaf (one with handler set and no children) cannot be extended; trying to do so would produce a hybrid node argparse can't represent (handler + subparsers on the same parser). Raises TypeError rather than silently inventing such a node.

Source code in src/terok_util/cli_types.py

def extend_at(self, path: Sequence[str], additions: Iterable[CommandDef]) -> CommandTree:
    """Return a new tree with *additions* appended at the path's children.

    Empty path extends the top-level forest.  Otherwise the
    [`CommandDef`][terok_util.cli_types.CommandDef] at *path*
    must be a **group** — a leaf (one with ``handler`` set and no
    ``children``) cannot be extended; trying to do so would produce
    a hybrid node argparse can't represent (handler + subparsers
    on the same parser).  Raises ``TypeError`` rather than
    silently inventing such a node.
    """
    addition_tuple = tuple(additions)
    if not path:
        return CommandTree(self._roots + addition_tuple)
    target = self.find_at(path)
    if target.handler is not None and not target.children:
        raise TypeError(
            f"cannot extend leaf {'.'.join(path)!r}: extend_at requires a group "
            f"(handler=None with children), but the target has a handler set"
        )
    return CommandTree(_extend_forest(self._roots, tuple(path), addition_tuple, ()))

walk()

Yield (path, command) for every node in the tree, depth-first.

Source code in src/terok_util/cli_types.py

def walk(self) -> Iterator[tuple[tuple[str, ...], CommandDef]]:
    """Yield ``(path, command)`` for every node in the tree, depth-first."""
    for root in self._roots:
        yield from _walk_node(root, ())

wire(target)

Wire this tree's verbs as subparsers under target, recursively.

target may be either an ArgumentParser (a fresh add_subparsers() action is created) or an existing argparse._SubParsersAction (the tree mounts straight under it; the private name has no public docs target). The second form lets a consumer mix legacy register-style subparsers with structural CommandTree ones under the same root parser without colliding on argparse's one-subparsers-per-parser rule.

The same CommandDef wired at multiple positions (deep nesting + shortcuts) yields independent argparse subparser instances, but each subparser's dispatch reads back the same handler object — so concept translations applied via overlay apply uniformly across every entry point that references the modified node.

Source code in src/terok_util/cli_types.py

def wire(self, target: argparse.ArgumentParser | argparse._SubParsersAction) -> None:
    """Wire this tree's verbs as subparsers under *target*, recursively.

    *target* may be either an
    [`ArgumentParser`][argparse.ArgumentParser] (a fresh
    ``add_subparsers()`` action is created) or an existing
    ``argparse._SubParsersAction`` (the tree mounts straight under
    it; the private name has no public docs target).  The second
    form lets a consumer mix legacy register-style subparsers
    with structural
    [`CommandTree`][terok_util.cli_types.CommandTree] ones under
    the same root parser without colliding on argparse's
    one-subparsers-per-parser rule.

    The same [`CommandDef`][terok_util.cli_types.CommandDef]
    wired at multiple positions (deep nesting + shortcuts) yields
    independent argparse subparser instances, but each subparser's
    dispatch reads back the same handler object — so concept
    translations applied via ``overlay`` apply uniformly across
    every entry point that references the modified node.
    """
    if isinstance(target, argparse._SubParsersAction):
        sub = target
    else:
        sub = target.add_subparsers()
    for cmd in self._roots:
        _wire_command(sub, cmd)

dispatch(args) staticmethod

Invoke the handler stored on args by CommandTree.wire.

Bridges argparse's parsed-args namespace to the handler kwargs the CommandDef declared. Async handlers are detected and run via asyncio.run so consumers don't need separate dispatch paths per handler flavour.

Source code in src/terok_util/cli_types.py

@staticmethod
def dispatch(args: argparse.Namespace) -> None:
    """Invoke the handler stored on *args* by [`CommandTree.wire`][terok_util.cli_types.CommandTree.wire].

    Bridges argparse's parsed-args namespace to the handler kwargs
    the [`CommandDef`][terok_util.cli_types.CommandDef] declared.
    Async handlers are detected and run via ``asyncio.run`` so
    consumers don't need separate dispatch paths per handler
    flavour.
    """
    cmd: CommandDef | None = getattr(args, "_cmd", None)
    if cmd is None:
        # Argparse stopped before a leaf command (bare ``terok`` /
        # ``terok shield`` etc.).  Argparse already printed help; exit
        # with the same "no command given" status it would.
        raise SystemExit(2)
    if cmd.handler is None:
        raise SystemExit(f"Command {cmd.name!r} has no handler")
    kwargs = {_arg_dest(arg): getattr(args, _arg_dest(arg), arg.default) for arg in cmd.args}
    # Trailing args after ``--`` (currently only ``run`` consumes them).
    if hasattr(args, "podman_args"):
        kwargs["podman_args"] = args.podman_args
    result = cmd.handler(**kwargs)
    if inspect.iscoroutine(result):
        import asyncio  # noqa: PLC0415

        asyncio.run(result)

KeyRow

Bases: NamedTuple

One registered SSH key, fully resolved for display and matching.

scope instance-attribute

comment instance-attribute

key_type instance-attribute

fingerprint instance-attribute

private_key instance-attribute

public_key instance-attribute

ConfigStack()

Ordered collection of config scopes, lowest-priority first.

Usage::

stack = ConfigStack()
stack.push(ConfigScope("global", global_path, global_data))
stack.push(ConfigScope("project", proj_path, proj_data))
resolved = stack.resolve()

Initialise an empty config stack.

Source code in src/terok_util/config_stack.py

def __init__(self) -> None:
    """Initialise an empty config stack."""
    self._scopes: list[ConfigScope] = []

scopes property

Read-only access to the scope list (for diagnostics).

push(scope)

Append a scope (higher priority than all previous).

Source code in src/terok_util/config_stack.py

def push(self, scope: ConfigScope) -> None:
    """Append a scope (higher priority than all previous)."""
    self._scopes.append(scope)

resolve()

Deep-merge all scopes in order and return the result.

Source code in src/terok_util/config_stack.py

def resolve(self) -> dict:
    """Deep-merge all scopes in order and return the result."""
    result: dict = {}
    for scope in self._scopes:
        result = deep_merge(result, scope.data)
    return result

resolve_section(key)

Resolve only a single top-level section across all scopes.

Respects the same semantics as resolve — in particular, None values trigger deletion via deep_merge.

Returns {} when the highest-priority scope has a non-dict value for key (e.g. services: tcp instead of services: {mode: tcp}). Callers can call resolve and inspect the raw shape if they need to distinguish "missing" from "wrong-shape", but resolve_section is contract-typed as a mapping accessor and so coerces non-mappings to empty.

Source code in src/terok_util/config_stack.py

def resolve_section(self, key: str) -> dict:
    """Resolve only a single top-level section across all scopes.

    Respects the same semantics as
    [`resolve`][terok_util.config_stack.ConfigStack.resolve] — in
    particular, ``None`` values trigger deletion via
    [`deep_merge`][terok_util.config_stack.deep_merge].

    Returns ``{}`` when the highest-priority scope has a non-dict
    value for *key* (e.g. ``services: tcp`` instead of
    ``services: {mode: tcp}``).  Callers can call
    [`resolve`][terok_util.config_stack.ConfigStack.resolve] and
    inspect the raw shape if they need to distinguish "missing"
    from "wrong-shape", but ``resolve_section`` is contract-typed
    as a mapping accessor and so coerces non-mappings to empty.
    """
    wrapper: dict = {}
    for scope in self._scopes:
        if key in scope.data:
            wrapper = deep_merge(wrapper, {key: scope.data[key]})
    section = wrapper.get(key, {})
    return section if isinstance(section, dict) else {}

BestEffortLogger(log_path_fn)

Append timestamped lines to a state-file log; soft-fail on any error.

The destination is supplied as a callable rather than an eager Path so XDG / env-var overrides applied between construction and write time still take effect.

Parameters:

Name	Type	Description	Default
log_path_fn	Callable[[], Path]	Zero-arg callable returning the destination path. Called on every write so tests overriding HOME / XDG_STATE_HOME see their override applied even when the logger was constructed under the previous environment.	required

Bind the destination resolver.

Source code in src/terok_util/logging.py

def __init__(self, log_path_fn: Callable[[], Path]) -> None:
    """Bind the destination resolver."""
    self._log_path_fn = log_path_fn

log(message, *, level='DEBUG')

Append one [timestamp] LEVEL: message line. Never raises.

File creation goes through os.open with mode 0o600 so the log lands owner-only by construction — atomically, without relying on the process umask. The mode bits are honoured by the kernel only on creation; existing files keep whatever perms they were created with.

Source code in src/terok_util/logging.py

def log(self, message: str, *, level: str = "DEBUG") -> None:
    """Append one ``[timestamp] LEVEL: message`` line.  Never raises.

    File creation goes through ``os.open`` with mode ``0o600`` so the
    log lands owner-only by construction — atomically, without
    relying on the process umask.  The mode bits are honoured by
    the kernel only on creation; existing files keep whatever perms
    they were created with.
    """
    try:
        log_path = self._log_path_fn()
        log_path.parent.mkdir(parents=True, exist_ok=True)
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        fd = os.open(
            log_path,
            os.O_APPEND | os.O_CREAT | os.O_WRONLY,
            0o600,
        )
        with os.fdopen(fd, "a", encoding="utf-8") as f:
            f.write(f"[{timestamp}] {level}: {message}\n")
    except Exception:  # nosec B110 — intentionally silent
        pass

debug(message)

Append a DEBUG-level line.

Source code in src/terok_util/logging.py

def debug(self, message: str) -> None:
    """Append a DEBUG-level line."""
    self.log(message, level="DEBUG")

warning(message)

Append a WARNING-level line.

Source code in src/terok_util/logging.py

def warning(self, message: str) -> None:
    """Append a WARNING-level line."""
    self.log(message, level="WARNING")

warn_user(component, message)

Print a structured warning to stderr and append it to the log file.

Stderr output is run through sanitize_tty so attacker bytes in component / message (e.g. originating from foreign config files) can't smuggle terminal escapes into the operator's terminal. The file-side write is unsanitised so the log keeps the original bytes for forensic review.

Source code in src/terok_util/logging.py

def warn_user(self, component: str, message: str) -> None:
    """Print a structured warning to stderr and append it to the log file.

    Stderr output is run through
    [`sanitize_tty`][terok_util.security.sanitize_tty] so attacker
    bytes in *component* / *message* (e.g. originating from foreign
    config files) can't smuggle terminal escapes into the operator's
    terminal.  The file-side write is unsanitised so the log keeps
    the original bytes for forensic review.
    """
    try:
        print(
            f"Warning [{sanitize_tty(component)}]: {sanitize_tty(message)}",
            file=sys.stderr,
        )
    except Exception:  # nosec B110 — intentionally silent
        pass
    self.warning(f"[{component}] {message}")

deep_merge(base, override)

Recursively merge override into base, returning a new dict.

Rules

• Dicts are merged recursively by default.
• A None value in override deletes the corresponding key.
• A bare "_inherit" string keeps the base value unchanged (equivalent to omitting the key, but explicit).
• Lists in override replace the base list wholesale unless the list contains the sentinel string "_inherit", in which case the sentinel is replaced by the base list elements (splice).
• A dict in override that contains _inherit: true keeps all parent keys and overlays the rest (the _inherit key itself is stripped from the result).

Source code in src/terok_util/config_stack.py

def deep_merge(base: dict, override: dict) -> dict:
    """Recursively merge *override* into *base*, returning a **new** dict.

    Rules
    -----
    * Dicts are merged recursively by default.
    * A ``None`` value in *override* **deletes** the corresponding key.
    * A bare ``"_inherit"`` string keeps the base value unchanged
      (equivalent to omitting the key, but explicit).
    * Lists in *override* replace the base list wholesale **unless** the
      list contains the sentinel string ``"_inherit"``, in which case the
      sentinel is replaced by the base list elements (splice).
    * A dict in *override* that contains ``_inherit: true`` keeps all
      parent keys and overlays the rest (the ``_inherit`` key itself is
      stripped from the result).
    """
    merged: dict = {}

    all_keys = set(base) | set(override)
    for key in all_keys:
        if key in override:
            ov = override[key]
            # None → delete
            if ov is None:
                continue
            # Bare _inherit string → keep base value (explicit no-op)
            if ov == _INHERIT:
                if key in base:
                    merged[key] = base[key]
                continue
            bv = base.get(key)
            if isinstance(ov, dict) and isinstance(bv, dict):
                merged[key] = _merge_dicts(bv, ov)
            elif isinstance(ov, list) and isinstance(bv, list):
                merged[key] = _merge_lists(bv, ov)
            elif isinstance(ov, dict) and ov.get(_INHERIT) is True:
                # _inherit with no dict parent — strip sentinel, use rest
                merged[key] = {k: v for k, v in ov.items() if k != _INHERIT}
            else:
                merged[key] = ov
        else:
            # key only in base
            merged[key] = base[key]
    return merged

ensure_dir(path)

Create a directory (and parents) if it doesn't exist.

Source code in src/terok_util/fs.py

def ensure_dir(path: Path) -> None:
    """Create a directory (and parents) if it doesn't exist."""
    path.mkdir(parents=True, exist_ok=True)

ensure_dir_writable(path, label)

Create path if needed and verify it is writable, or exit with an error.

Source code in src/terok_util/fs.py

def ensure_dir_writable(path: Path, label: str) -> None:
    """Create *path* if needed and verify it is writable, or exit with an error."""
    try:
        path.mkdir(parents=True, exist_ok=True)
    except Exception as e:
        raise SystemExit(f"{label} directory is not writable: {path} ({e})")
    if not path.is_dir():
        raise SystemExit(f"{label} path is not a directory: {path}")
    if not os.access(path, os.W_OK | os.X_OK):
        uid = os.getuid()
        gid = os.getgid()
        raise SystemExit(
            f"{label} directory is not writable: {path}\n"
            f"Fix permissions for the user running this process (uid={uid}, gid={gid}). "
            f"Example: sudo chown -R {uid}:{gid} {path}"
        )

write_sensitive_file(path, content)

Atomically create path with mode 0o600 and write content.

Returns True if the file was created, False if it already existed. Parent directories are created with mode 0o700.

Refuses to operate if path.parent is a symbolic link — chmod would otherwise follow the link target. Opens the file with O_NOFOLLOW so a planted symlink at the final path cannot redirect the write.

Source code in src/terok_util/fs.py

def write_sensitive_file(path: Path, content: str) -> bool:
    """Atomically create *path* with mode ``0o600`` and write *content*.

    Returns ``True`` if the file was created, ``False`` if it already existed.
    Parent directories are created with mode ``0o700``.

    Refuses to operate if *path.parent* is a symbolic link — chmod would
    otherwise follow the link target.  Opens the file with ``O_NOFOLLOW``
    so a planted symlink at the final path cannot redirect the write.
    """
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.parent.is_symlink():
        raise RuntimeError(f"Refusing to use symlinked directory for sensitive file: {path.parent}")
    os.chmod(path.parent, 0o700)
    nofollow = getattr(os, "O_NOFOLLOW", 0)
    try:
        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | nofollow, 0o600)
    except FileExistsError:
        return False
    try:
        # ``os.write`` may write fewer bytes than requested; loop until the
        # whole buffer lands.  On any failure, unlink the partially-written
        # file so the next call sees a clean state rather than a half-baked
        # secret.
        view = memoryview(content.encode())
        while view:
            view = view[os.write(fd, view) :]
    except BaseException:
        with contextlib.suppress(FileNotFoundError):
            os.unlink(path)
        raise
    finally:
        os.close(fd)
    return True

config_file_paths()

Ordered config.yml locations with scope labels (lowest → highest priority).

TEROK_CONFIG_FILE → single override (no layering). Otherwise: /etc/terok/config.yml (system) → ~/.config/terok/config.yml (user). Root processes see only the system path.

Public so consumers can render an "edit one of these to override X" hint to the operator (which file gets the highest priority, where on disk the operator would put the override, etc.).

Source code in src/terok_util/paths.py

def config_file_paths() -> list[tuple[str, Path]]:
    """Ordered config.yml locations with scope labels (lowest → highest priority).

    ``TEROK_CONFIG_FILE`` → single override (no layering).  Otherwise:
    ``/etc/terok/config.yml`` (system) → ``~/.config/terok/config.yml``
    (user).  Root processes see only the system path.

    Public so consumers can render an "edit one of these to override X"
    hint to the operator (which file gets the highest priority, where
    on disk the operator would put the override, etc.).
    """
    env = os.getenv(_TEROK_CONFIG_FILE_ENV)
    if env:
        return [("override", Path(env).expanduser())]
    result: list[tuple[str, Path]] = [
        ("system", Path("/etc") / _NAMESPACE / "config.yml"),
    ]
    if not _is_root():
        if _user_config_dir is not None:
            user_base = Path(_user_config_dir(_NAMESPACE))
        else:
            user_base = Path.home() / ".config" / _NAMESPACE
        result.append(("user", user_base / "config.yml"))
    return result

host_uid()

Return the current process's UID as the initial user namespace sees it.

Inside an unprivileged user namespace (rootless podman / crun hook, sandboxed CI runner, unshare -U), os.geteuid() returns the inner-userns UID — typically 0 even when the operator really ran the program as UID 1000. Network peers (D-Bus SO_PEERCRED, AUTH EXTERNAL) and kernel-level checks see the outer (host) UID via the userns uid_map translation, so a process that advertises its inner UID over the wire is rejected for a credential mismatch. This helper hands callers the outer UID those peers expect.

The mapping comes from /proc/self/uid_map. When it is unavailable (macOS, BSD, exotic chroot) or no row covers the effective UID, the bare geteuid() answer is returned — correct on systems without Linux user namespaces.

Source code in src/terok_util/paths.py

def host_uid() -> int:
    """Return the current process's UID as the initial user namespace sees it.

    Inside an unprivileged user namespace (rootless ``podman`` / ``crun``
    hook, sandboxed CI runner, ``unshare -U``), ``os.geteuid()`` returns
    the *inner*-userns UID — typically ``0`` even when the operator
    really ran the program as UID 1000.  Network peers (D-Bus
    ``SO_PEERCRED``, ``AUTH EXTERNAL``) and kernel-level checks see the
    *outer* (host) UID via the userns ``uid_map`` translation, so a
    process that advertises its inner UID over the wire is rejected for
    a credential mismatch.  This helper hands callers the outer UID those
    peers expect.

    The mapping comes from ``/proc/self/uid_map``.  When it is
    unavailable (macOS, BSD, exotic chroot) or no row covers the
    effective UID, the bare ``geteuid()`` answer is returned — correct on
    systems without Linux user namespaces.
    """
    try:
        euid = os.geteuid()
    except AttributeError:
        return 0 if getpass.getuser() == "root" else -1
    try:
        with open("/proc/self/uid_map", encoding="ascii") as fh:
            for line in fh:
                parts = line.split()
                if len(parts) != 3:
                    continue
                try:
                    inner_start, outer_start, length = (int(p) for p in parts)
                except ValueError:
                    continue
                if inner_start <= euid < inner_start + length:
                    return outer_start + (euid - inner_start)
    except OSError:
        pass
    return euid

namespace_config_dir(subdir='', *, env_var=None)

Resolve a config directory under the terok/ namespace.

Priority: env_var → /etc/terok/<subdir> (root) → platformdirs → ~/.config/terok/<subdir>. env_var is keyword-only.

Source code in src/terok_util/paths.py

def namespace_config_dir(subdir: str = "", *, env_var: str | None = None) -> Path:
    """Resolve a config directory under the ``terok/`` namespace.

    Priority: *env_var* → ``/etc/terok/<subdir>`` (root) → platformdirs
    → ``~/.config/terok/<subdir>``.  *env_var* is keyword-only.
    """
    if env_var:
        val = os.getenv(env_var)
        if val:
            return Path(val).expanduser()
    base: Path
    if _is_root():
        base = Path("/etc") / _NAMESPACE
    elif _user_config_dir is not None:
        base = Path(_user_config_dir(_NAMESPACE))
    else:
        base = Path.home() / ".config" / _NAMESPACE
    return _safe_subdir(base, subdir)

namespace_runtime_dir(subdir='', *, env_var=None)

Resolve a runtime directory under the terok/ namespace.

Priority: env_var → /run/terok/<subdir> (root) → $XDG_RUNTIME_DIR/terok/<subdir> → $XDG_STATE_HOME/terok/<subdir> → ~/.local/state/terok/<subdir>. env_var is keyword-only.

Source code in src/terok_util/paths.py

def namespace_runtime_dir(subdir: str = "", *, env_var: str | None = None) -> Path:
    """Resolve a runtime directory under the ``terok/`` namespace.

    Priority: *env_var* → ``/run/terok/<subdir>`` (root)
    → ``$XDG_RUNTIME_DIR/terok/<subdir>`` → ``$XDG_STATE_HOME/terok/<subdir>``
    → ``~/.local/state/terok/<subdir>``.  *env_var* is keyword-only.
    """
    if env_var:
        val = os.getenv(env_var)
        if val:
            return Path(val).expanduser()
    base: Path
    if _is_root():
        base = Path("/run") / _NAMESPACE
    else:
        xdg_runtime = os.getenv("XDG_RUNTIME_DIR")
        if xdg_runtime:
            base = Path(xdg_runtime) / _NAMESPACE
        else:
            xdg_state = os.getenv("XDG_STATE_HOME")
            base = (
                Path(xdg_state) / _NAMESPACE
                if xdg_state
                else Path.home() / ".local" / "state" / _NAMESPACE
            )
    return _safe_subdir(base, subdir)

namespace_state_dir(subdir='', *, env_var=None)

Resolve a state directory under the terok/ namespace.

Priority:

1. env_var (package-specific override, e.g. TEROK_SANDBOX_STATE_DIR)
2. TEROK_ROOT env var (namespace override)
3. config.yml → paths.root (Podman model — all packages honour it)
4. Platform default (/var/lib/terok/<subdir> for root, XDG data dir otherwise)

env_var is keyword-only so a positional second argument can never accidentally be reinterpreted as an override name.

Source code in src/terok_util/paths.py

def namespace_state_dir(subdir: str = "", *, env_var: str | None = None) -> Path:
    """Resolve a state directory under the ``terok/`` namespace.

    Priority:

    1. *env_var* (package-specific override, e.g. ``TEROK_SANDBOX_STATE_DIR``)
    2. ``TEROK_ROOT`` env var (namespace override)
    3. ``config.yml`` → ``paths.root`` (Podman model — all packages honour it)
    4. Platform default (``/var/lib/terok/<subdir>`` for root, XDG data
       dir otherwise)

    *env_var* is keyword-only so a positional second argument can never
    accidentally be reinterpreted as an override name.
    """
    if env_var:
        val = os.getenv(env_var)
        if val:
            return Path(val).expanduser()
    root = _namespace_root()
    base = root if root else _platform_state_base()
    return _safe_subdir(base, subdir)

read_config_section(section)

Read a top-level section from layered terok configs (cached, fail-silent).

Merges system and user config.yml files via ConfigStack — user values override system defaults at the leaf level. Lazy-imports config_stack so importing paths doesn't drag the YAML parser into a process that only needs the platform defaults.

Source code in src/terok_util/paths.py

def read_config_section(section: str) -> dict[str, str]:
    """Read a top-level section from layered terok configs (cached, fail-silent).

    Merges system and user ``config.yml`` files via
    [`ConfigStack`][terok_util.config_stack.ConfigStack] — user values
    override system defaults at the leaf level.  Lazy-imports
    ``config_stack`` so importing ``paths`` doesn't drag the YAML
    parser into a process that only needs the platform defaults.
    """
    if section in _config_section_cache:
        return _config_section_cache[section]

    result: dict[str, str] = {}
    try:
        from .config_stack import ConfigStack, load_yaml_scope

        stack = ConfigStack()
        for label, path in config_file_paths():
            stack.push(load_yaml_scope(label, path))
        merged = stack.resolve_section(section)
        result = {k: str(v) for k, v in merged.items() if v is not None}
    except Exception:  # noqa: BLE001 — fail-silent; bad config should not crash path resolution  # nosec B110 — best-effort probe; failure is non-fatal
        pass
    _config_section_cache[section] = result
    return result

read_config_top_level(key)

Read a top-level scalar / list / mapping from layered terok configs.

Counterpart to read_config_section for keys whose value isn't a dict — e.g. the ecosystem-wide experimental: true opt-in or a bare log_level: debug knob. Returns the merged value (user wins over system) or None when the key is absent or the config files can't be loaded. Cached for the lifetime of the process; reaches for the _config_top_level_cache private to flush in tests.

Source code in src/terok_util/paths.py

def read_config_top_level(key: str) -> object | None:
    """Read a top-level scalar / list / mapping from layered terok configs.

    Counterpart to
    [`read_config_section`][terok_util.paths.read_config_section] for
    keys whose value isn't a dict — e.g. the ecosystem-wide
    ``experimental: true`` opt-in or a bare ``log_level: debug`` knob.
    Returns the merged value (user wins over system) or ``None`` when
    the key is absent or the config files can't be loaded.  Cached for
    the lifetime of the process; reaches for the ``_config_top_level_cache``
    private to flush in tests.
    """
    if key in _config_top_level_cache:
        return _config_top_level_cache[key]

    result: object | None = None
    try:
        from .config_stack import ConfigStack, load_yaml_scope

        stack = ConfigStack()
        for label, path in config_file_paths():
            stack.push(load_yaml_scope(label, path))
        result = stack.resolve().get(key)
    except Exception:  # noqa: BLE001 — fail-silent; bad config should not crash field resolution  # nosec B110 — best-effort probe; failure is non-fatal
        pass
    _config_top_level_cache[key] = result
    return result

podman_userns_args()

Return user namespace args for rootless podman so UID 1000 maps correctly.

Maps the host user to container UID/GID 1000, the conventional non-root dev user in terok container images. Returns an empty list when running as root — root podman runs in the host user namespace and has no need to remap.

Source code in src/terok_util/podman.py

def podman_userns_args() -> list[str]:
    """Return user namespace args for rootless podman so UID 1000 maps correctly.

    Maps the host user to container UID/GID 1000, the conventional non-root
    ``dev`` user in terok container images.  Returns an empty list when
    running as root — root podman runs in the host user namespace and
    has no need to remap.
    """
    if os.geteuid() == 0:
        return []
    return ["--userns=keep-id:uid=1000,gid=1000"]

sanitize_tty(s)

Replace terminal control characters with safe representations.

Whitespace controls (newline, carriage return, tab) become spaces. All other characters in Unicode category C (control, format, surrogate, private use, unassigned) are rendered as \xNN hex escapes. Printable text passes through unchanged.

Source code in src/terok_util/security.py

def sanitize_tty(s: str) -> str:
    """Replace terminal control characters with safe representations.

    Whitespace controls (newline, carriage return, tab) become spaces.
    All other characters in Unicode category ``C`` (control, format,
    surrogate, private use, unassigned) are rendered as ``\\xNN`` hex
    escapes.  Printable text passes through unchanged.
    """
    out: list[str] = []
    for ch in s:
        if ch in ("\n", "\r", "\t"):
            out.append(" ")
        elif unicodedata.category(ch).startswith("C"):
            out.append(f"\\x{ord(ch):02x}")
        else:
            out.append(ch)
    return "".join(out)