mode

mode

Hook mode: OCI hooks + per-container netns.

Uses OCI hooks to apply per-container nftables rules inside each container's network namespace. No root required — only podman and nft.

Orchestrates collaborators per lifecycle phase:

• RulesetBuilder (nft.rules) — generates and verifies nft rulesets
• DnsResolver (dns.resolver) — pre-start domain resolution
• ProfileLoader (profiles) — allowlist profile composition
• AuditLogger (audit) — event logging
• CommandRunner (run) — subprocess execution (nft, nsenter)
• dnsmasq (dns.dnsmasq) — runtime DNS with nftset auto-population
• hook_install (hooks.install) — OCI hook file generation
• state (state) — per-container state bundle I/O

logger = logging.getLogger(__name__) module-attribute

HookMode(*, config, runner, audit, dns, profiles, ruleset)

Hook-mode shield backend (Strategy, implements ShieldModeBackend).

Coordinates the full lifecycle of OCI-hook-based container firewalling. Delegates to RulesetBuilder for nft generation, DnsResolver for name resolution, ProfileLoader for allowlists, dnsmasq for runtime DNS, and state for per-container persistence.

Create a hook mode backend with all collaborators.

Parameters:

Name	Type	Description	Default
config	ShieldConfig	Shield configuration (provides state_dir).	required
runner	CommandRunner	Command runner for subprocess calls.	required
audit	AuditLogger	Audit logger for event logging.	required
dns	DnsResolver	DNS resolver for domain resolution and caching.	required
profiles	ProfileLoader	Profile loader for allowlist profiles.	required
ruleset	RulesetBuilder	Ruleset builder for nft generation and verification.	required

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

def __init__(
    self,
    *,
    config: ShieldConfig,
    runner: "CommandRunner",
    audit: "AuditLogger",
    dns: "DnsResolver",
    profiles: "ProfileLoader",
    ruleset: RulesetBuilder,
) -> None:
    """Create a hook mode backend with all collaborators.

    Args:
        config: Shield configuration (provides state_dir).
        runner: Command runner for subprocess calls.
        audit: Audit logger for event logging.
        dns: DNS resolver for domain resolution and caching.
        profiles: Profile loader for allowlist profiles.
        ruleset: Ruleset builder for nft generation and verification.
    """
    self._config = config
    self._runner = runner
    self._audit = audit
    self._dns = dns
    self._profiles = profiles
    self._ruleset = ruleset
    self._podman_info: PodmanInfo | None = None
    self._gateways: tuple[str, str] | None = None

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

Prepare for container start in hook mode.

Verifies setup, composes profiles, resolves DNS, writes allowlist, detects DNS tier, sets annotations, and returns the podman CLI arguments needed for shield protection.

Parameters:

