rules

rules

nftables ruleset generation and verification.

Generates per-container nftables rulesets as an ordered tier policy and provides set operations for runtime allow/deny/override management, plus verification of applied rulesets against security invariants.

The UP ruleset is a single output chain whose body is an ordered list of tier rules. nft evaluates them top to bottom: an accept/reject is a terminal verdict (short-circuit), and a non-match falls through to the next tier (a Pass). Tier order is the authority order::

preamble          accept lo / established / DNS / infra ports
t00 hard-deny     reject @HARD_DENY_RANGES   (link-local/IMDS — absolute)
t10 override      accept @override           (break-glass, above the deny)
t20 security-deny reject @deny + @PRIVATE_RANGES  (vault hosts + RFC1918)
t30/40 allow      accept @allow              (provider + project)
bypass window     accept @bypass_window      (kernel-timed allow-all)
terminal          reject (log BLOCKED)

Because the deny tier sits above the allow tier, an explicit deny wins over an allow; an override (t10) sits above the deny and is the only way to reach a security-denied host. The hard-deny floor (t00) sits above the override and is absolute.

Security boundary: only stdlib + nft.constants imports. All inputs are validated before interpolation into nft commands.

RulesetBuilder(*, dns=PASTA_DNS, loopback_ports=(), gateway_v4='', gateway_v6='', set_timeout='')

Builder for nftables ruleset generation and verification.

Security boundary: only stdlib + nft.constants imports. All inputs validated before interpolation.

Binds dns, loopback_ports, gateways, and the dnsmasq set timeout once at construction so callers do not repeat them on every call.

Create a builder with validated DNS, gateway, and port config.

Parameters:

Name	Type	Description	Default
dns	str	DNS server address (pasta default forwarder).	PASTA_DNS
loopback_ports	tuple[int, ...]	TCP ports to allow on the host-loopback map address.	()
gateway_v4	str	IPv4 gateway address (e.g. slirp4netns 10.0.2.2).	''
gateway_v6	str	IPv6 gateway address (e.g. slirp4netns fd00::2).	''
set_timeout	str	dnsmasq-tier element timeout for the allow sets (e.g. 30m).	''

Source code in src/terok_shield/nft/rules.py

def __init__(
    self,
    *,
    dns: str = PASTA_DNS,
    loopback_ports: tuple[int, ...] = (),
    gateway_v4: str = "",
    gateway_v6: str = "",
    set_timeout: str = "",
) -> None:
    """Create a builder with validated DNS, gateway, and port config.

    Args:
        dns: DNS server address (pasta default forwarder).
        loopback_ports: TCP ports to allow on the host-loopback map address.
        gateway_v4: IPv4 gateway address (e.g. slirp4netns ``10.0.2.2``).
        gateway_v6: IPv6 gateway address (e.g. slirp4netns ``fd00::2``).
        set_timeout: dnsmasq-tier element timeout for the allow sets (e.g. ``30m``).
    """
    dns = safe_ip(dns)
    for p in loopback_ports:
        _safe_port(p)
    if set_timeout:
        _safe_timeout(set_timeout)
    self._dns = dns
    self._loopback_ports = loopback_ports
    self._gateway_v4 = _safe_ipv4(gateway_v4) if gateway_v4 else ""
    self._gateway_v6 = _safe_ipv6(gateway_v6) if gateway_v6 else ""
    self._set_timeout = set_timeout

build_up()

Generate the UP (deny-all + ordered tiers) ruleset.

Applied by the OCI hook into the container's own netns. Dual-stack. Infra ports (DNS, host-loopback proxy, gateway) are accepted in the preamble before any tier, so the control plane survives the hard-deny of link-local space. See the module docstring for the tier order.

Source code in src/terok_shield/nft/rules.py

