computer-playbook/tests/unit/python/roles/svc-net-firewall/test_fragments.py

"""Every role's nftables fragment parses, and replaces its table instead of merging.

``nft -f`` on ``table inet x { ... }`` MERGES into an existing table: a rule
dropped from the declaration stays in the kernel, and the ruleset can only ever
grow. The three-line idiom below is what makes a fragment a replacement:

    table inet x          <- creates it when absent, so the delete cannot fail
    delete table inet x   <- removes whatever the last deploy left
    table inet x { ... }  <- the declaration, applied in the same transaction

``nft -f`` is atomic, so the table is never observed empty. Without the first
line the delete fails on a fresh host; without the second the fragment merges.

The checks run the real ``nft`` over the rendered text, so a fragment that is
valid Jinja but invalid nftables fails here rather than on the host. They render
every deploy mode, because a fragment may branch on one, and feed plausible
rather than real values: nft needs shapes, not this host's addresses.
"""

from __future__ import annotations

import os
import re
import shutil
import subprocess
import tempfile
import unittest
from contextlib import contextmanager
from functools import lru_cache
from pathlib import Path

import jinja2

from plugins.filter.swarm_ingress_guard import (
    swarm_ingress_published,
    swarm_node_addresses,
)
from utils.cache.files import iter_project_files, read_text
from utils.cache.yaml import load_yaml_any
from utils.roles.mapping import ROLE_FILE_VARS_MAIN

from . import PROJECT_ROOT

FRAGMENT = "templates/nftables.conf.j2"
TABLE = "infinito_probe"

PROBE_PORTS = {
    "TOR_TRANS_PORT": "9040",
    "TOR_DNS_PORT": "9053",
    "TOR_SOCKS_PORT": "9050",
    "TOR_DNSMASQ_PORT": "53",
}

_PORT_VARIABLE = re.compile(r"\{\{\s*([A-Z][A-Z0-9_]*)\s*\}\}")


def _guarded_ports() -> list[dict]:
    """The role's own guard list, with its port variables filled from stand-ins.

    The shape has to come from the declaration: an entry that gains a key, a
    protocol or a whole port must reach the rendered table here, or this check
    keeps passing on a fixture the role stopped shipping.
    """
    declared = load_yaml_any(
        str(PROJECT_ROOT / "roles" / "svc-net-tor" / ROLE_FILE_VARS_MAIN)
    )["TOR_EGRESS_GUARDED_PORTS"]
    return [
        {**entry, "port": PROBE_PORTS[_PORT_VARIABLE.search(entry["port"]).group(1)]}
        for entry in declared
    ]


COMMON = {
    "FIREWALL_TABLE": TABLE,
    "TOR_EGRESS_CLIENT_CIDRS": ["127.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16"],
    "TOR_EGRESS_GUARDED_PORTS": _guarded_ports(),
    "TOR_EGRESS_VIRTUAL_NET_IPV4": "10.192.0.0/10",
    **PROBE_PORTS,
    "TOR_CONTAINER_DNS_HOST": "172.17.0.1",
    "NETWORK_INTERNAL_CIDRS": [
        "127.0.0.0/8",
        "10.0.0.0/8",
        "172.16.0.0/12",
        "192.168.0.0/16",
    ],
    "SWARM_INGRESS_GUARDED_PORTS": swarm_ingress_published(
        [
            (
                '[{"Protocol":"tcp","TargetPort":5432,"PublishedPort":5432,"PublishMode":"ingress"},'
                '{"Protocol":"udp","TargetPort":3478,"PublishedPort":3478,"PublishMode":"ingress"}]'
            )
        ]
    ),
    "SWARM_INGRESS_PEER_ADDRESSES": swarm_node_addresses(["10.0.0.2"]),
}

MODES = ("compose", "swarm")


