terok_shield

terok_shield

terok-shield: nftables-based egress firewalling for Podman containers.

Public API facade. The Shield class coordinates collaborators:

• HookMode (hooks.mode) — per-container nft ruleset lifecycle
• DnsResolver (dns.resolver) — domain resolution and caching
• ProfileLoader (profiles) — allowlist profile composition
• RulesetBuilder (nft.rules) — nftables ruleset generation
• AuditLogger (audit) — per-container JSONL audit trail
• CommandRunner (run) — subprocess execution boundary

Core and support modules are imported lazily — from terok_shield import ShieldConfig does not pull in nft, dnsmasq, or subprocess helpers. Heavy imports are deferred until Shield is instantiated.

HOOK_ENTRYPOINT_NAME = 'terok-shield-hook' module-attribute

Canonical filename of the shield OCI hook entrypoint script.

Used (a) under ~/.local/share/containers/oci/hooks.d/ for user-wide installation and (b) under each per-container state_dir after Shield.pre_start() materialises it. Keeping both sites consuming the same constant means renaming the entrypoint is a single edit.

COMMANDS = CommandTree((_lazy('status', 'Show shield configuration overview', 'observe:STATUS'), _lazy('prepare', 'Prepare shield and print podman flags', 'launch:PREPARE'), _lazy('run', 'Launch a shielded container via podman', 'launch:RUN'), _lazy('resolve', 'Resolve DNS profiles and cache IPs', 'launch:RESOLVE'), _lazy('allow', 'Live-allow a domain or IP for a container', 'control:ALLOW'), _lazy('deny', 'Live-deny a domain or IP for a container', 'control:DENY'), _lazy('down', 'Switch container to the DOWN posture (accept + log)', 'control:DOWN'), _lazy('up', 'Restore deny-all mode for a container', 'control:UP'), _lazy('reset', 'Forget DNS-learned allow state (back to authored policy seeds)', 'control:RESET'), _lazy('quarantine', 'Total network blackout (drop all, log dropped traffic)', 'control:QUARANTINE'), _lazy('rules', 'Show current nft rules for a container', 'control:RULES'), _lazy('watch', 'Stream shield events — audit log, NFLOG packets, and DNS blocks on the dnsmasq tiers', 'stream:WATCH'), _lazy('simple-clearance', 'Terminal clearance fallback — prompts operator for each blocked connection (no D-Bus)', 'stream:SIMPLE_CLEARANCE'), _lazy('logs', 'Show audit log entries', 'observe:LOGS'), _lazy('profiles', 'List available shield profiles', 'observe:PROFILES'), _lazy('setup', 'Install global OCI hooks for restart persistence', 'launch:SETUP'), _lazy('check-environment', 'Check podman environment for compatibility issues', 'observe:CHECK_ENVIRONMENT'), _lazy('preview', 'Show ruleset that would be applied', 'control:PREVIEW'))) module-attribute

logger = logging.getLogger(__name__) module-attribute

__version__ = _meta_version('terok-shield') module-attribute

__all__ = ['ArgDef', 'COMMANDS', 'CommandDef', 'DnsTier', 'EnvironmentCheck', 'ExecError', 'HOOK_ENTRYPOINT_NAME', 'HooksInstaller', 'Shield', 'ShieldConfig', 'ShieldMode', 'ShieldRuntime', 'ShieldState', 'ensure_user_hooks_dir_configured', 'recorded_dns_tier', 'user_hooks_dir_configured'] module-attribute

DnsTier

Bases: Enum

How domain allowlists reach the nft allow sets.

dnsmasq with --nftset adds every answered address to the

allow sets before the reply reaches the workload. Follows IP rotation, covers subdomains, accepts wildcard entries, names blocked domains.

DNSMASQ_STATIC: a dnsmasq built without nftset support. The allow sets are resolved once at launch; the query log still names blocked domains. LOOKUP: no dnsmasq. The allow sets are resolved once at launch with dig or drill. GETENT: no dnsmasq and no lookup tool. The allow sets are resolved once at launch with getent hosts.

DNSMASQ_LIVE = 'dnsmasq-live' class-attribute instance-attribute

DNSMASQ_STATIC = 'dnsmasq-static' class-attribute instance-attribute

LOOKUP = 'lookup' class-attribute instance-attribute

GETENT = 'getent' class-attribute instance-attribute

live property