def build_up(self) -> str:
    """Generate the UP (deny-all + ordered tiers) ruleset.

    Applied by the OCI hook into the container's own netns.  Dual-stack.
    Infra ports (DNS, host-loopback proxy, gateway) are accepted in the
    preamble *before* any tier, so the control plane survives the hard-deny
    of link-local space.  See the module docstring for the tier order.
    """
    body = self._join(
        self._preamble_lines(),
        self._range_reject(HARD_DENY_RANGES, PRIVATE_LOG_PREFIX),  # t00 absolute
        self._match(TIER_OVERRIDE, "accept", ALLOWED_LOG_PREFIX),  # t10 break-glass
        self._match(TIER_SECURITY_DENY, _REJECT, DENIED_LOG_PREFIX),  # t20 deny set
        self._range_reject(PRIVATE_RANGES, PRIVATE_LOG_PREFIX),  # t20 RFC1918
        self._match(TIER_PROVIDER_ALLOW, "accept", ALLOWED_LOG_PREFIX),  # t30 provider
        self._match(TIER_PROJECT_ALLOW, "accept", ALLOWED_LOG_PREFIX),  # t40 project
        self._match(SET_BYPASS_WINDOW, "accept", BYPASS_LOG_PREFIX),  # timed window
        self._terminal(),  # terminal deny
    )
    return self._table(self._set_decls(), body, policy="drop")

build_down(*, disengaged=False)

Generate the DOWN-posture (manual shield down) ruleset.

Output policy is accept and every new connection is logged with the bypass prefix. Plain DOWN still enforces the hard-deny floor and the security-deny tier (deny set + private ranges), with the t10 override kept above the deny — a break-glass host must stay reachable in every posture that enforces the deny.

Parameters:

Name	Type	Description	Default
disengaged	bool	If True (DISENGAGED), enforce nothing: no hard-deny floor, no deny set, no private-range rejects — every destination is accepted and logged. The tier sets stay declared (unreferenced) so allow-set contents survive the round trip back to shield up. Private addresses below the deny are reachable in the other postures through a t10 override — a host, or a whole CIDR (logged as a warning); DISENGAGED opens everything without one.	False

Source code in src/terok_shield/nft/rules.py

def build_down(self, *, disengaged: bool = False) -> str:
    """Generate the DOWN-posture (manual ``shield down``) ruleset.

    Output policy is ``accept`` and every new connection is logged with
    the bypass prefix.  Plain DOWN still enforces the hard-deny floor and
    the security-deny tier (deny set + private ranges), with the t10
    override kept *above* the deny — a break-glass host must stay
    reachable in every posture that enforces the deny.

    Args:
        disengaged: If True (DISENGAGED), enforce nothing: no hard-deny
            floor, no deny set, no private-range rejects — every
            destination is accepted and logged.  The tier sets stay
            declared (unreferenced) so allow-set contents survive the
            round trip back to ``shield up``.  Private addresses below
            the deny are reachable in the other postures through a t10
            override — a host, or a whole CIDR (logged as a warning);
            DISENGAGED opens everything without one.
    """
    sections = [self._preamble_lines()]
    if not disengaged:
        sections.extend(
            (
                self._range_reject(HARD_DENY_RANGES, PRIVATE_LOG_PREFIX),
                self._match(TIER_OVERRIDE, "accept", BYPASS_LOG_PREFIX),
                self._match(TIER_SECURITY_DENY, _REJECT, DENIED_LOG_PREFIX),
                self._range_reject(PRIVATE_RANGES, PRIVATE_LOG_PREFIX),
            )
        )
    sections.append(
        f'        ct state new log group {NFLOG_GROUP} prefix "{BYPASS_LOG_PREFIX}: " counter'
    )
    return self._table(self._set_decls(), self._join(*sections), policy="accept")

build_quarantine() staticmethod

Generate the quarantine-mode (total blackout) ruleset.

Drops all traffic except loopback and established connections. No DNS, no allowlists, no gateway ports. All dropped packets are tagged for the audit log.

Source code in src/terok_shield/nft/rules.py