def _run(argv: list[str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(argv, capture_output=True, text=True, check=False)


@lru_cache(maxsize=1)
def _nft_can_check() -> bool:
    """Whether `nft -c` works here.

    It is not a parser: it opens netlink and builds a cache against the live
    kernel first, so on a runner without CAP_NET_ADMIN it fails on a valid
    ruleset. Probing the binary's presence would make this suite red there.
    """
    if shutil.which("nft") is None:
        return False
    with tempfile.NamedTemporaryFile("w", suffix=".conf") as handle:
        handle.write("table inet infinito_capability_probe\n")
        handle.flush()
        return _run(["nft", "-c", "-f", handle.name]).returncode == 0


@contextmanager
def _network_namespace(name: str):
    """An empty namespace, so a loaded ruleset cannot reach the live one."""
    created = _run(["ip", "netns", "add", name])
    if created.returncode != 0:
        raise unittest.SkipTest(f"cannot create a network namespace: {created.stderr}")
    try:
        yield ["ip", "netns", "exec", name]
    finally:
        _run(["ip", "netns", "del", name])


def _fragments() -> list[str]:
    suffix = "/" + FRAGMENT
    return sorted(
        path
        for path in iter_project_files(extensions=(".j2",), exclude_tests=True)
        if path.endswith(suffix)
    )


def _render(path: str, mode: str = "swarm") -> str:
    environment = jinja2.Environment(
        keep_trailing_newline=True,
        trim_blocks=True,
        autoescape=jinja2.select_autoescape(default_for_string=False),
    )
    return environment.from_string(read_text(path)).render(
        DEPLOYMENT_MODE=mode, **COMMON
    )


class TestFirewallFragments(unittest.TestCase):
    def test_at_least_one_fragment_exists(self) -> None:
        """Otherwise every assertion below passes over an empty list."""
        self.assertTrue(_fragments(), "no role declares templates/nftables.conf.j2")

    def test_each_fragment_replaces_its_table(self) -> None:
        for path in _fragments():
            rel = Path(path).relative_to(PROJECT_ROOT).as_posix()
            with self.subTest(fragment=rel):
                lines = [
                    line.strip() for line in _render(path).splitlines() if line.strip()
                ]
                declarations = [
                    line for line in lines if re.match(r"^(delete )?table inet ", line)
                ]
                self.assertEqual(
                    declarations[:3],
                    [
                        f"table inet {TABLE}",
                        f"delete table inet {TABLE}",
                        f"table inet {TABLE} {{",
                    ],
                    f"{rel} must open with the create, delete and declare idiom, "
                    "or nft merges the new rules into the old table",
                )

    def test_each_fragment_parses(self) -> None:
        if not _nft_can_check():
            self.skipTest("nft -c needs CAP_NET_ADMIN, which this runner lacks")
        nft = shutil.which("nft")
        for path in _fragments():
            rel = Path(path).relative_to(PROJECT_ROOT).as_posix()
            for mode in MODES:
                with self.subTest(fragment=rel, mode=mode):
                    with tempfile.NamedTemporaryFile("w", suffix=".conf") as handle:
                        handle.write(_render(path, mode))
                        handle.flush()
                        result = subprocess.run(
                            [nft, "-c", "-f", handle.name],
                            capture_output=True,
                            text=True,
                            check=False,
                        )
                    self.assertEqual(
                        result.returncode,
                        0,
                        f"{rel} is not valid nftables in {mode}:\n{result.stderr}",
                    )

    def test_each_fragment_loads_and_reloads(self) -> None:
        """A dry run cannot show what happens on the second deploy.

        Loading the same fragment twice is what proves the idiom: the second
        pass meets the table the first one left and must replace it, which is
        exactly the case ``nft -c`` never sees.
        """
        if not _nft_can_check() or shutil.which("ip") is None:
            self.skipTest("loading a ruleset needs nft with CAP_NET_ADMIN and ip")
        for index, path in enumerate(_fragments()):
            rel = Path(path).relative_to(PROJECT_ROOT).as_posix()
            for mode in MODES:
                with self.subTest(fragment=rel, mode=mode):
                    with (
                        tempfile.NamedTemporaryFile("w", suffix=".conf") as handle,
                        _network_namespace(f"nftprobe{os.getpid()}x{index}") as netns,
                    ):
                        handle.write(_render(path, mode))
                        handle.flush()
                        listings = []
                        for attempt in ("first", "second"):
                            loaded = _run([*netns, "nft", "-f", handle.name])
                            self.assertEqual(
                                loaded.returncode,
                                0,
                                f"{rel} failed to load on the {attempt} pass "
                                f"in {mode}:\n{loaded.stderr}",
                            )
                            listed = _run(
                                [*netns, "nft", "list", "table", "inet", TABLE]
                            )
                            self.assertEqual(
                                listed.returncode,
                                0,
                                f"{rel} loaded but left no table on the {attempt} "
                                f"pass in {mode}:\n{listed.stderr}",
                            )
                            listings.append(listed.stdout)
                    self.assertIn(f"table inet {TABLE}", listings[0])
                    self.assertEqual(
                        listings[0],
                        listings[1],
                        f"{rel} grew on the second pass in {mode}, so it merges "
                        "into the table it found instead of replacing it",
                    )


if __name__ == "__main__":
    unittest.main()