True when the allow sets follow DNS answers as they arrive.

runs_dnsmasq property

True when a per-container dnsmasq serves the container's DNS.

hint property

What a degraded tier lacks and what restores the live one; empty for the live tier.

parse(recorded) classmethod

The tier a recorded name means; None for a name that is not a tier.

A retired name reads as the tier it named, so a container recorded under it restarts instead of being recreated.

Source code in src/terok_shield/config.py

@classmethod
def parse(cls, recorded: str) -> DnsTier | None:
    """The tier a recorded name means; ``None`` for a name that is not a tier.

    A retired name reads as the tier it named, so a container recorded under
    it restarts instead of being recreated.
    """
    try:
        return cls(_RETIRED_TIER_NAMES.get(recorded, recorded))
    except ValueError:
        return None

ShieldConfig(state_dir, mode=ShieldMode.HOOK, default_profiles=('dev-standard',), loopback_ports=(), audit_enabled=True, profiles_dir=None, runtime=ShieldRuntime.DEFAULT, dns_cache_dir=None, dnsmasq_path=None) dataclass

Per-container shield configuration.

The library is a pure function of its inputs. Given a ShieldConfig with state_dir, it writes to that directory and nowhere else. No env-var reading, no config-file parsing.

state_dir instance-attribute

mode = ShieldMode.HOOK class-attribute instance-attribute

default_profiles = ('dev-standard',) class-attribute instance-attribute

loopback_ports = () class-attribute instance-attribute

audit_enabled = True class-attribute instance-attribute

profiles_dir = None class-attribute instance-attribute

runtime = ShieldRuntime.DEFAULT class-attribute instance-attribute

dns_cache_dir = None class-attribute instance-attribute

Resolved-allowlist cache shared across containers.

The one deliberate exception to the state_dir-only rule: many tasks with the same allowlist share one resolve. None selects dns_cache_dir under the shield state root. Only the tiers that resolve at launch use it.

dnsmasq_path = None class-attribute instance-attribute

The dnsmasq binary to run; None finds one on the current host PATH.

Set it for a dnsmasq built outside the distro package, for example one built with nftset support in the operator's home.

ShieldMode

Bases: Enum

Operating mode for the shield firewall.

Currently only HOOK is supported. Future modes (e.g. bridge) will add members here.

HOOK = 'hook' class-attribute instance-attribute

ShieldRuntime

Bases: Enum

Container runtime category — drives DNS-reachability assumptions.

crun / runc / youki. The container shares the netns,

so dnsmasq on 127.0.0.1 is reachable directly.

KRUN: libkrun microVM. The guest has its own loopback isolated from the netns, so dnsmasq must bind to a link-local address on netns lo that the guest can reach via passt.

DEFAULT = 'default' class-attribute instance-attribute

KRUN = 'krun' class-attribute instance-attribute

from_runtime_name(name) classmethod

Map a podman --runtime <name> string (or None) to the enum.

Centralises the wire-format vocabulary so callers don't repeat "krun" → KRUN mappings inline. Anything other than "krun" (including None and unknown runtime names) maps to DEFAULT — the loopback-shared-with-netns assumption holds for every runtime shield has been tested against besides krun.

Source code in src/terok_shield/config.py

@classmethod
def from_runtime_name(cls, name: str | None) -> ShieldRuntime:
    """Map a podman ``--runtime <name>`` string (or ``None``) to the enum.

    Centralises the wire-format vocabulary so callers don't repeat
    ``"krun" → KRUN`` mappings inline.  Anything other than
    ``"krun"`` (including ``None`` and unknown runtime names) maps
    to ``DEFAULT`` — the loopback-shared-with-netns assumption holds
    for every runtime shield has been tested against besides krun.
    """
    return cls.KRUN if name == "krun" else cls.DEFAULT

ShieldState

Bases: Enum

Per-container shield state, derived from the live nft ruleset.

QUARANTINE: Total network blackout — all traffic dropped, dropped traffic logged. UP: Normal enforcing mode (deny-all with allowlists). DOWN: Accept-by-default posture, private-range protection retained (RFC 1918 + RFC 4193). DISENGAGED: Accept-everything posture — no deny set, no private-range or hard-deny reject. OFFLINE: No ruleset found (container stopped or unshielded). ERROR: Ruleset present but unrecognised.

QUARANTINE = 'quarantine' class-attribute instance-attribute