@staticmethod
def build_quarantine() -> str:
    """Generate the quarantine-mode (total blackout) ruleset.

    Drops all traffic except loopback and established connections.
    No DNS, no allowlists, no gateway ports.  All dropped packets
    are tagged for the audit log.
    """
    blocked_log = f'        log group {NFLOG_GROUP} prefix "{BLOCKED_LOG_PREFIX}: " drop'
    return textwrap.dedent(f"""\
        table {NFT_TABLE} {{
            chain output {{
                type filter hook output priority filter; policy drop;
                oifname "lo" accept
                ct state established,related accept
        {blocked_log}
            }}

            chain input {{
                type filter hook input priority filter; policy drop;
                iifname "lo" accept
                ct state established,related accept
                drop
            }}
        }}
    """)

verify_up(nft_output)

Check applied UP ruleset invariants. Returns errors (empty = OK).

Expects output from nft list table inet terok_shield (scoped to the managed table), not nft list ruleset. Verifies the table header, policy drop, both chains, the reject type, every tier set, the terminal deny-all rule, and both range-reject floors.

Source code in src/terok_shield/nft/rules.py

def verify_up(self, nft_output: str) -> list[str]:
    """Check applied UP ruleset invariants.  Returns errors (empty = OK).

    Expects output from ``nft list table inet terok_shield`` (scoped to the
    managed table), not ``nft list ruleset``.  Verifies the table header,
    ``policy drop``, both chains, the reject type, every tier set, the
    terminal deny-all rule, and both range-reject floors.
    """
    errors: list[str] = []
    if f"table {NFT_TABLE}" not in nft_output:
        errors.append(f"managed table '{NFT_TABLE}' not found in output")
    if "policy drop" not in nft_output:
        errors.append("policy is not drop")
    errors.extend(self._verify_common(nft_output))
    # Terminal deny-all: a standalone log+reject with the BLOCKED prefix
    # (no daddr selector, unlike the tier rules).  Require the ``reject``
    # verdict on the same rule so a regression to a silent ``drop`` fails
    # verification instead of passing on the log line alone.
    if not re.search(
        rf'^\s*log\s+.*prefix\s+"{re.escape(BLOCKED_LOG_PREFIX)}.*\breject\b',
        nft_output,
        re.MULTILINE,
    ):
        errors.append("terminal reject-all rule missing")
    errors.extend(self._verify_ranges(nft_output, HARD_DENY_RANGES, "Hard-deny"))
    errors.extend(self._verify_ranges(nft_output, PRIVATE_RANGES, "Private-range"))
    return errors

verify_down(nft_output, *, disengaged=False)

Check applied DOWN-posture ruleset invariants. Returns errors (empty = OK).

Verifies the table header, policy accept on output / drop on input, both chains, every tier set, and the bypass nflog prefix. Plain DOWN must carry both range-reject floors; DISENGAGED must carry neither floor nor the deny-set reject — a partially applied ruleset that keeps any reject must not pass as DISENGAGED.

Source code in src/terok_shield/nft/rules.py

def verify_down(self, nft_output: str, *, disengaged: bool = False) -> list[str]:
    """Check applied DOWN-posture ruleset invariants.  Returns errors (empty = OK).

    Verifies the table header, ``policy accept`` on output / ``drop`` on
    input, both chains, every tier set, and the bypass nflog prefix.
    Plain DOWN must carry both range-reject floors; DISENGAGED must carry
    neither floor nor the deny-set reject — a partially applied ruleset
    that keeps any reject must not pass as DISENGAGED.
    """
    errors: list[str] = []
    if f"table {NFT_TABLE}" not in nft_output:
        errors.append(f"managed table '{NFT_TABLE}' not found in output")
    if "policy accept" not in nft_output:
        errors.append("output policy is not accept")
    if "policy drop" not in nft_output:
        errors.append("input policy is not drop")
    errors.extend(self._verify_common(nft_output, expect_reject=not disengaged))
    if BYPASS_LOG_PREFIX not in nft_output:
        errors.append("bypass nflog prefix missing")
    floors = ((HARD_DENY_RANGES, "Hard-deny"), (PRIVATE_RANGES, "Private-range"))
    if disengaged:
        for nets, label in floors:
            errors.extend(self._verify_ranges_absent(nft_output, nets, label))
        if re.search(rf"@{TIER_SECURITY_DENY}_v[46]\b.*\breject\b", nft_output):
            errors.append("deny-set reject rule present in DISENGAGED ruleset")
    else:
        for nets, label in floors:
            errors.extend(self._verify_ranges(nft_output, nets, label))
    return errors