Name	Type	Description	Default
security_deny	Sequence[str]	Hosts/IPs an upstream layer (executor's roster projection, carried by sandbox) generates for the t20 security-deny tier — vault hosts denied direct egress.	()
provider_allow	Sequence[str]	Hosts/IPs generated for the t30 provider-allow tier — agent/provider egress endpoints.	()
project_allow	Sequence[str]	Hosts/IPs authored by the orchestrator for the t40 project-allow tier (git remote, custom domains) — merged with the composed profiles.	()
override	Sequence[str]	Hosts/IPs/CIDRs authored for the t10 break-glass override tier, which sits above the security-deny; statically resolved and seeded into a separate nft set. A CIDR opens a whole subnet above the deny — accepted, but logged as a warning and an override_range audit event. Shield owns writing every tier, so callers pass data, never touch the bundle.	()

Raises:

Type	Description
SetupRequiredError	When global hooks are not installed or need refreshing.

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

def pre_start(
    self,
    container: str,
    profiles: list[str],
    *,
    security_deny: Sequence[str] = (),
    provider_allow: Sequence[str] = (),
    project_allow: Sequence[str] = (),
    override: Sequence[str] = (),
) -> list[str]:
    """Prepare for container start in hook mode.

    Verifies setup, composes profiles, resolves DNS, writes
    allowlist, detects DNS tier, sets annotations, and returns
    the podman CLI arguments needed for shield protection.

    Args:
        security_deny: Hosts/IPs an upstream layer (executor's roster
            projection, carried by sandbox) generates for the t20
            security-deny tier — vault hosts denied direct egress.
        provider_allow: Hosts/IPs generated for the t30 provider-allow
            tier — agent/provider egress endpoints.
        project_allow: Hosts/IPs authored by the orchestrator for the t40
            project-allow tier (git remote, custom domains) — merged with
            the composed profiles.
        override: Hosts/IPs/CIDRs authored for the t10 break-glass override
            tier, which sits *above* the security-deny; statically resolved
            and seeded into a separate nft set.  A CIDR opens a whole
            subnet above the deny — accepted, but logged as a warning and
            an ``override_range`` audit event.  Shield owns writing every
            tier, so callers pass data, never touch the bundle.

    Raises:
        terok_util.SetupRequiredError: When global hooks are not installed
            or need refreshing.
    """
    sd = self._config.state_dir.resolve()
    require_setup(HooksInstaller().check_setup(live=True))
    info = self._get_podman_info()
    StateBundle(sd).ensure_dirs()

    # Detect DNS tier, upstream DNS, and gateway addresses
    dnsmasq_bin = dnsmasq.locate(self._config.dnsmasq_path, self._runner)
    tier = self._detect_dns_tier(container, sd, dnsmasq_bin)
    mode = info.network_mode or "pasta"
    upstream_dns = _upstream_dns_for_mode(mode)
    gw_v4, gw_v6 = self._gateways = _gateways_for_mode(mode)

    # Persist the launch-detected facts first: they are what ``refresh``
    # reuses instead of re-detecting, and ``_write_ruleset`` reads the
    # ports back out of the bundle (SSOT) rather than off the config, so
    # later up/down rebuilds share one source.
    bundle = StateBundle(sd)
    bundle.upstream_dns.write_text(f"{upstream_dns}\n")
    bundle.dns_tier.write_text(f"{tier.value}\n")
    bundle.dnsmasq_command.write_text(f"{dnsmasq_bin}\n")
    bundle.network_mode.write_text(f"{mode}\n")
    bundle.loopback_ports.write_text("".join(f"{p}\n" for p in self._config.loopback_ports))
    self._author_policy(
        container,
        sd,
        profiles,
        tier,
        upstream_dns,
        (gw_v4, gw_v6),
        security_deny=security_deny,
        provider_allow=provider_allow,
        project_allow=project_allow,
        override=override,
    )

    # Build podman args
    args = self._build_network_args(mode)

    # WORKAROUND(pasta-dns-bind): bind-mount shield's own resolv.conf over
    # the container's on every tier instead of using podman --dns. --dns
    # makes pasta bind host port 53, which fails for a rootless container.
    # Podman's default resolv.conf lists the host's own nameservers; on an
    # AppArmor host one of those is a blocked LAN router, so the container
    # resolves nothing (#1246). Drop this when rootless podman/pasta can
    # set the container's resolvers without binding host port 53.
    args += ["--volume", f"{StateBundle(sd).resolv_conf}:/etc/resolv.conf:ro,Z"]

    # Annotations: profiles, name, state_dir, version, dns.  loopback_ports
    # lives in the state bundle (per-container, written above), not as an
    # annotation — annotations are write-only on shield's side.
    args += [
        "--annotation",
        f"{ANNOTATION_KEY}={ANNOTATION_LIST_SEP.join(profiles)}",
        "--annotation",
        f"{ANNOTATION_NAME_KEY}={container}",
        "--annotation",
        f"{ANNOTATION_STATE_DIR_KEY}={sd}",
        "--annotation",
        f"{ANNOTATION_VERSION_KEY}={state.BUNDLE_VERSION}",
        "--annotation",
        f"{ANNOTATION_AUDIT_ENABLED_KEY}={str(self._config.audit_enabled).lower()}",
        "--annotation",
        f"{ANNOTATION_UPSTREAM_DNS_KEY}={upstream_dns}",
        "--annotation",
        f"{ANNOTATION_DNS_TIER_KEY}={tier.value}",
    ]

    self._audit.log_event(container, "setup", detail="using setup-installed global hooks")

    args += [
        "--cap-drop",
        "NET_ADMIN",
        "--cap-drop",
        "NET_RAW",
    ]
    return args

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

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

The policy-authoring half of pre_start without the launch half: rewrites every tier from the caller's current data, refreshes the static-resolution caches, and regenerates ruleset.nft + the dnsmasq config — so the OCI hook applies current policy at the next podman start instead of replaying the bundle frozen at creation. Reuses every launch-detected fact the bundle persisted — DNS tier, upstream DNS, network mode, loopback ports — rather than re-detecting: the container's mounts and annotations were built for those, a fresh detection could disagree with them, and a restart stays free of podman info.

Raises:

Type	Description
RuntimeError	When the bundle carries no persisted DNS tier / upstream DNS / network mode (pre_start never ran for this state dir).

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

def refresh(
    self,
    container: str,
    profiles: list[str],
    *,
    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.

    The policy-authoring half of
    [`pre_start`][terok_shield.hooks.mode.HookMode.pre_start] without the
    launch half: rewrites every tier from the caller's current data,
    refreshes the static-resolution caches, and regenerates
    ``ruleset.nft`` + the dnsmasq config — so the OCI hook applies
    *current* policy at the next ``podman start`` instead of replaying
    the bundle frozen at creation.  Reuses every launch-detected fact the
    bundle persisted — DNS tier, upstream DNS, network mode, loopback
    ports — rather than re-detecting: the container's mounts and
    annotations were built for those, a fresh detection could disagree
    with them, and a restart stays free of ``podman info``.

    Raises:
        RuntimeError: When the bundle carries no persisted DNS tier /
            upstream DNS / network mode (``pre_start`` never ran for this
            state dir).
    """
    sd = self._config.state_dir.resolve()
    bundle = StateBundle(sd)
    tier = bundle.read_dns_tier()
    mode = bundle.network_mode.read_text().strip() if bundle.network_mode.is_file() else ""
    upstream_dns = self._read_upstream_dns()
    if tier is None or not upstream_dns or not mode:
        raise RuntimeError(
            "shield bundle has no persisted DNS tier / upstream DNS / network mode — "
            "pre_start never completed for this container; re-create the task"
        )
    self._gateways = _gateways_for_mode(mode)
    self._author_policy(
        container,
        sd,
        profiles,
        tier,
        upstream_dns,
        self._gateways,
        security_deny=security_deny,
        provider_allow=provider_allow,
        project_allow=project_allow,
        override=override,
    )

allow_domain(container, domain)

Record +domain in the runtime overlay and reload dnsmasq.

The overlay (policy/live) flips any prior deny of domain and survives reloads; the dnsmasq restart picks up the new nftset= line so future IP rotations of domain are auto-populated. The IP-level allow (nft set update) is handled separately by allow_ip().

No-op when the container runs no dnsmasq (the static IP-level allow already happened via allow_ip()).

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

def allow_domain(self, container: str, domain: str) -> None:
    """Record ``+domain`` in the runtime overlay and reload dnsmasq.

    The overlay (``policy/live``) flips any prior deny of *domain* and
    survives reloads; the dnsmasq restart picks up the new ``nftset=``
    line so future IP rotations of *domain* are auto-populated.  The
    IP-level allow (nft set update) is handled separately by ``allow_ip()``.

    No-op when the container runs no dnsmasq (the static IP-level allow
    already happened via ``allow_ip()``).
    """
    sd = self._config.state_dir.resolve()
    if not _runs_dnsmasq(sd):
        return
    StateBundle(sd).overlay_set("+", domain)
    self._reload_dnsmasq(container, sd)

deny_domain(container, domain)

Record -domain in the runtime overlay and reload dnsmasq.

Counterpart of allow_domain(): the dnsmasq restart picks up the local= sinkhole, so domain stops resolving (NXDOMAIN) and the deny fails fast in the DNS plane instead of timing out against the filter.

No-op when the container runs no dnsmasq.

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

def deny_domain(self, container: str, domain: str) -> None:
    """Record ``-domain`` in the runtime overlay and reload dnsmasq.

    Counterpart of ``allow_domain()``: the dnsmasq restart picks up the
    ``local=`` sinkhole, so *domain* stops resolving (NXDOMAIN) and the
    deny fails fast in the DNS plane instead of timing out against the
    filter.

    No-op when the container runs no dnsmasq.
    """
    sd = self._config.state_dir.resolve()
    if not _runs_dnsmasq(sd):
        return
    StateBundle(sd).overlay_set("-", domain)
    self._reload_dnsmasq(container, sd)

allow_ip(container, ip)

Live-allow an IP for a running container via nsenter.

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

def allow_ip(self, container: str, ip: str) -> None:
    """Live-allow an IP for a running container via nsenter."""
    ip = safe_ip(ip)
    sd = self._config.state_dir.resolve()
    bundle = StateBundle(sd)

    # Un-deny: drop from the nft deny set if it is currently denied.
    if ip in bundle.read_denied_ips():
        nft_cmd = delete_deny_elements_dual([ip])
        if nft_cmd:
            self._nft_apply_best_effort(container, nft_cmd)

    # When the live set has a default timeout (30 m), permanent IPs must use
    # 'timeout 0s' so they are never evicted by the set's per-element expiry clock.
    tier = bundle.read_dns_tier()
    element = f"{{ {ip} timeout 0s }}" if tier is not None and tier.live else f"{{ {ip} }}"

    self._runner.nft_via_nsenter(
        container,
        "add",
        "element",
        "inet",
        "terok_shield",
        self._set_for_ip(ip),
        element,
    )
    # Persist to the runtime overlay (flips any prior deny of this IP).
    bundle.overlay_set("+", ip)

deny_ip(container, ip)

Live-deny an IP for a running container via nsenter.

Removes from the nft allow set (best-effort), adds to the nft deny set, and records -ip in policy/live so the deny sticks across shield up / restart and flips any prior allow.

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

def deny_ip(self, container: str, ip: str) -> None:
    """Live-deny an IP for a running container via nsenter.

    Removes from the nft allow set (best-effort), adds to the nft deny set,
    and records ``-ip`` in ``policy/live`` so the deny sticks across
    ``shield up`` / restart and flips any prior allow.
    """
    ip = safe_ip(ip)
    sd = self._config.state_dir.resolve()
    bundle = StateBundle(sd)

    # Best-effort nft delete (IP may not be in the set)
    try:
        self._runner.nft_via_nsenter(
            container,
            "delete",
            "element",
            "inet",
            "terok_shield",
            self._set_for_ip(ip),
            f"{{ {ip} }}",
        )
    except ExecError as e:
        stderr = str(e).lower()
        if not any(
            pat in stderr for pat in ("no such file", "element does not exist", "not in set")
        ):
            logger.warning("nft delete element failed for %s: %s", ip, e)

    # Add to nft deny set (prevents dnsmasq from re-allowing)
    nft_cmd = add_deny_elements_dual([ip])
    if nft_cmd:
        self._nft_apply_best_effort(container, nft_cmd)

    # Persist to the runtime overlay (flips any prior allow; sticks across restart).
    bundle.overlay_set("-", ip)

shield_down(container, *, disengaged=False)

Switch a running container to the DOWN posture (DISENGAGED when disengaged).

Plain DOWN accepts by default but keeps the deny set and both range floors; DISENGAGED enforces nothing, so its deny sets are left empty — the next shield up repopulates them from the composed policy.

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

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

    Plain DOWN accepts by default but keeps the deny set and both range
    floors; DISENGAGED enforces nothing, so its deny sets are left empty —
    the next ``shield up`` repopulates them from the composed policy.
    """
    sd = self._config.state_dir.resolve()
    ruleset = self._container_ruleset(container)
    rs = ruleset.build_down(disengaged=disengaged)
    current = self.shield_state(container)
    if current == ShieldState.OFFLINE:
        stdin = rs
    else:
        stdin = f"delete table {NFT_TABLE}\n{rs}"
    snapshot = [] if current == ShieldState.OFFLINE else self._snapshot_allow_sets(container)
    self._runner.nft_via_nsenter(container, stdin=stdin)

    # Carry the allow-set contents (seeds + dnsmasq-learned IPs) across
    # the rebuild — the down posture does not evaluate them, but the later
    # ``shield up`` snapshots this table, so dropping them here would
    # forget every learned IP after one down/up round trip.
    self._restore_allow_sets(container, snapshot, skip=())

    # Repopulate deny sets so the deny policy is enforced even when shield
    # is down, and the t10 override set so break-glass hosts stay above it.
    # DISENGAGED references neither set, so there is nothing to reseed.
    if not disengaged:
        self._reseed_deny_and_override(container, sd)

    output = self._runner.nft_via_nsenter(
        container,
        "list",
        "table",
        "inet",
        NFT_TABLE_NAME,
    )
    errors = ruleset.verify_down(output, disengaged=disengaged)
    if errors:
        raise RuntimeError(f"Shield-down ruleset verification failed: {'; '.join(errors)}")

shield_quarantine(container)

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

Reads no settings — no DNS, no allowlists, no loopback ports, no gateway probe, no profile lookup. build_quarantine / verify_quarantine are static; the only inputs are the container name and the live ruleset state (table-or-no-table). Any config-conditional branch added here is a bug.

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

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

    Reads no settings — no DNS, no allowlists, no loopback ports,
    no gateway probe, no profile lookup.  ``build_quarantine`` /
    ``verify_quarantine`` are static; the only inputs are the
    container name and the live ruleset state (table-or-no-table).
    Any config-conditional branch added here is a bug.
    """
    rs = RulesetBuilder.build_quarantine()
    current = self.shield_state(container)
    stdin = rs if current == ShieldState.OFFLINE else f"delete table {NFT_TABLE}\n{rs}"
    self._runner.nft_via_nsenter(container, stdin=stdin)
    output = self._runner.nft_via_nsenter(
        container,
        "list",
        "table",
        "inet",
        NFT_TABLE_NAME,
    )
    errors = RulesetBuilder.verify_quarantine(output)
    if errors:
        raise RuntimeError(f"Quarantine ruleset verification failed: {'; '.join(errors)}")

shield_up(container)

Restore normal deny-all mode for a running container.

The rebuild is delete table + re-apply, which would forget every dnsmasq-learned allow-set element — a container coming out of the down posture would suddenly lose IPs its workload already resolved (clients cache answers, so they do not necessarily re-query). The allow sets are therefore snapshotted before the rebuild and restored after it.

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

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

    The rebuild is ``delete table`` + re-apply, which would forget every
    dnsmasq-learned allow-set element — a container coming out of the down
    posture would suddenly lose IPs its workload already resolved (clients
    cache answers, so they do not necessarily re-query).  The allow sets
    are therefore snapshotted before the rebuild and restored after it.
    """
    sd = self._config.state_dir.resolve()

    ruleset = self._container_ruleset(container)
    rs = ruleset.build_up()
    current = self.shield_state(container)
    if current == ShieldState.OFFLINE:
        stdin = rs
    else:
        stdin = f"delete table {NFT_TABLE}\n{rs}"
    snapshot = [] if current == ShieldState.OFFLINE else self._snapshot_allow_sets(container)
    self._runner.nft_via_nsenter(container, stdin=stdin)

    # Re-add effective IPs (allowed minus denied)
    unique_ips = StateBundle(sd).read_effective_ips()
    if unique_ips:
        elements_cmd = ruleset.add_elements_dual(unique_ips)
        if elements_cmd:
            self._runner.nft_via_nsenter(container, stdin=elements_cmd)

    # Repopulate the deny sets and the t10 override set from the bundle
    denied_ips = self._reseed_deny_and_override(container, sd)

    # Restore the snapshot, minus everything the rebuild already re-added
    # (a duplicate/overlapping element would abort the nft transaction)
    # and minus denied entries (deny_ip() removed them from the allow set
    # deliberately — a down/up round trip must not resurrect them).
    self._restore_allow_sets(container, snapshot, skip=[*unique_ips, *denied_ips])

    # Gateway addresses are baked into the ruleset — no repopulation needed.

    output = self._runner.nft_via_nsenter(
        container,
        "list",
        "table",
        "inet",
        NFT_TABLE_NAME,
    )
    errors = ruleset.verify_up(output)
    if errors:
        raise RuntimeError(f"Ruleset verification failed: {'; '.join(errors)}")

shield_reset(container)

Forget learned allow-set state — back to the just-launched contents.

Flushes both tier-40 project-allow sets and re-seeds them from the effective policy in a single nft transaction, so authored literals never blink out. dnsmasq-learned IPs vanish until the workload resolves the corresponding names again; the operator overlay (policy/live) and the deny tier are untouched.

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

def shield_reset(self, container: str) -> None:
    """Forget learned allow-set state — back to the just-launched contents.

    Flushes both tier-40 project-allow sets and re-seeds them from the
    effective policy in a single nft transaction, so authored literals
    never blink out.  dnsmasq-learned IPs vanish until the workload
    resolves the corresponding names again; the operator overlay
    (``policy/live``) and the deny tier are untouched.
    """
    sd = self._config.state_dir.resolve()
    ruleset = self._container_ruleset(container)
    stdin = (
        f"flush set {NFT_TABLE} {TIER_PROJECT_ALLOW}_v4\n"
        f"flush set {NFT_TABLE} {TIER_PROJECT_ALLOW}_v6\n"
    )
    stdin += ruleset.add_elements_dual(StateBundle(sd).read_effective_ips())
    self._runner.nft_via_nsenter(container, stdin=stdin)

shield_state(container)

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

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

def shield_state(self, container: str) -> ShieldState:
    """Query the live nft ruleset to determine the container's shield state."""
    output = self.list_rules(container)
    if not output.strip():
        return ShieldState.OFFLINE

    # verify_* returns a list of errors; empty list = ruleset is valid.
    # Block is checked first: its minimal ruleset (no sets, no DNS)
    # would fail all other verifiers.
    if not self._ruleset.verify_quarantine(output):
        return ShieldState.QUARANTINE

    if not self._ruleset.verify_down(output, disengaged=False):
        return ShieldState.DOWN
    if not self._ruleset.verify_down(output, disengaged=True):
        return ShieldState.DISENGAGED

    if not self._ruleset.verify_up(output):
        return ShieldState.UP

    return ShieldState.ERROR

list_rules(container)

List current nft rules for a running container.

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

def list_rules(self, container: str) -> str:
    """List current nft rules for a running container."""
    try:
        return self._runner.nft_via_nsenter(
            container,
            "list",
            "table",
            "inet",
            "terok_shield",
            check=False,
        )
    except ExecError:
        return ""

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

Generate the ruleset that would be applied to a container.

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

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