UP = 'up' class-attribute instance-attribute

DOWN = 'down' class-attribute instance-attribute

DISENGAGED = 'disengaged' class-attribute instance-attribute

OFFLINE = 'offline' class-attribute instance-attribute

ERROR = 'error' class-attribute instance-attribute

HooksInstaller(target_dir=_default_target_dir()) dataclass

Persistent installation of terok-shield's OCI hook pair.

The createRuntime/poststop hook pair must persist across container restarts: podman ≥ 5.x drops per-container --hooks-dir on stop/start (containers/podman#17935), so global hooks are the only reliable activation path until that upstream regression is fixed.

Scripts, ballast, and JSON descriptors all land in target_dir (default: namespace_state_dir("shield") / "hooks"). containers.conf is patched to register that path so podman discovers the descriptors on the next container start.

Symmetric lifecycle: install writes, uninstall removes. Both are idempotent.

target_dir = field(default_factory=_default_target_dir) class-attribute instance-attribute

Directory the hook scripts, ballast, and JSON descriptors all live in.

check_setup(*, live=False)

Check Shield's receipt and hooks, optionally probing launch prerequisites.

Source code in src/terok_shield/hooks/install.py

def check_setup(self, *, live: bool = False) -> tuple[SetupCheck, ...]:
    """Check Shield's receipt and hooks, optionally probing launch prerequisites."""
    checks = [self._receipt().check(), *self._check_artifacts()]
    if live:
        checks.extend(self._check_tools())
    return tuple(checks)

install()

Install global standalone hooks after preflight; certify only verified work.

Source code in src/terok_shield/hooks/install.py

@setup_lock()
def install(self) -> None:
    """Install global standalone hooks after preflight; certify only verified work."""
    require_no_downgrade(self.check_setup())
    require_setup(self._check_tools())
    receipt = self._receipt()
    receipt.clear()
    install_reader_resource()
    self.target_dir.mkdir(parents=True, exist_ok=True)
    _write_role_files(self.target_dir)
    ensure_user_hooks_dir_configured(self.target_dir)
    require_setup(self._check_artifacts())
    receipt.write()

uninstall()

Remove every hook file install would write.

Idempotent — missing files are tolerated. containers.conf is left untouched: other terok packages may still register their own hooks_dir entries the operator wants to keep.

Source code in src/terok_shield/hooks/install.py

@setup_lock()
def uninstall(self) -> None:
    """Remove every hook file [`install`][terok_shield.hooks.install.HooksInstaller.install] would write.

    Idempotent — missing files are tolerated.  ``containers.conf``
    is left untouched: other terok packages may still register
    their own ``hooks_dir`` entries the operator wants to keep.
    """
    self._receipt().clear()
    for name in (*_SCRIPT_FILES, *_DESCRIPTOR_FILES):
        (self.target_dir / name).unlink(missing_ok=True)

is_installed()

True when target_dir carries the canonical createRuntime hook JSON.

Use check_setup for receipt, interpreter, and artifact validation.

Source code in src/terok_shield/hooks/install.py

def is_installed(self) -> bool:
    """True when ``target_dir`` carries the canonical createRuntime hook JSON.

    Use ``check_setup`` for receipt, interpreter, and artifact validation.
    """
    return (self.target_dir / _nft_hook_json("createRuntime")).is_file()

ExecError(cmd, rc, stderr)

Bases: Exception

Raised when a subprocess fails.

Store command details and format the error message.

Source code in src/terok_shield/run.py

def __init__(self, cmd: list[str], rc: int, stderr: str) -> None:
    """Store command details and format the error message."""
    self.cmd = cmd
    self.rc = rc
    self.stderr = stderr
    super().__init__(f"{cmd!r} failed (rc={rc}): {stderr.strip()}")

cmd = cmd instance-attribute

rc = rc instance-attribute

stderr = stderr instance-attribute

EnvironmentCheck(dns_tier='', ok=True, podman_version=(0,), hooks='not-installed', health='ok', issues=list(), needs_setup=False, setup_hint='') dataclass

Result of Shield.check_environment.

Machine-readable fields for programmatic consumers (terok TUI, scripts). Human-readable issues and setup_hint for CLI display.

Attributes:

Name	Type	Description
ok	bool	True if no issues found.
podman_version	tuple[int, ...]	Detected podman version tuple.
hooks	str	Hook installation type (per-container, global, not-installed).
health	str	Environment health (ok, setup-needed, stale-hooks).
dns_tier	str	Active DNS resolution tier, a DnsTier value.
issues	list[str]	List of human-readable issue descriptions.
needs_setup	bool	True if one-time setup is required.
setup_hint	str	Setup instructions (empty if not needed).

dns_tier = '' class-attribute instance-attribute

ok = True class-attribute instance-attribute

podman_version = (0,) class-attribute instance-attribute

hooks = 'not-installed' class-attribute instance-attribute

health = 'ok' class-attribute instance-attribute

issues = field(default_factory=list) class-attribute instance-attribute

needs_setup = False class-attribute instance-attribute

setup_hint = '' class-attribute instance-attribute

Shield(config, *, runner=None, audit=None, dns=None, profiles=None, ruleset=None, hub_events=None)

Public API facade — coordinates collaborators per container.

Delegates to HookMode for netns/nft operations, DnsResolver for name resolution, ProfileLoader for allowlists, RulesetBuilder for ruleset generation, and AuditLogger for the audit trail. All collaborators are injectable for testing.

Create the shield facade.

Parameters:

Name	Type	Description	Default
config	ShieldConfig	Shield configuration (must include state_dir).	required
runner	CommandRunner | None	Command runner (default: SubprocessRunner).	None
audit	AuditLogger | None	Audit logger (default: from config.state_dir).	None
dns	DnsResolver | None	DNS resolver (default: from runner).	None
profiles	ProfileLoader | None	Profile loader (default: from config.profiles_dir).	None
ruleset	RulesetBuilder | None	Ruleset builder (default: from config loopback_ports).	None
hub_events	HubEventEmitter | None	Best-effort emitter for shield_up / shield_down events bound for the terok-clearance hub (default: a fresh HubEventEmitter). The emitter routes each event to the supervisor's per-container socket using the container_id supplied on every up / down call. Pass a no-op stub in tests that should not touch the socket.	None

Source code in src/terok_shield/__init__.py

def __init__(
    self,
    config: ShieldConfig,
    *,
    runner: "CommandRunner | None" = None,
    audit: "AuditLogger | None" = None,
    dns: "DnsResolver | None" = None,
    profiles: "ProfileLoader | None" = None,
    ruleset: "RulesetBuilder | None" = None,
    hub_events: "HubEventEmitter | None" = None,
) -> None:
    """Create the shield facade.

    Args:
        config: Shield configuration (must include state_dir).
        runner: Command runner (default: ``SubprocessRunner``).
        audit: Audit logger (default: from config.state_dir).
        dns: DNS resolver (default: from runner).
        profiles: Profile loader (default: from config.profiles_dir).
        ruleset: Ruleset builder (default: from config loopback_ports).
        hub_events: Best-effort emitter for ``shield_up`` / ``shield_down``
            events bound for the terok-clearance hub (default: a fresh
            [`HubEventEmitter`][terok_shield._hub_events.HubEventEmitter]).  The
            emitter routes each event to the supervisor's
            per-container socket using the ``container_id`` supplied
            on every [`up`][terok_shield.Shield.up] /
            [`down`][terok_shield.Shield.down] call.  Pass a no-op
            stub in tests that should not touch the socket.
    """
    from ._hub_events import HubEventEmitter
    from .audit import AuditLogger
    from .dns.resolver import DnsResolver
    from .nft.rules import RulesetBuilder
    from .profiles import ProfileLoader
    from .run import SubprocessRunner

    self.config = config
    self.runner = runner or SubprocessRunner()
    self.audit = audit or AuditLogger(
        audit_path=StateBundle(config.state_dir).audit,
        enabled=config.audit_enabled,
    )
    self.dns = dns or DnsResolver(runner=self.runner, host_cache_dir=config.dns_cache_dir)
    self.profiles = profiles or ProfileLoader(
        user_dir=config.profiles_dir or Path("/nonexistent"),
    )
    self.ruleset = ruleset or RulesetBuilder(loopback_ports=config.loopback_ports)
    self.hub_events = hub_events or HubEventEmitter()
    self._mode = self._create_mode(config.mode)

config = config instance-attribute

runner = runner or SubprocessRunner() instance-attribute

audit = audit or AuditLogger(audit_path=StateBundle(config.state_dir).audit, enabled=config.audit_enabled) instance-attribute

dns = dns or DnsResolver(runner=self.runner, host_cache_dir=config.dns_cache_dir) instance-attribute

profiles = profiles or ProfileLoader(user_dir=config.profiles_dir or Path('/nonexistent')) instance-attribute

ruleset = ruleset or RulesetBuilder(loopback_ports=config.loopback_ports) instance-attribute

hub_events = hub_events or HubEventEmitter() instance-attribute

check_environment()

Check the podman environment for compatibility issues.

Proactive check for API consumers (e.g. terok). Returns an EnvironmentCheck with detected issues and setup hints. Does not raise — the caller decides how to handle issues.

Source code in src/terok_shield/__init__.py

def check_environment(self) -> EnvironmentCheck:
    """Check the podman environment for compatibility issues.

    Proactive check for API consumers (e.g. terok).  Returns an
    [`EnvironmentCheck`][terok_shield.EnvironmentCheck] with detected issues and setup hints.
    Does not raise — the caller decides how to handle issues.
    """
    from terok_util import SetupStatus

    from .dns import apparmor, dnsmasq
    from .hooks.install import HooksInstaller
    from .podman_info import parse_podman_info
    from .run import ShieldNeedsSetup

    output = self.runner.run(["podman", "info", "-f", "json"], check=False)
    info = parse_podman_info(output)
    issues: list[str] = []
    needs_setup = False
    setup_hint = ""
    health = "ok"

    try:
        dnsmasq_bin = dnsmasq.locate(self.config.dnsmasq_path, self.runner)
    except ShieldNeedsSetup as exc:
        issues.append(str(exc))
        dnsmasq_bin = ""
    tier, apparmor_blocked = apparmor.detect_dns_tier_under_apparmor(
        self.runner, self.config.state_dir, dnsmasq_bin
    )
    dns_tier = tier.value
    if apparmor_blocked:
        issues.append(
            f"AppArmor confines {dnsmasq_bin} from the shield state directory. "
            "Install the terok AppArmor profile addendum (docs/apparmor.md)."
        )
    if not tier.live:
        issues.append(f"DNS tier {tier.value}: {tier.hint}")

    checks = HooksInstaller().check_setup()
    failed = [check for check in checks if check.status != SetupStatus.READY]
    hooks = "global" if not failed else "not-installed"
    if failed:
        health = "setup-needed"
        needs_setup = True
        setup_hint = "Run 'terok-shield setup' to refresh global hooks."
        issues.extend(check.diagnostic for check in failed)

    return EnvironmentCheck(
        ok=not issues,
        podman_version=info.version,
        hooks=hooks,
        health=health,
        dns_tier=dns_tier,
        issues=issues,
        needs_setup=needs_setup,
        setup_hint=setup_hint,
    )

status()

Return current shield status information.

Source code in src/terok_shield/__init__.py

def status(self) -> dict:
    """Return current shield status information."""
    return {
        "mode": self.config.mode.value,
        "profiles": self.profiles.list_profiles(),
        "audit_enabled": self.config.audit_enabled,
    }

pre_start(container, profiles=None, *, security_deny=(), provider_allow=(), project_allow=(), override=())

Prepare shield for container start. Returns extra podman args.

The four tier arguments are the orchestrator-generated policy tiers, which shield writes into the bundle so callers pass data and never touch the layout: security_deny → t20 (vault hosts denied direct), provider_allow → t30 (provider egress), project_allow → t40 (git remote + custom, merged with the composed profiles), override → t10 (break-glass allow above the deny; a CIDR is accepted but logged as a warning).

Source code in src/terok_shield/__init__.py

def pre_start(
    self,
    container: str,
    profiles: list[str] | None = None,
    *,
    security_deny: Sequence[str] = (),
    provider_allow: Sequence[str] = (),
    project_allow: Sequence[str] = (),
    override: Sequence[str] = (),
) -> list[str]:
    """Prepare shield for container start.  Returns extra podman args.

    The four tier arguments are the orchestrator-generated policy tiers,
    which shield writes into the bundle so callers pass data and never touch
    the layout: *security_deny* → t20 (vault hosts denied direct),
    *provider_allow* → t30 (provider egress), *project_allow* → t40 (git
    remote + custom, merged with the composed profiles), *override* → t10
    (break-glass allow above the deny; a CIDR is accepted but logged as a warning).
    """
    if profiles is None:
        profiles = list(self.config.default_profiles)
    result = self._mode.pre_start(
        container,
        profiles,
        security_deny=security_deny,
        provider_allow=provider_allow,
        project_allow=project_allow,
        override=override,
    )
    self.audit.log_event(container, "setup", detail=f"profiles={','.join(profiles)}")
    return result

refresh(container, profiles=None, *, security_deny=(), provider_allow=(), project_allow=(), override=())

Recompute an existing container's policy bundle before a plain restart.

Same tier arguments as pre_start, but for a container that already exists: rewrites the tiers and static-resolution caches and regenerates the pre-applied artifacts (ruleset.nft, dnsmasq config), so the next podman start enforces current policy instead of the bundle frozen at creation. Returns nothing — the container keeps its launch-time podman args.

Source code in src/terok_shield/__init__.py

def refresh(
    self,
    container: str,
    profiles: list[str] | None = None,
    *,
    security_deny: Sequence[str] = (),
    provider_allow: Sequence[str] = (),
    project_allow: Sequence[str] = (),
    override: Sequence[str] = (),
) -> None:
    """Recompute an existing container's policy bundle before a plain restart.

    Same tier arguments as [`pre_start`][terok_shield.Shield.pre_start],
    but for a container that already exists: rewrites the tiers and
    static-resolution caches and regenerates the pre-applied artifacts
    (``ruleset.nft``, dnsmasq config), so the next ``podman start``
    enforces *current* policy instead of the bundle frozen at creation.
    Returns nothing — the container keeps its launch-time podman args.
    """
    if profiles is None:
        profiles = list(self.config.default_profiles)
    self._mode.refresh(
        container,
        profiles,
        security_deny=security_deny,
        provider_allow=provider_allow,
        project_allow=project_allow,
        override=override,
    )
    self.audit.log_event(container, "refresh", detail=f"profiles={','.join(profiles)}")

allow(container, target)

Live-allow a domain or IP for a running container.

Source code in src/terok_shield/__init__.py

def allow(self, container: str, target: str) -> list[str]:
    """Live-allow a domain or IP for a running container."""
    from .run import ExecError

    self._refuse_static_wildcard(target)
    is_domain = not _is_ip(target)
    ips = [target] if not is_domain else self.dns.resolve_domains([target])
    allowed: list[str] = []
    for ip in ips:
        try:
            self._mode.allow_ip(container, ip)
        except (ExecError, OSError) as exc:
            logger.warning("allow_ip failed for %s on %s: %s", ip, container, exc)
            continue
        allowed.append(ip)
        self.audit.log_event(container, "allowed", dest=ip, detail=f"target={target}")
    # Update dnsmasq config for domain targets (so future IP rotations are captured)
    if is_domain and allowed:
        self._mode.allow_domain(container, target)
    return allowed

deny(container, target)

Live-deny a domain or IP for a running container.

Source code in src/terok_shield/__init__.py

def deny(self, container: str, target: str) -> list[str]:
    """Live-deny a domain or IP for a running container."""
    from .run import ExecError

    self._refuse_static_wildcard(target)
    is_domain = not _is_ip(target)
    ips = [target] if not is_domain else self.dns.resolve_domains([target])
    denied: list[str] = []
    for ip in ips:
        try:
            self._mode.deny_ip(container, ip)
        except (ExecError, OSError) as exc:
            logger.warning("deny_ip failed for %s on %s: %s", ip, container, exc)
            continue
        denied.append(ip)
        self.audit.log_event(container, "denied", dest=ip, detail=f"target={target}")
    # Remove domain from dnsmasq config (stops future auto-population)
    if is_domain and denied:
        self._mode.deny_domain(container, target)
    return denied

rules(container)

Return current nft rules for a container.

Source code in src/terok_shield/__init__.py

def rules(self, container: str) -> str:
    """Return current nft rules for a container."""
    return self._mode.list_rules(container)

down(container, container_id, *, disengaged=False)

Switch a running container to the DOWN posture.

container is the operator-facing podman name (audit log key); container_id is the full podman UUID — the routing key for the per-container hub socket the supervisor listens on. The caller knows both at every emit site, so neither carries a default.

With disengaged, the container takes the DISENGAGED posture instead: nothing is enforced — no deny set, no private-range or hard-deny reject — and every new connection is only logged.

Source code in src/terok_shield/__init__.py

def down(self, container: str, container_id: str, *, disengaged: bool = False) -> None:
    """Switch a running container to the DOWN posture.

    *container* is the operator-facing podman name (audit log key);
    *container_id* is the full podman UUID — the routing key for
    the per-container hub socket the supervisor listens on.  The
    caller knows both at every emit site, so neither carries a
    default.

    With *disengaged*, the container takes the DISENGAGED posture
    instead: nothing is enforced — no deny set, no private-range or
    hard-deny reject — and every new connection is only logged.
    """
    self._mode.shield_down(container, disengaged=disengaged)
    self.audit.log_event(
        container,
        "shield_down",
        detail="disengaged=True" if disengaged else None,
    )
    self.hub_events.shield_down(
        container,
        container_id,
        disengaged=disengaged,
        dossier=self._read_dossier(),
    )

quarantine(container)

Total network blackout — drop all traffic, log dropped traffic.

Source code in src/terok_shield/__init__.py

def quarantine(self, container: str) -> None:
    """Total network blackout — drop all traffic, log dropped traffic."""
    self._mode.shield_quarantine(container)
    self.audit.log_event(container, "shield_quarantine")

up(container, container_id)

Restore normal deny-all mode for a running container.

container / container_id — see down.

Source code in src/terok_shield/__init__.py

def up(self, container: str, container_id: str) -> None:
    """Restore normal deny-all mode for a running container.

    *container* / *container_id* — see
    [`down`][terok_shield.Shield.down].
    """
    self._mode.shield_up(container)
    self.audit.log_event(container, "shield_up")
    self.hub_events.shield_up(container, container_id, dossier=self._read_dossier())

reset(container)

Forget DNS-learned allow state, keeping the authored policy seeds.

The live tier accumulates every IP the workload legitimately resolved; reset returns the allow sets to their just-launched contents (policy literals only) without touching the deny tier or the operator's runtime overlay.

Source code in src/terok_shield/__init__.py

def reset(self, container: str) -> None:
    """Forget DNS-learned allow state, keeping the authored policy seeds.

    The live tier accumulates every IP the workload legitimately
    resolved; ``reset`` returns the allow sets to their just-launched
    contents (policy literals only) without touching the deny tier or
    the operator's runtime overlay.
    """
    self._mode.shield_reset(container)
    self.audit.log_event(container, "shield_reset")

state(container)

Query the live nft ruleset to determine a container's shield state.

Source code in src/terok_shield/__init__.py

def state(self, container: str) -> ShieldState:
    """Query the live nft ruleset to determine a container's shield state."""
    return self._mode.shield_state(container)

preview(*, down=False, disengaged=False)

Generate the ruleset that would be applied to a container.

Source code in src/terok_shield/__init__.py

def preview(self, *, down: bool = False, disengaged: bool = False) -> str:
    """Generate the ruleset that would be applied to a container."""
    return self._mode.preview(down=down, disengaged=disengaged)

resolve(profiles=None, *, force=False)

Resolve DNS profiles and cache the results.

Source code in src/terok_shield/__init__.py

def resolve(
    self,
    profiles: list[str] | None = None,
    *,
    force: bool = False,
) -> list[str]:
    """Resolve DNS profiles and cache the results."""
    if profiles is None:
        profiles = list(self.config.default_profiles)
    entries = self.profiles.compose_profiles(profiles)
    if not entries:
        return []
    bundle = StateBundle(self.config.state_dir)
    bundle.ensure_dirs()
    bundle.write_tier("project_allow", "".join(f"+{e}\n" for e in entries))
    max_age = 0 if force else 3600
    return self.dns.resolve_and_cache(
        bundle.read_effective().allow_targets(),
        bundle.resolved_cache,
        max_age=max_age,
        source_mtime=bundle.policy_mtime(),
    )

profiles_list()

List available profile names.

Source code in src/terok_shield/__init__.py

def profiles_list(self) -> list[str]:
    """List available profile names."""
    return self.profiles.list_profiles()

tail_log(n=50)

Yield the last n audit events.

Source code in src/terok_shield/__init__.py

def tail_log(self, n: int = 50) -> Iterator[dict]:
    """Yield the last *n* audit events."""
    return self.audit.tail_log(n)

compose_profiles(names)

Load and merge multiple profiles.

Source code in src/terok_shield/__init__.py

def compose_profiles(self, names: list[str]) -> list[str]:
    """Load and merge multiple profiles."""
    return self.profiles.compose_profiles(names)

recorded_dns_tier(state_dir)

The DNS tier a shielded container launched with, read from state_dir.

Thin public wrapper over StateBundle.read_dns_tier so callers that only want the tier need not know the bundle layout.

Source code in src/terok_shield/state.py

def recorded_dns_tier(state_dir: Path) -> DnsTier | None:
    """The DNS tier a shielded container launched with, read from *state_dir*.

    Thin public wrapper over
    [`StateBundle.read_dns_tier`][terok_shield.state.StateBundle.read_dns_tier]
    so callers that only want the tier need not know the bundle layout.
    """
    return StateBundle(state_dir).read_dns_tier()

ensure_user_hooks_dir_configured(hooks_dir=None)

Ensure ~/.config/containers/containers.conf lists hooks_dir.

The canonical SSOT for the rootless OCI hooks directory across every terok package: shield calls it at setup time; other installers (e.g. terok-sandbox's per-container supervisor) call it before dropping their own descriptors so they don't have to re-implement the containers.conf patcher. Idempotent.

hooks_dir defaults to namespace_state_dir("shield") / "hooks" — shield's canonical install location under paths.root.

Creates the conf file if absent. Inserts hooks_dir into the existing [engine] section or appends a new section if none exists. Skips silently when hooks_dir is already listed. When a different hooks_dir is configured, appends ours to the list rather than failing — the operator owns containers.conf and may have intentionally pinned other locations.

Pure line-based editing — comments and formatting are preserved.

Source code in src/terok_shield/hooks/install.py

@setup_lock()
def ensure_user_hooks_dir_configured(hooks_dir: Path | None = None) -> None:
    """Ensure ``~/.config/containers/containers.conf`` lists *hooks_dir*.

    The canonical SSOT for the rootless OCI hooks directory across
    every terok package: shield calls it at ``setup`` time; other
    installers (e.g. terok-sandbox's per-container supervisor) call
    it before dropping their own descriptors so they don't have to
    re-implement the containers.conf patcher.  Idempotent.

    *hooks_dir* defaults to ``namespace_state_dir("shield") / "hooks"``
    — shield's canonical install location under ``paths.root``.

    Creates the conf file if absent.  Inserts ``hooks_dir`` into the
    existing ``[engine]`` section or appends a new section if none
    exists.  Skips silently when *hooks_dir* is already listed.  When
    a different ``hooks_dir`` is configured, appends ours to the list
    rather than failing — the operator owns containers.conf and may
    have intentionally pinned other locations.

    Pure line-based editing — comments and formatting are preserved.
    """
    if hooks_dir is None:
        hooks_dir = _default_target_dir()
    conf_path = _user_containers_conf()
    hooks_str = str(hooks_dir)
    hooks_line = f'hooks_dir = ["{hooks_str}"]'

    if not conf_path.is_file():
        conf_path.parent.mkdir(parents=True, exist_ok=True)
        conf_path.write_text(f"[engine]\n{hooks_line}\n")
        return

    existing = _parse_hooks_dir_from_conf(conf_path)
    if not existing:
        _insert_hooks_line(conf_path, hooks_line)
        return

    if hooks_str in existing or str(hooks_dir.expanduser()) in existing:
        return  # already configured
    _append_to_hooks_dir(conf_path, hooks_str)

user_hooks_dir_configured(hooks_dir)

Whether the user's containers.conf registers this package-owned hook directory.

Source code in src/terok_shield/hooks/install.py

def user_hooks_dir_configured(hooks_dir: Path) -> bool:
    """Whether the user's containers.conf registers this package-owned hook directory."""
    return str(hooks_dir.expanduser()) in {
        str(Path(entry).expanduser())
        for entry in _parse_hooks_dir_from_conf(_user_containers_conf())
    }

__getattr__(name)

Lazy import for re-exported core/support layer names.

Source code in src/terok_shield/__init__.py

def __getattr__(name: str) -> object:
    """Lazy import for re-exported core/support layer names."""
    if name in _LAZY_IMPORTS:
        mod_path, attr = _LAZY_IMPORTS[name]
        mod = _importlib.import_module(mod_path)
        value = getattr(mod, attr)
        globals()[name] = value  # cache for subsequent access
        return value
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")