verify_quarantine(nft_output) staticmethod

Check applied quarantine ruleset invariants. Returns errors (empty = OK).

Verifies the table header, both chains with policy drop, the blocked log prefix, and that no allow sets exist (total blackout).

Source code in src/terok_shield/nft/rules.py

@staticmethod
def verify_quarantine(nft_output: str) -> list[str]:
    """Check applied quarantine ruleset invariants.  Returns errors (empty = OK).

    Verifies the table header, both chains with ``policy drop``, the blocked
    log prefix, and that no allow sets exist (total blackout).
    """
    errors: list[str] = []
    if f"table {NFT_TABLE}" not in nft_output:
        errors.append(f"managed table '{NFT_TABLE}' not found in output")
    if "policy drop" not in nft_output:
        errors.append("policy is not drop")
    for chain in ("output", "input"):
        if f"chain {chain}" not in nft_output:
            errors.append(f"{chain} chain missing")
    if BLOCKED_LOG_PREFIX not in nft_output:
        errors.append("blocked nflog prefix missing")
    for base in (TIER_PROVIDER_ALLOW, TIER_PROJECT_ALLOW):
        for fam in ("v4", "v6"):
            sname = f"{base}_{fam}"
            if sname in nft_output:
                errors.append(f"{sname} set present in quarantine mode")
    return errors

add_elements_dual(ips)

Add IPs to the tier-40 project-allow sets, honouring the dnsmasq permanent-element rule.

When a set_timeout is configured (live tier), profile/live IPs are written with timeout 0s so they do not auto-expire with the dnsmasq-learned entries.

Source code in src/terok_shield/nft/rules.py

def add_elements_dual(self, ips: list[str]) -> str:
    """Add IPs to the tier-40 project-allow sets, honouring the dnsmasq permanent-element rule.

    When a ``set_timeout`` is configured (live tier), profile/live IPs are
    written with ``timeout 0s`` so they do not auto-expire with the
    dnsmasq-learned entries.
    """
    return add_elements_dual(ips, permanent=bool(self._set_timeout))

add_elements_dual(ips, *, permanent=False)

Add IPs to the tier-40 project-allow sets, split by address family.

Parameters:

Name	Type	Description	Default
permanent	bool	annotate elements with timeout 0s so profile/live IPs never expire in the dnsmasq-tier allow set (which carries a default element timeout for learned IPs).	False

Source code in src/terok_shield/nft/rules.py

def add_elements_dual(ips: list[str], *, permanent: bool = False) -> str:
    """Add IPs to the tier-40 project-allow sets, split by address family.

    Args:
        permanent: annotate elements with ``timeout 0s`` so profile/live IPs
            never expire in the dnsmasq-tier allow set (which carries a default
            element timeout for learned IPs).
    """
    return _emit_dual(add_elements, TIER_PROJECT_ALLOW, ips, timeout_zero=permanent)

add_deny_elements_dual(ips)

Add IPs to the tier-20 security-deny sets, split by address family.

Source code in src/terok_shield/nft/rules.py

def add_deny_elements_dual(ips: list[str]) -> str:
    """Add IPs to the tier-20 security-deny sets, split by address family."""
    return _emit_dual(add_elements, TIER_SECURITY_DENY, ips)

add_override_elements_dual(ips)

Add IPs to the tier-10 override (break-glass) sets, split by address family.

Source code in src/terok_shield/nft/rules.py

def add_override_elements_dual(ips: list[str]) -> str:
    """Add IPs to the tier-10 override (break-glass) sets, split by address family."""
    return _emit_dual(add_elements, TIER_OVERRIDE, ips)

delete_deny_elements_dual(ips)

Remove IPs from the tier-20 security-deny sets, split by address family.

Source code in src/terok_shield/nft/rules.py

def delete_deny_elements_dual(ips: list[str]) -> str:
    """Remove IPs from the tier-20 security-deny sets, split by address family."""
    return _emit_dual(delete_elements, TIER_SECURITY_DENY, ips)

arm_bypass_window(timeout)

Open the timed allow-all window.

Adds 0.0.0.0/0 / ::/0 to the bypass_window sets with a kernel timeout so the window closes itself when the element expires — no userspace timer, fail-closed (any disruption only closes it sooner).

Source code in src/terok_shield/nft/rules.py

def arm_bypass_window(timeout: str) -> str:
    """Open the timed allow-all window.

    Adds ``0.0.0.0/0`` / ``::/0`` to the ``bypass_window`` sets with a kernel
    ``timeout`` so the window closes itself when the element expires — no
    userspace timer, fail-closed (any disruption only closes it sooner).
    """
    _safe_timeout(timeout)
    return (
        f"add element {NFT_TABLE} {SET_BYPASS_WINDOW}_v4 {{ 0.0.0.0/0 timeout {timeout} }}\n"
        f"add element {NFT_TABLE} {SET_BYPASS_WINDOW}_v6 {{ ::/0 timeout {timeout} }}\n"
    )

disarm_bypass_window()

Close the timed allow-all window immediately by flushing both sets.

Source code in src/terok_shield/nft/rules.py

def disarm_bypass_window() -> str:
    """Close the timed allow-all window immediately by flushing both sets."""
    return (
        f"flush set {NFT_TABLE} {SET_BYPASS_WINDOW}_v4\n"
        f"flush set {NFT_TABLE} {SET_BYPASS_WINDOW}_v6\n"
    )

parse_set_elements(nft_output)

Parse nft list set output into (ip, timeout) pairs.

timeout is the element's printed timeout token ("30m", "0s", compound "1h22m") or "" when the element carries none. The remaining expires countdown is deliberately dropped — a restore via restore_elements re-grants the full timeout. Atoms that fail validation are skipped rather than raised: the snapshot/restore path is best-effort by design (a dropped learned IP is re-learned on the workload's next DNS answer), and a parse quirk must never abort a shield state transition.

Source code in src/terok_shield/nft/rules.py

def parse_set_elements(nft_output: str) -> list[tuple[str, str]]:
    """Parse ``nft list set`` output into ``(ip, timeout)`` pairs.

    *timeout* is the element's printed timeout token (``"30m"``, ``"0s"``,
    compound ``"1h22m"``) or ``""`` when the element carries none.  The
    remaining ``expires`` countdown is deliberately dropped — a restore via
    [`restore_elements`][terok_shield.nft.rules.restore_elements] re-grants
    the full timeout.  Atoms that fail validation are skipped rather than
    raised: the snapshot/restore path is best-effort by design (a dropped
    learned IP is re-learned on the workload's next DNS answer), and a
    parse quirk must never abort a shield state transition.
    """
    block = _ELEMENTS_BLOCK_RE.search(nft_output)
    if not block:
        return []
    pairs: list[tuple[str, str]] = []
    for atom in block.group(1).split(","):
        tokens = atom.split()
        if not tokens:
            continue
        try:
            ip = safe_ip(tokens[0])
        except ValueError:
            continue
        timeout = ""
        if len(tokens) >= 3 and tokens[1] == "timeout" and _ELEMENT_TIMEOUT_RE.fullmatch(tokens[2]):
            timeout = tokens[2]
        pairs.append((ip, timeout))
    return pairs

restore_elements(set_name, elements, table=NFT_TABLE)

Generate an nft command re-adding dumped (ip, timeout) elements.

Counterpart of parse_set_elements: every input is re-validated before interpolation (the snapshot crosses a subprocess boundary, so it is treated as untrusted like any other input). Timed elements are re-granted their full timeout. Returns "" when there is nothing to restore.

Source code in src/terok_shield/nft/rules.py

def restore_elements(set_name: str, elements: list[tuple[str, str]], table: str = NFT_TABLE) -> str:
    """Generate an nft command re-adding dumped ``(ip, timeout)`` elements.

    Counterpart of [`parse_set_elements`][terok_shield.nft.rules.parse_set_elements]:
    every input is re-validated before interpolation (the snapshot crosses a
    subprocess boundary, so it is treated as untrusted like any other input).
    Timed elements are re-granted their full timeout.  Returns ``""`` when
    there is nothing to restore.
    """
    set_name = _safe_ident(set_name)
    table = " ".join(_safe_ident(part) for part in table.split())
    parts: list[str] = []
    for ip, timeout in elements:
        ip = safe_ip(ip)
        if not timeout:
            parts.append(ip)
            continue
        if not _ELEMENT_TIMEOUT_RE.fullmatch(timeout):
            raise ValueError(f"Invalid element timeout: {timeout!r}")
        parts.append(f"{ip} timeout {timeout}")
    if not parts:
        return ""
    return f"add element {table} {set_name} {{ {', '.join(parts)} }}\n"

add_elements(set_name, ips, table=NFT_TABLE, *, timeout_zero=False)

Generate an nft command to add validated IPs to a set.

Both set_name and table are validated against injection. Returns empty string if no valid IPs.

Parameters:

Name	Type	Description	Default
timeout_zero	bool	annotate each element with timeout 0s so it never expires, even in a set that carries a default element timeout.	False

Source code in src/terok_shield/nft/rules.py

def add_elements(
    set_name: str, ips: list[str], table: str = NFT_TABLE, *, timeout_zero: bool = False
) -> str:
    """Generate an nft command to add validated IPs to a set.

    Both ``set_name`` and ``table`` are validated against injection.
    Returns empty string if no valid IPs.

    Args:
        timeout_zero: annotate each element with ``timeout 0s`` so it never
            expires, even in a set that carries a default element timeout.
    """
    set_name = _safe_ident(set_name)
    table = " ".join(_safe_ident(part) for part in table.split())
    valid = [safe_ip(ip) for ip in ips if _try_validate(ip)]
    if not valid:
        return ""
    if timeout_zero:
        elements = ", ".join(f"{ip} timeout 0s" for ip in valid)
    else:
        elements = ", ".join(valid)
    return f"add element {table} {set_name} {{ {elements} }}\n"

delete_elements(set_name, ips, table=NFT_TABLE)

Generate an nft command to delete validated IPs from a set.

Both set_name and table are validated against injection. Returns empty string if no valid IPs.

Source code in src/terok_shield/nft/rules.py

def delete_elements(set_name: str, ips: list[str], table: str = NFT_TABLE) -> str:
    """Generate an nft command to delete validated IPs from a set.

    Both ``set_name`` and ``table`` are validated against injection.
    Returns empty string if no valid IPs.
    """
    set_name = _safe_ident(set_name)
    table = " ".join(_safe_ident(part) for part in table.split())
    valid = [safe_ip(ip) for ip in ips if _try_validate(ip)]
    if not valid:
        return ""
    elements = ", ".join(valid)
    return f"delete element {table} {set_name} {{ {elements} }}\n"

safe_ip(value)

Validate and normalize an IPv4 or IPv6 address or CIDR notation.

Prevents nft command injection by ensuring the value is a valid IP address or network. Returns the canonical string form so comparisons across state files are reliable regardless of input notation.

Raises ValueError on invalid input.

Source code in src/terok_shield/nft/rules.py

def safe_ip(value: str) -> str:
    """Validate and normalize an IPv4 or IPv6 address or CIDR notation.

    Prevents nft command injection by ensuring the value is a valid IP address
    or network.  Returns the canonical string form so comparisons across state
    files are reliable regardless of input notation.

    Raises ValueError on invalid input.
    """
    v = value.strip()
    try:
        if "/" in v:
            return str(ipaddress.ip_network(v, strict=False))
        return str(ipaddress.ip_address(v))
    except ValueError as e:
        raise ValueError(f"Invalid IP/CIDR: {v!r}") from e