Reference: Python API

Auto-generated Python API documentation powered by mkdocstrings.

---

Core Pipeline & Orchestration

nse.core.pipeline

Test pipeline: orchestrates a single NSE test run.

The oracle contract

A firewall test is a negative assertion ("this packet did not get through"), and a negative assertion is worthless unless the instrument is known to be working. So every run carries its own positive control: a canary packet is injected before the test packets and again after them, and the run is only reported as a result if the kernel trace for both canaries was observed.

• The pre-canary replaces the old readiness probe, which signalled that this process had scheduled its read loop — not that nft monitor trace had actually subscribed to the kernel. Packets injected in that window were lost.
• The post-canary proves the monitor was still watching when the last test packet went in. It is what catches a trace deadline expiring mid-run, which used to silently truncate the verdict stream.

If either canary is not observed the pipeline emits an error event and the caller fails the test. It is never possible for this pipeline to report a clean verdict list that it did not actually measure.

run_test_pipeline(request, controller=None, queue=None, run=None) async

Full test lifecycle coroutine.

Executes a test request in isolated Linux network namespaces, injecting packets and collecting kernel nftables trace events.

The returned list never contains canary events: they are the pipeline's own instrument check, not part of the user's test. If a canary is missed, the list contains an error event instead of a verdict stream, so a caller cannot mistake "the oracle saw nothing" for "nothing happened".

Source code in nse/core/pipeline.py

async def run_test_pipeline(
    request: TestRequest,
    controller: NetnsController | None = None,
    queue: asyncio.Queue[TraceEvent | None] | None = None,
    run: TestRun | None = None,
) -> list[TraceEvent]:
    """
    Full test lifecycle coroutine.

    Executes a test request in isolated Linux network namespaces, injecting
    packets and collecting kernel nftables trace events.

    The returned list never contains canary events: they are the pipeline's own
    instrument check, not part of the user's test. If a canary is missed, the
    list contains an ``error`` event instead of a verdict stream, so a caller
    cannot mistake "the oracle saw nothing" for "nothing happened".
    """
    if controller is None:
        controller = NetnsController()

    if run is None:
        test_id = uuid.uuid4().hex[:12]
        netns_name = f"nse_{test_id}"
        run = TestRun(test_id=test_id, netns_name=netns_name, request=request)
        if queue is not None:
            run.event_queue = queue

    event_queue = run.event_queue
    collected_events: list[TraceEvent] = []
    canary_trace_ids: set[str] = set()

    async def emit_event(evt: TraceEvent | None) -> None:
        if evt is not None:
            collected_events.append(evt)
        await event_queue.put(evt)

    async def emit_oracle_error(message: str) -> None:
        """Report a failure of the *instrument*, not of the ruleset under test."""
        logger.error("[%s] Oracle error: %s", run.test_id, message)
        run.status = "error"
        await emit_event(
            TraceEvent(
                type="error",
                trace_id=run.test_id,
                verdict="ERROR",
                raw_message=f"oracle: {message}",
            )
        )

    run.status = "running"
    req = request
    names = derive_names(run.test_id)
    netns = names.netns
    router_ns = names.router_ns
    server_ns = names.server_ns
    veth_host = names.veth_host
    veth_router_host = names.veth_router_host
    veth_router_server = names.veth_router_server
    veth_server = names.veth_server

    is_gateway = req.topology == TopologyType.GATEWAY
    listeners = []
    harvester = TraceHarvester()
    engine = RuleEngine(use_nsenter=controller.use_nsenter)
    injector = ScapyInjector(use_nsenter=controller.use_nsenter)

    loop = asyncio.get_running_loop()
    target_netns = router_ns if is_gateway else netns
    veth_peer_target = veth_router_host if is_gateway else _VETH_PEER
    # The "host" end of the injection link is in the root namespace for both
    # topologies; the router namespace only owns the peer end.
    host_netns: str | None = None

    try:
        # ------------------------------------------------------------------
        # 1. Setup Network Topology
        # ------------------------------------------------------------------
        if is_gateway:
            logger.info("[%s] Setting up Gateway topology: netns=%s", run.test_id, router_ns)
            await loop.run_in_executor(
                None,
                controller.create_gateway_topology,
                router_ns,
                server_ns,
                veth_host,
                veth_router_host,
                veth_router_server,
                veth_server,
            )
        else:
            logger.info("[%s] Setting up Simple topology: netns=%s", run.test_id, netns)
            await loop.run_in_executor(None, controller.create_netns, netns)
            await loop.run_in_executor(
                None,
                controller.create_veth_pair,
                netns,
                veth_host,
                _VETH_PEER,
            )

        # ------------------------------------------------------------------
        # 2. Spawning background mock listeners inside server/sandbox namespace
        # ------------------------------------------------------------------
        listener_netns = server_ns if is_gateway else netns
        logger.info(
            "[%s] Spawning background mock listeners inside %s", run.test_id, listener_netns
        )

        for pkt in req.packets:
            if pkt.dst_port:
                proc = start_mock_listener(
                    netns_name=listener_netns,
                    proto=pkt.protocol,
                    port=pkt.dst_port,
                    use_nsenter=controller.use_nsenter,
                )
                listeners.append({"proto": pkt.protocol, "port": pkt.dst_port, "proc": proc})

        # ------------------------------------------------------------------
        # 3. Load nftables Rules
        # ------------------------------------------------------------------
        logger.info("[%s] Loading nftables rules into netns %s", run.test_id, target_netns)
        await loop.run_in_executor(None, engine.load, req.rules, target_netns)

        # ------------------------------------------------------------------
        # 4. Start nft monitor trace
        # ------------------------------------------------------------------
        logger.info("[%s] Starting nft monitor trace", run.test_id)

        def on_trace_event(evt: TraceEvent) -> None:
            collected_events.append(evt)

        initial_timeout = max(
            _TRACE_IDLE_TIMEOUT,
            2.0 + _INJECT_SETTLE * len(req.packets) + _CANARY_ATTEMPTS * _CANARY_INTERVAL,
        )
        await harvester.start(
            netns_name=target_netns,
            queue=event_queue,
            timeout=initial_timeout,
            use_nsenter=controller.use_nsenter,
            on_event=on_trace_event,
        )
        await harvester.wait_ready(timeout=2.0)

        # ------------------------------------------------------------------
        # 5. Readiness canary — prove the oracle can see before trusting it
        # ------------------------------------------------------------------
        first_packet = req.packets[0]
        canary = _canary_spec(dst_ip=first_packet.dst_ip, src_ip=first_packet.src_ip)
        probe_kwargs = {
            "harvester": harvester,
            "injector": injector,
            "spec": canary,
            "target_netns": target_netns,
            "veth_host": veth_host,
            "veth_peer": veth_peer_target,
            "host_netns": host_netns,
        }

        logger.info("[%s] Probing the oracle with a readiness canary", run.test_id)
        if not await _probe_oracle(label="readiness", **probe_kwargs):  # type: ignore[arg-type]
            await emit_oracle_error(
                "readiness canary was never observed - `nft monitor trace` is not "
                "reporting kernel events, so no verdict from this run would be "
                "evidence of anything"
            )
            await emit_event(None)
            return _without_canaries(collected_events, canary_trace_ids)
        canary_trace_ids.update(harvester.seen_trace_ids)

        # ------------------------------------------------------------------
        # 6. Inject Packet Sequence (Per-packet ordering-based injection)
        # ------------------------------------------------------------------
        for idx, pkt in enumerate(req.packets, start=1):
            logger.info(
                "[%s] Injecting packet %d/%d (%s -> %s:%s)",
                run.test_id,
                idx,
                len(req.packets),
                pkt.src_ip,
                pkt.dst_ip,
                pkt.dst_port,
            )
            harvester.extend_deadline(_TRACE_IDLE_TIMEOUT)
            await loop.run_in_executor(
                None,
                injector.inject,
                pkt,
                target_netns,
                veth_host,
                veth_peer_target,
                host_netns,
            )
            # Short per-packet delay to let kernel trace process the verdict deterministically
            await asyncio.sleep(_INJECT_SETTLE)

        await asyncio.sleep(_DRAIN_DELAY)

        # ------------------------------------------------------------------
        # 7. Liveness canary — prove the oracle was STILL seeing at the end
        # ------------------------------------------------------------------
        logger.info("[%s] Probing the oracle with a liveness canary", run.test_id)
        before_liveness = set(harvester.seen_trace_ids)
        if not await _probe_oracle(label="liveness", **probe_kwargs):  # type: ignore[arg-type]
            await emit_oracle_error(
                "liveness canary was never observed - the trace monitor stopped "
                "reporting before the run finished, so the verdict stream is "
                "truncated by an unknown amount"
            )
        canary_trace_ids.update(set(harvester.seen_trace_ids) - before_liveness)

        # ------------------------------------------------------------------
        # 8. Collect conntrack entries & finish
        # ------------------------------------------------------------------
        ct_entries = await loop.run_in_executor(
            None, read_conntrack_table, target_netns, controller.use_nsenter
        )
        for ct in ct_entries:
            await emit_event(
                TraceEvent(
                    type="conntrack",
                    trace_id=run.test_id,
                    rule_text=f"state={ct['state']} proto={ct['proto']} {ct['src']}:{ct['sport']} -> {ct['dst']}:{ct['dport']}",
                )
            )

        # ------------------------------------------------------------------
        # 9. Close the monitor and report anything that made it untrustworthy
        # ------------------------------------------------------------------
        await harvester.aclose()
        for problem in harvester.health_errors():
            await emit_oracle_error(problem)

        if run.status != "error":
            run.status = "done"
            logger.info("[%s] Test pipeline finished successfully", run.test_id)

    except RuleValidationError as exc:
        logger.warning("[%s] Rule validation error: %s", run.test_id, exc)
        run.status = "error"
        await emit_event(
            TraceEvent(
                type="error",
                trace_id=run.test_id,
                verdict="ERROR",
                raw_message=str(exc.errors),
            )
        )
        await emit_event(None)
        await harvester.aclose()

    except Exception as exc:
        logger.exception("[%s] Pipeline error", run.test_id)
        run.status = "error"
        await emit_event(
            TraceEvent(
                type="error",
                trace_id=run.test_id,
                verdict="ERROR",
                raw_message=str(exc),
            )
        )
        await emit_event(None)
        await harvester.aclose()

    finally:
        # ------------------------------------------------------------------
        # Teardown mock listeners
        # ------------------------------------------------------------------
        logger.info("[%s] Tearing down mock listeners", run.test_id)
        for listener in listeners:
            listener_proc: subprocess.Popen[str] = cast(subprocess.Popen[str], listener["proc"])
            try:
                listener_proc.terminate()
                listener_proc.wait(timeout=0.5)
            except (OSError, subprocess.SubprocessError):
                with contextlib.suppress(OSError, subprocess.SubprocessError):
                    listener_proc.kill()

        # ------------------------------------------------------------------
        # Teardown namespaces & host interfaces (blocking → executor)
        # ------------------------------------------------------------------
        logger.info("[%s] Tearing down network topology", run.test_id)

        if is_gateway:
            await loop.run_in_executor(None, controller.destroy_netns, router_ns)
            await loop.run_in_executor(None, controller.destroy_netns, server_ns)
        else:
            await loop.run_in_executor(None, controller.destroy_netns, netns)

        # Explicitly clean up host veth
        await loop.run_in_executor(
            None,
            lambda: subprocess.run(
                [resolve("ip"), "link", "del", veth_host],
                capture_output=True,
                check=False,
            ),
        )

    return _without_canaries(collected_events, canary_trace_ids)

---

Netns Controller

nse.core.netns_controller

NetnsController: manages ephemeral Linux network namespaces.

All subprocess calls use iproute2 (ip) and must be run as root.

NamespaceSandbox

Represents an active isolated network namespace sandbox. Provides utility methods to execute commands and inject packets inside the sandbox context.

Source code in nse/core/netns_controller.py

class NamespaceSandbox:
    """
    Represents an active isolated network namespace sandbox.
    Provides utility methods to execute commands and inject packets inside the sandbox context.
    """

    def __init__(self, controller: NetnsController, name: str) -> None:
        self.controller = controller
        self.name = name
        # We derive interface names from the sandbox name
        self.ext_iface = f"vhr-{name[:8]}"
        self.peer_iface = f"vrh-{name[:8]}"

    async def exec(self, command: str) -> subprocess.CompletedProcess[bytes]:
        """
        Execute a shell command inside the network namespace context asynchronously.
        """
        cmd = self.controller.exec_prefix(self.name) + command.split()
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await proc.communicate()
        returncode = proc.returncode if proc.returncode is not None else 1
        if returncode != 0:
            raise subprocess.CalledProcessError(
                returncode,
                cmd,
                output=stdout,
                stderr=stderr,
            )
        return subprocess.CompletedProcess(
            args=cmd,
            returncode=returncode,
            stdout=stdout,
            stderr=stderr,
        )

    async def inject_packet(
        self,
        protocol: str,
        dst_port: int,
        dst_ip: str,
        src_ip: str | None = None,
    ) -> None:
        """
        Inject a layer 3/4 packet on the host side of the veth link targeting this namespace.
        """
        from nse.core.scapy_injector import ScapyInjector
        from nse.models.test_request import PacketSpec

        injector = ScapyInjector()
        packet = PacketSpec(
            protocol=cast(Any, protocol),
            dst_port=dst_port,
            dst_ip=dst_ip,
            src_ip=src_ip or ("10.0.1.1" if "." in dst_ip else "fd00:1::1"),
        )

        loop = asyncio.get_running_loop()
        # Sniffing/injection scapy operations can be blocking, run in executor
        await loop.run_in_executor(
            None,
            injector.inject,
            packet,
            self.name,
            self.ext_iface,
            self.peer_iface,
        )

exec(command) async

Execute a shell command inside the network namespace context asynchronously.

Source code in nse/core/netns_controller.py

async def exec(self, command: str) -> subprocess.CompletedProcess[bytes]:
    """
    Execute a shell command inside the network namespace context asynchronously.
    """
    cmd = self.controller.exec_prefix(self.name) + command.split()
    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    returncode = proc.returncode if proc.returncode is not None else 1
    if returncode != 0:
        raise subprocess.CalledProcessError(
            returncode,
            cmd,
            output=stdout,
            stderr=stderr,
        )
    return subprocess.CompletedProcess(
        args=cmd,
        returncode=returncode,
        stdout=stdout,
        stderr=stderr,
    )

inject_packet(protocol, dst_port, dst_ip, src_ip=None) async

Inject a layer 3/4 packet on the host side of the veth link targeting this namespace.

Source code in nse/core/netns_controller.py

async def inject_packet(
    self,
    protocol: str,
    dst_port: int,
    dst_ip: str,
    src_ip: str | None = None,
) -> None:
    """
    Inject a layer 3/4 packet on the host side of the veth link targeting this namespace.
    """
    from nse.core.scapy_injector import ScapyInjector
    from nse.models.test_request import PacketSpec

    injector = ScapyInjector()
    packet = PacketSpec(
        protocol=cast(Any, protocol),
        dst_port=dst_port,
        dst_ip=dst_ip,
        src_ip=src_ip or ("10.0.1.1" if "." in dst_ip else "fd00:1::1"),
    )

    loop = asyncio.get_running_loop()
    # Sniffing/injection scapy operations can be blocking, run in executor
    await loop.run_in_executor(
        None,
        injector.inject,
        packet,
        self.name,
        self.ext_iface,
        self.peer_iface,
    )

NetnsController

Central orchestrator for network namespace lifecycle and test management.

Source code in nse/core/netns_controller.py

class NetnsController:
    """
    Central orchestrator for network namespace lifecycle and test management.
    """

    def __init__(self, use_nsenter: bool | None = None) -> None:
        self._active_ns: set[str] = set()  # namespace names
        if use_nsenter is None:
            self.use_nsenter = is_in_container()
        else:
            self.use_nsenter = use_nsenter
        self.startup_sweep()

    def startup_sweep(self) -> None:
        """Clean up orphan namespaces and veth pairs left behind by previous crashes."""
        try:
            res = subprocess.run(
                [resolve("ip"), "netns", "list"],
                capture_output=True,
                text=True,
                check=False,
                timeout=5.0,
            )
            if res.returncode == 0:
                for line in res.stdout.splitlines():
                    ns_name = line.split()[0] if line.split() else ""
                    if ns_name.startswith(NETNS_SWEEP_PREFIXES):
                        logger.info("Startup sweep: removing orphan netns %s", ns_name)
                        subprocess.run(
                            [resolve("ip"), "netns", "del", ns_name],
                            capture_output=True,
                            check=False,
                            timeout=5.0,
                        )
        except Exception as exc:
            logger.debug("Startup sweep netns list failed: %s", exc)

        try:
            res = subprocess.run(
                [resolve("ip"), "link", "show"],
                capture_output=True,
                text=True,
                check=False,
                timeout=5.0,
            )
            if res.returncode == 0:
                for line in res.stdout.splitlines():
                    parts = line.split(":")
                    if len(parts) >= 2:
                        iface = parts[1].strip().split("@")[0]
                        if iface.startswith(VETH_SWEEP_PREFIXES):
                            logger.info("Startup sweep: removing orphan veth link %s", iface)
                            subprocess.run(
                                [resolve("ip"), "link", "del", iface],
                                capture_output=True,
                                check=False,
                                timeout=5.0,
                            )
        except Exception as exc:
            logger.debug("Startup sweep veth list failed: %s", exc)

    def exec_prefix(self, name: str) -> list[str]:
        if self.use_nsenter:
            return [resolve("nsenter"), f"--net=/var/run/netns/{name}", "--"]
        else:
            return [resolve("ip"), "netns", "exec", name]

    def _run_in_netns(self, name: str, cmd: list[str]) -> subprocess.CompletedProcess[str]:
        return _run(self.exec_prefix(name) + cmd)

    # ------------------------------------------------------------------
    # Namespace lifecycle
    # ------------------------------------------------------------------

    def create_netns(self, name: str) -> None:
        """Create a new network namespace. Raises on failure."""
        logger.debug("Creating netns: %s", name)
        _run([resolve("ip"), "netns", "add", name])
        self._active_ns.add(name)
        # Disable DAD inside the netns to speed up IPv6 interface readiness
        try:
            self._run_in_netns(
                name,
                [
                    resolve("sysctl"),
                    "-w",
                    "net.ipv6.conf.all.accept_dad=0",
                ],
            )
            self._run_in_netns(
                name,
                [
                    resolve("sysctl"),
                    "-w",
                    "net.ipv6.conf.default.accept_dad=0",
                ],
            )
        except subprocess.CalledProcessError as e:
            logger.warning("Could not set accept_dad sysctls inside netns %s: %s", name, e)

    def destroy_netns(self, name: str) -> None:
        """Delete a network namespace with retry backoff. Idempotent."""
        logger.debug("Destroying netns: %s", name)
        delays = [0.1, 0.5, 1.0]
        for idx, delay in enumerate(delays):
            try:
                _run([resolve("ip"), "netns", "del", name])
                break
            except subprocess.CalledProcessError as err:
                stderr = err.stderr or ""
                if "No such file or directory" in stderr or "Invalid argument" in stderr:
                    logger.debug("netns %s already gone.", name)
                    break
                if idx < len(delays) - 1:
                    time.sleep(delay)
                else:
                    logger.warning("Failed to destroy netns %s after 3 attempts: %s", name, err)
            except subprocess.TimeoutExpired:
                logger.warning("Timeout destroying netns %s", name)
                break
        self._active_ns.discard(name)

    @contextlib.asynccontextmanager
    async def create_namespace(
        self,
        name: str,
        host_ip: str | list[str] = "10.0.1.1/24",
        peer_ip: str | list[str] = "10.0.1.2/24",
    ) -> AsyncIterator[NamespaceSandbox]:
        """
        Context manager to safely construct and teardown an isolated namespace.
        """
        sandbox = NamespaceSandbox(self, name)

        # Setup
        self.create_netns(name)

        # Bring loopback interface up
        self._run_in_netns(name, [resolve("ip"), "link", "set", "lo", "up"])

        # Setup links
        self.create_veth_pair(
            netns_name=name,
            veth_host=sandbox.ext_iface,
            veth_peer=sandbox.peer_iface,
            host_ip=host_ip,
            peer_ip=peer_ip,
        )

        try:
            yield sandbox
        finally:
            # Cleanup links
            with contextlib.suppress(subprocess.CalledProcessError, OSError):
                _run([resolve("ip"), "link", "del", sandbox.ext_iface])

            # Cleanup netns
            self.destroy_netns(name)

    def create_veth_pair(
        self,
        netns_name: str,
        veth_host: str,
        veth_peer: str,
        host_ip: str | list[str] = "10.0.0.1/24",
        peer_ip: str | list[str] = "10.0.0.2/24",
    ) -> None:
        """
        Create a veth pair, move one end into *netns_name*, and assign IPs.
        """
        logger.debug(
            "Creating veth pair %s <-> %s in netns %s",
            veth_host,
            veth_peer,
            netns_name,
        )

        # Parse host and peer IPs (which could be single strings or list of strings)
        host_ips = [host_ip] if isinstance(host_ip, str) else list(host_ip)
        peer_ips = [peer_ip] if isinstance(peer_ip, str) else list(peer_ip)

        v4_host, v6_host = None, None
        v4_peer, v6_peer = None, None

        for ip in host_ips:
            if ":" in ip:
                v6_host = ip
            else:
                v4_host = ip
        for ip in peer_ips:
            if ":" in ip:
                v6_peer = ip
            else:
                v4_peer = ip

        # Inject defaults if missing to support hybrid/both tests easily
        if not v4_host:
            v4_host = "10.0.0.1/24"
        if not v4_peer:
            v4_peer = "10.0.0.2/24"
        if not v6_host:
            v6_host = "fd00::1/64"
        if not v6_peer:
            v6_peer = "fd00::2/64"

        # Create veth pair in the root namespace
        _run([resolve("ip"), "link", "add", veth_host, "type", "veth", "peer", "name", veth_peer])
        # Move the peer end into the target namespace
        _run([resolve("ip"), "link", "set", veth_peer, "netns", netns_name])

        # --- Host side ---
        if v4_host:
            _run([resolve("ip"), "addr", "add", v4_host, "dev", veth_host])
        if v6_host:
            _run([resolve("ip"), "addr", "add", v6_host, "dev", veth_host])
        _run([resolve("ip"), "link", "set", veth_host, "up"])

        # --- Namespace side ---
        if v4_peer:
            self._run_in_netns(
                netns_name,
                [
                    resolve("ip"),
                    "addr",
                    "add",
                    v4_peer,
                    "dev",
                    veth_peer,
                ],
            )
        if v6_peer:
            self._run_in_netns(
                netns_name,
                [
                    resolve("ip"),
                    "addr",
                    "add",
                    v6_peer,
                    "dev",
                    veth_peer,
                ],
            )
        self._run_in_netns(netns_name, [resolve("ip"), "link", "set", veth_peer, "up"])
        self._run_in_netns(netns_name, [resolve("ip"), "link", "set", "lo", "up"])

    def create_gateway_topology(
        self,
        router_ns: str,
        server_ns: str,
        veth_host: str,
        veth_router_host: str,
        veth_router_server: str,
        veth_server: str,
        host_v4: str = "10.0.1.1/24",
        router_host_v4: str = "10.0.1.2/24",
        router_server_v4: str = "10.0.2.1/24",
        server_v4: str = "10.0.2.2/24",
        host_v6: str = "fd00:1::1/64",
        router_host_v6: str = "fd00:1::2/64",
        router_server_v6: str = "fd00:2::1/64",
        server_v6: str = "fd00:2::2/64",
    ) -> None:
        """
        Create router and server namespaces, build double veth links,
        enable IPv4/IPv6 forwarding inside the router, and add transit routes.
        """
        logger.info("Setting up gateway topology: %s <-> %s", router_ns, server_ns)
        # Create namespaces
        self.create_netns(router_ns)
        self.create_netns(server_ns)

        # Enable IPv4/IPv6 forwarding on router namespace
        self._run_in_netns(router_ns, [resolve("sysctl"), "-w", "net.ipv4.ip_forward=1"])
        self._run_in_netns(
            router_ns,
            [
                resolve("sysctl"),
                "-w",
                "net.ipv6.conf.all.forwarding=1",
            ],
        )

        # 1. Create Host <-> Router veth pair
        _run(
            [
                resolve("ip"),
                "link",
                "add",
                veth_host,
                "type",
                "veth",
                "peer",
                "name",
                veth_router_host,
            ]
        )
        _run([resolve("ip"), "link", "set", veth_router_host, "netns", router_ns])

        _run([resolve("ip"), "addr", "add", host_v4, "dev", veth_host])
        _run([resolve("ip"), "addr", "add", host_v6, "dev", veth_host])
        _run([resolve("ip"), "link", "set", veth_host, "up"])

        self._run_in_netns(
            router_ns,
            [
                resolve("ip"),
                "addr",
                "add",
                router_host_v4,
                "dev",
                veth_router_host,
            ],
        )
        self._run_in_netns(
            router_ns,
            [
                resolve("ip"),
                "addr",
                "add",
                router_host_v6,
                "dev",
                veth_router_host,
            ],
        )
        self._run_in_netns(
            router_ns,
            [
                resolve("ip"),
                "link",
                "set",
                veth_router_host,
                "up",
            ],
        )

        # 2. Create Router <-> Server veth pair
        self._run_in_netns(
            router_ns,
            [
                resolve("ip"),
                "link",
                "add",
                veth_router_server,
                "type",
                "veth",
                "peer",
                "name",
                veth_server,
            ],
        )
        self._run_in_netns(
            router_ns,
            [
                resolve("ip"),
                "link",
                "set",
                veth_server,
                "netns",
                server_ns,
            ],
        )

        self._run_in_netns(
            router_ns,
            [
                resolve("ip"),
                "addr",
                "add",
                router_server_v4,
                "dev",
                veth_router_server,
            ],
        )
        self._run_in_netns(
            router_ns,
            [
                resolve("ip"),
                "addr",
                "add",
                router_server_v6,
                "dev",
                veth_router_server,
            ],
        )
        self._run_in_netns(
            router_ns,
            [
                resolve("ip"),
                "link",
                "set",
                veth_router_server,
                "up",
            ],
        )

        self._run_in_netns(
            server_ns,
            [
                resolve("ip"),
                "addr",
                "add",
                server_v4,
                "dev",
                veth_server,
            ],
        )
        self._run_in_netns(
            server_ns,
            [
                resolve("ip"),
                "addr",
                "add",
                server_v6,
                "dev",
                veth_server,
            ],
        )
        self._run_in_netns(server_ns, [resolve("ip"), "link", "set", veth_server, "up"])

        # Bring up loopbacks
        self._run_in_netns(router_ns, [resolve("ip"), "link", "set", "lo", "up"])
        self._run_in_netns(server_ns, [resolve("ip"), "link", "set", "lo", "up"])

        # 3. Setup transit routing
        # Route on Host: Server subnet via Router host IP
        host_rt_via = router_host_v4.split("/")[0]
        host_rt_via6 = router_host_v6.split("/")[0]
        _run([resolve("ip"), "route", "add", "10.0.2.0/24", "via", host_rt_via, "dev", veth_host])
        _run(
            [
                resolve("ip"),
                "-6",
                "route",
                "add",
                "fd00:2::/64",
                "via",
                host_rt_via6,
                "dev",
                veth_host,
            ]
        )

        # Route on Server: Host subnet via Router server IP (default route is cleanest)
        srv_rt_via = router_server_v4.split("/")[0]
        srv_rt_via6 = router_server_v6.split("/")[0]
        self._run_in_netns(
            server_ns,
            [
                resolve("ip"),
                "route",
                "add",
                "default",
                "via",
                srv_rt_via,
                "dev",
                veth_server,
            ],
        )
        self._run_in_netns(
            server_ns,
            [
                resolve("ip"),
                "-6",
                "route",
                "add",
                "default",
                "via",
                srv_rt_via6,
                "dev",
                veth_server,
            ],
        )

    def cleanup_all(self) -> None:
        """
        Destroy all known namespaces.  Called on SIGINT/SIGTERM.
        Safe to call multiple times.
        """
        logger.info("Cleaning up %d namespace(s)…", len(self._active_ns))
        for name in list(self._active_ns):
            self.destroy_netns(name)

cleanup_all()

Destroy all known namespaces. Called on SIGINT/SIGTERM. Safe to call multiple times.

Source code in nse/core/netns_controller.py

def cleanup_all(self) -> None:
    """
    Destroy all known namespaces.  Called on SIGINT/SIGTERM.
    Safe to call multiple times.
    """
    logger.info("Cleaning up %d namespace(s)…", len(self._active_ns))
    for name in list(self._active_ns):
        self.destroy_netns(name)

create_gateway_topology(router_ns, server_ns, veth_host, veth_router_host, veth_router_server, veth_server, host_v4='10.0.1.1/24', router_host_v4='10.0.1.2/24', router_server_v4='10.0.2.1/24', server_v4='10.0.2.2/24', host_v6='fd00:1::1/64', router_host_v6='fd00:1::2/64', router_server_v6='fd00:2::1/64', server_v6='fd00:2::2/64')

Create router and server namespaces, build double veth links, enable IPv4/IPv6 forwarding inside the router, and add transit routes.

Source code in nse/core/netns_controller.py

def create_gateway_topology(
    self,
    router_ns: str,
    server_ns: str,
    veth_host: str,
    veth_router_host: str,
    veth_router_server: str,
    veth_server: str,
    host_v4: str = "10.0.1.1/24",
    router_host_v4: str = "10.0.1.2/24",
    router_server_v4: str = "10.0.2.1/24",
    server_v4: str = "10.0.2.2/24",
    host_v6: str = "fd00:1::1/64",
    router_host_v6: str = "fd00:1::2/64",
    router_server_v6: str = "fd00:2::1/64",
    server_v6: str = "fd00:2::2/64",
) -> None:
    """
    Create router and server namespaces, build double veth links,
    enable IPv4/IPv6 forwarding inside the router, and add transit routes.
    """
    logger.info("Setting up gateway topology: %s <-> %s", router_ns, server_ns)
    # Create namespaces
    self.create_netns(router_ns)
    self.create_netns(server_ns)

    # Enable IPv4/IPv6 forwarding on router namespace
    self._run_in_netns(router_ns, [resolve("sysctl"), "-w", "net.ipv4.ip_forward=1"])
    self._run_in_netns(
        router_ns,
        [
            resolve("sysctl"),
            "-w",
            "net.ipv6.conf.all.forwarding=1",
        ],
    )

    # 1. Create Host <-> Router veth pair
    _run(
        [
            resolve("ip"),
            "link",
            "add",
            veth_host,
            "type",
            "veth",
            "peer",
            "name",
            veth_router_host,
        ]
    )
    _run([resolve("ip"), "link", "set", veth_router_host, "netns", router_ns])

    _run([resolve("ip"), "addr", "add", host_v4, "dev", veth_host])
    _run([resolve("ip"), "addr", "add", host_v6, "dev", veth_host])
    _run([resolve("ip"), "link", "set", veth_host, "up"])

    self._run_in_netns(
        router_ns,
        [
            resolve("ip"),
            "addr",
            "add",
            router_host_v4,
            "dev",
            veth_router_host,
        ],
    )
    self._run_in_netns(
        router_ns,
        [
            resolve("ip"),
            "addr",
            "add",
            router_host_v6,
            "dev",
            veth_router_host,
        ],
    )
    self._run_in_netns(
        router_ns,
        [
            resolve("ip"),
            "link",
            "set",
            veth_router_host,
            "up",
        ],
    )

    # 2. Create Router <-> Server veth pair
    self._run_in_netns(
        router_ns,
        [
            resolve("ip"),
            "link",
            "add",
            veth_router_server,
            "type",
            "veth",
            "peer",
            "name",
            veth_server,
        ],
    )
    self._run_in_netns(
        router_ns,
        [
            resolve("ip"),
            "link",
            "set",
            veth_server,
            "netns",
            server_ns,
        ],
    )

    self._run_in_netns(
        router_ns,
        [
            resolve("ip"),
            "addr",
            "add",
            router_server_v4,
            "dev",
            veth_router_server,
        ],
    )
    self._run_in_netns(
        router_ns,
        [
            resolve("ip"),
            "addr",
            "add",
            router_server_v6,
            "dev",
            veth_router_server,
        ],
    )
    self._run_in_netns(
        router_ns,
        [
            resolve("ip"),
            "link",
            "set",
            veth_router_server,
            "up",
        ],
    )

    self._run_in_netns(
        server_ns,
        [
            resolve("ip"),
            "addr",
            "add",
            server_v4,
            "dev",
            veth_server,
        ],
    )
    self._run_in_netns(
        server_ns,
        [
            resolve("ip"),
            "addr",
            "add",
            server_v6,
            "dev",
            veth_server,
        ],
    )
    self._run_in_netns(server_ns, [resolve("ip"), "link", "set", veth_server, "up"])

    # Bring up loopbacks
    self._run_in_netns(router_ns, [resolve("ip"), "link", "set", "lo", "up"])
    self._run_in_netns(server_ns, [resolve("ip"), "link", "set", "lo", "up"])

    # 3. Setup transit routing
    # Route on Host: Server subnet via Router host IP
    host_rt_via = router_host_v4.split("/")[0]
    host_rt_via6 = router_host_v6.split("/")[0]
    _run([resolve("ip"), "route", "add", "10.0.2.0/24", "via", host_rt_via, "dev", veth_host])
    _run(
        [
            resolve("ip"),
            "-6",
            "route",
            "add",
            "fd00:2::/64",
            "via",
            host_rt_via6,
            "dev",
            veth_host,
        ]
    )

    # Route on Server: Host subnet via Router server IP (default route is cleanest)
    srv_rt_via = router_server_v4.split("/")[0]
    srv_rt_via6 = router_server_v6.split("/")[0]
    self._run_in_netns(
        server_ns,
        [
            resolve("ip"),
            "route",
            "add",
            "default",
            "via",
            srv_rt_via,
            "dev",
            veth_server,
        ],
    )
    self._run_in_netns(
        server_ns,
        [
            resolve("ip"),
            "-6",
            "route",
            "add",
            "default",
            "via",
            srv_rt_via6,
            "dev",
            veth_server,
        ],
    )

create_namespace(name, host_ip='10.0.1.1/24', peer_ip='10.0.1.2/24') async

Context manager to safely construct and teardown an isolated namespace.

Source code in nse/core/netns_controller.py

@contextlib.asynccontextmanager
async def create_namespace(
    self,
    name: str,
    host_ip: str | list[str] = "10.0.1.1/24",
    peer_ip: str | list[str] = "10.0.1.2/24",
) -> AsyncIterator[NamespaceSandbox]:
    """
    Context manager to safely construct and teardown an isolated namespace.
    """
    sandbox = NamespaceSandbox(self, name)

    # Setup
    self.create_netns(name)

    # Bring loopback interface up
    self._run_in_netns(name, [resolve("ip"), "link", "set", "lo", "up"])

    # Setup links
    self.create_veth_pair(
        netns_name=name,
        veth_host=sandbox.ext_iface,
        veth_peer=sandbox.peer_iface,
        host_ip=host_ip,
        peer_ip=peer_ip,
    )

    try:
        yield sandbox
    finally:
        # Cleanup links
        with contextlib.suppress(subprocess.CalledProcessError, OSError):
            _run([resolve("ip"), "link", "del", sandbox.ext_iface])

        # Cleanup netns
        self.destroy_netns(name)

create_netns(name)

Create a new network namespace. Raises on failure.

Source code in nse/core/netns_controller.py

def create_netns(self, name: str) -> None:
    """Create a new network namespace. Raises on failure."""
    logger.debug("Creating netns: %s", name)
    _run([resolve("ip"), "netns", "add", name])
    self._active_ns.add(name)
    # Disable DAD inside the netns to speed up IPv6 interface readiness
    try:
        self._run_in_netns(
            name,
            [
                resolve("sysctl"),
                "-w",
                "net.ipv6.conf.all.accept_dad=0",
            ],
        )
        self._run_in_netns(
            name,
            [
                resolve("sysctl"),
                "-w",
                "net.ipv6.conf.default.accept_dad=0",
            ],
        )
    except subprocess.CalledProcessError as e:
        logger.warning("Could not set accept_dad sysctls inside netns %s: %s", name, e)

create_veth_pair(netns_name, veth_host, veth_peer, host_ip='10.0.0.1/24', peer_ip='10.0.0.2/24')

Create a veth pair, move one end into netns_name, and assign IPs.

Source code in nse/core/netns_controller.py

def create_veth_pair(
    self,
    netns_name: str,
    veth_host: str,
    veth_peer: str,
    host_ip: str | list[str] = "10.0.0.1/24",
    peer_ip: str | list[str] = "10.0.0.2/24",
) -> None:
    """
    Create a veth pair, move one end into *netns_name*, and assign IPs.
    """
    logger.debug(
        "Creating veth pair %s <-> %s in netns %s",
        veth_host,
        veth_peer,
        netns_name,
    )

    # Parse host and peer IPs (which could be single strings or list of strings)
    host_ips = [host_ip] if isinstance(host_ip, str) else list(host_ip)
    peer_ips = [peer_ip] if isinstance(peer_ip, str) else list(peer_ip)

    v4_host, v6_host = None, None
    v4_peer, v6_peer = None, None

    for ip in host_ips:
        if ":" in ip:
            v6_host = ip
        else:
            v4_host = ip
    for ip in peer_ips:
        if ":" in ip:
            v6_peer = ip
        else:
            v4_peer = ip

    # Inject defaults if missing to support hybrid/both tests easily
    if not v4_host:
        v4_host = "10.0.0.1/24"
    if not v4_peer:
        v4_peer = "10.0.0.2/24"
    if not v6_host:
        v6_host = "fd00::1/64"
    if not v6_peer:
        v6_peer = "fd00::2/64"

    # Create veth pair in the root namespace
    _run([resolve("ip"), "link", "add", veth_host, "type", "veth", "peer", "name", veth_peer])
    # Move the peer end into the target namespace
    _run([resolve("ip"), "link", "set", veth_peer, "netns", netns_name])

    # --- Host side ---
    if v4_host:
        _run([resolve("ip"), "addr", "add", v4_host, "dev", veth_host])
    if v6_host:
        _run([resolve("ip"), "addr", "add", v6_host, "dev", veth_host])
    _run([resolve("ip"), "link", "set", veth_host, "up"])

    # --- Namespace side ---
    if v4_peer:
        self._run_in_netns(
            netns_name,
            [
                resolve("ip"),
                "addr",
                "add",
                v4_peer,
                "dev",
                veth_peer,
            ],
        )
    if v6_peer:
        self._run_in_netns(
            netns_name,
            [
                resolve("ip"),
                "addr",
                "add",
                v6_peer,
                "dev",
                veth_peer,
            ],
        )
    self._run_in_netns(netns_name, [resolve("ip"), "link", "set", veth_peer, "up"])
    self._run_in_netns(netns_name, [resolve("ip"), "link", "set", "lo", "up"])

destroy_netns(name)

Delete a network namespace with retry backoff. Idempotent.

Source code in nse/core/netns_controller.py

def destroy_netns(self, name: str) -> None:
    """Delete a network namespace with retry backoff. Idempotent."""
    logger.debug("Destroying netns: %s", name)
    delays = [0.1, 0.5, 1.0]
    for idx, delay in enumerate(delays):
        try:
            _run([resolve("ip"), "netns", "del", name])
            break
        except subprocess.CalledProcessError as err:
            stderr = err.stderr or ""
            if "No such file or directory" in stderr or "Invalid argument" in stderr:
                logger.debug("netns %s already gone.", name)
                break
            if idx < len(delays) - 1:
                time.sleep(delay)
            else:
                logger.warning("Failed to destroy netns %s after 3 attempts: %s", name, err)
        except subprocess.TimeoutExpired:
            logger.warning("Timeout destroying netns %s", name)
            break
    self._active_ns.discard(name)

startup_sweep()

Clean up orphan namespaces and veth pairs left behind by previous crashes.

Source code in nse/core/netns_controller.py

def startup_sweep(self) -> None:
    """Clean up orphan namespaces and veth pairs left behind by previous crashes."""
    try:
        res = subprocess.run(
            [resolve("ip"), "netns", "list"],
            capture_output=True,
            text=True,
            check=False,
            timeout=5.0,
        )
        if res.returncode == 0:
            for line in res.stdout.splitlines():
                ns_name = line.split()[0] if line.split() else ""
                if ns_name.startswith(NETNS_SWEEP_PREFIXES):
                    logger.info("Startup sweep: removing orphan netns %s", ns_name)
                    subprocess.run(
                        [resolve("ip"), "netns", "del", ns_name],
                        capture_output=True,
                        check=False,
                        timeout=5.0,
                    )
    except Exception as exc:
        logger.debug("Startup sweep netns list failed: %s", exc)

    try:
        res = subprocess.run(
            [resolve("ip"), "link", "show"],
            capture_output=True,
            text=True,
            check=False,
            timeout=5.0,
        )
        if res.returncode == 0:
            for line in res.stdout.splitlines():
                parts = line.split(":")
                if len(parts) >= 2:
                    iface = parts[1].strip().split("@")[0]
                    if iface.startswith(VETH_SWEEP_PREFIXES):
                        logger.info("Startup sweep: removing orphan veth link %s", iface)
                        subprocess.run(
                            [resolve("ip"), "link", "del", iface],
                            capture_output=True,
                            check=False,
                            timeout=5.0,
                        )
    except Exception as exc:
        logger.debug("Startup sweep veth list failed: %s", exc)

---

Rule Engine

nse.core.rule_engine

RuleEngine: inject and validate nftables rulesets inside a netns.

RuleEngine

Validates and loads nftables rules into a network namespace.

Source code in nse/core/rule_engine.py

class RuleEngine:
    """
    Validates and loads nftables rules into a network namespace.
    """

    def __init__(self, use_nsenter: bool = False) -> None:
        self.use_nsenter = use_nsenter

    def exec_prefix(self, name: str) -> list[str]:
        if self.use_nsenter:
            return [resolve("nsenter"), f"--net=/var/run/netns/{name}", "--"]
        else:
            return [resolve("ip"), "netns", "exec", name]

    def validate(self, rules: str) -> None:
        """
        Dry-run validation using ``nft --check``.
        """
        with _temp_rules_file(rules) as path:
            result = subprocess.run(
                [resolve("nft"), "--check", "-f", path],
                capture_output=True,
                text=True,
                check=False,
                timeout=10.0,
            )
            if result.returncode != 0:
                errors = _parse_nft_errors(result.stderr, path)
                raise RuleValidationError(errors)

    def load(self, rules: str, netns_name: str) -> None:
        """
        Write rules to a temp file and load them inside *netns_name*.
        Automatically prepends tracing enablement (meta trace set 1).
        """
        if not netns_name:
            raise ValueError("netns_name must be provided, refusing to inject into init_net.")

        logger.info("Loading rules into netns %s", netns_name)

        full_ruleset = f"{_TRACE_INIT_RULESET}\n{rules}"

        with _temp_rules_file(full_ruleset) as path:
            cmd = [*self.exec_prefix(netns_name), resolve("nft"), "-f", path]
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                check=False,
                timeout=30.0,
            )
            if result.returncode != 0:
                errors = _parse_nft_errors(result.stderr, path)
                raise RuleValidationError(errors)

    def flush(self, netns_name: str) -> None:
        """Remove all nftables rules from a namespace (safe cleanup)."""
        cmd = [*self.exec_prefix(netns_name), resolve("nft"), "flush", "ruleset"]
        subprocess.run(
            cmd,
            capture_output=True,
            check=False,
            timeout=10.0,
        )

flush(netns_name)

Remove all nftables rules from a namespace (safe cleanup).

Source code in nse/core/rule_engine.py

def flush(self, netns_name: str) -> None:
    """Remove all nftables rules from a namespace (safe cleanup)."""
    cmd = [*self.exec_prefix(netns_name), resolve("nft"), "flush", "ruleset"]
    subprocess.run(
        cmd,
        capture_output=True,
        check=False,
        timeout=10.0,
    )

load(rules, netns_name)

Write rules to a temp file and load them inside netns_name. Automatically prepends tracing enablement (meta trace set 1).

Source code in nse/core/rule_engine.py

def load(self, rules: str, netns_name: str) -> None:
    """
    Write rules to a temp file and load them inside *netns_name*.
    Automatically prepends tracing enablement (meta trace set 1).
    """
    if not netns_name:
        raise ValueError("netns_name must be provided, refusing to inject into init_net.")

    logger.info("Loading rules into netns %s", netns_name)

    full_ruleset = f"{_TRACE_INIT_RULESET}\n{rules}"

    with _temp_rules_file(full_ruleset) as path:
        cmd = [*self.exec_prefix(netns_name), resolve("nft"), "-f", path]
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            check=False,
            timeout=30.0,
        )
        if result.returncode != 0:
            errors = _parse_nft_errors(result.stderr, path)
            raise RuleValidationError(errors)

validate(rules)

Dry-run validation using nft --check.

Source code in nse/core/rule_engine.py

def validate(self, rules: str) -> None:
    """
    Dry-run validation using ``nft --check``.
    """
    with _temp_rules_file(rules) as path:
        result = subprocess.run(
            [resolve("nft"), "--check", "-f", path],
            capture_output=True,
            text=True,
            check=False,
            timeout=10.0,
        )
        if result.returncode != 0:
            errors = _parse_nft_errors(result.stderr, path)
            raise RuleValidationError(errors)

RuleValidationError

Bases: Exception

Raised when nft -f rejects the supplied ruleset.

Source code in nse/core/rule_engine.py

class RuleValidationError(Exception):
    """Raised when `nft -f` rejects the supplied ruleset."""

    def __init__(self, errors: list[dict[str, Any]]) -> None:
        self.errors = errors
        super().__init__(f"nftables validation failed: {errors}")

---

Trace Harvester

nse.core.trace_harvester

TraceHarvester: parse nft monitor trace output into TraceEvent objects.

nft monitor trace emits lines like::

trace id 1be8aad4 ip filter input packet: iif "veth0" ...
trace id 1be8aad4 ip filter input rule 0x4 (handle 3) tcp dport 80 accept (verdict accept)
trace id 1be8aad4 ip filter input verdict accept
trace id 1be8aad4 ip filter input policy accept

Each line is parsed into a TraceEvent and pushed onto an asyncio.Queue. The sentinel value None is pushed when monitoring ends.

Reliability contract

This module is an oracle: downstream code decides PASS/FAIL from what it reports. An oracle that silently reports nothing is worse than one that crashes, so every way this class can stop seeing is made observable:

• :attr:state records why the read loop ended — a crash, a timeout and a clean end are three distinguishable outcomes, not one silent sentinel.
• :attr:unparsed_trace_lines counts lines that look like trace output but that the regexes did not understand. A kernel or nftables format change shows up here as a number, instead of vanishing at DEBUG level.
• :meth:wait_for_event lets a caller prove the harvester is actually seeing kernel events (see the canary probe in :mod:nse.core.pipeline) rather than merely having scheduled its read loop.

The deadline is a mutable attribute (:meth:extend_deadline) because the previous fixed timeout could expire while packets were still being injected, truncating the verdict stream without any error.

HarvestState

Bases: str, Enum

Terminal state of the trace read loop.

Source code in nse/core/trace_harvester.py

class HarvestState(str, enum.Enum):
    """Terminal state of the trace read loop."""

    NOT_STARTED = "not_started"
    RUNNING = "running"
    #: `nft monitor trace` closed its stdout after a stop() was requested.
    STOPPED = "stopped"
    #: stdout reached EOF without anybody asking for it (the monitor died).
    CLEAN_EOF = "clean_eof"
    #: The deadline expired while the monitor was still running.
    TIMEOUT = "timeout"
    #: The read loop raised.
    ERROR = "error"

    @property
    def is_healthy(self) -> bool:
        """True only for the states that mean 'the oracle worked as intended'."""
        return self in (HarvestState.STOPPED, HarvestState.RUNNING)

is_healthy property

True only for the states that mean 'the oracle worked as intended'.

TraceHarvester

Async subprocess wrapper for nft monitor trace.

Usage::

harvester = TraceHarvester()
await harvester.start(netns_name="nse_abc", queue=event_queue)
await harvester.wait_for_event(timeout=1.0)   # prove it can see
# later…
state = await harvester.aclose()

Source code in nse/core/trace_harvester.py

class TraceHarvester:
    """
    Async subprocess wrapper for `nft monitor trace`.

    Usage::

        harvester = TraceHarvester()
        await harvester.start(netns_name="nse_abc", queue=event_queue)
        await harvester.wait_for_event(timeout=1.0)   # prove it can see
        # later…
        state = await harvester.aclose()
    """

    def __init__(self) -> None:
        self._proc: asyncio.subprocess.Process | None = None
        self._task: asyncio.Task[None] | None = None
        self._ready_event = asyncio.Event()
        self._event_signal = asyncio.Event()
        self._stop_requested = False
        self._deadline: float = 0.0

        self.state: HarvestState = HarvestState.NOT_STARTED
        #: Exception that ended the read loop, if any.
        self.error: BaseException | None = None
        #: trace ids observed so far, in first-seen order.
        self.seen_trace_ids: list[str] = []
        #: Lines that look like trace output but that no pattern matched.
        self.unparsed_trace_lines: int = 0
        #: Lines that do not look like trace output at all (banners, warnings).
        self.unparsed_other_lines: int = 0
        #: A few unparsed lines kept verbatim, for the failure message.
        self.unparsed_samples: list[str] = []

    # ------------------------------------------------------------------
    # Readiness / liveness
    # ------------------------------------------------------------------

    async def wait_ready(self, timeout: float = 2.0) -> bool:
        """
        Wait until the read loop is scheduled and reading.

        .. warning::
           This proves only that *this process* is ready to read. It does **not**
           prove that ``nft monitor trace`` has subscribed to the kernel's
           netlink group, so it is not sufficient on its own to conclude that a
           later packet will be observed. Use :meth:`wait_for_event` with a
           canary packet for that.
        """
        try:
            await asyncio.wait_for(self._ready_event.wait(), timeout=timeout)
            return True
        except asyncio.TimeoutError:
            return False

    async def wait_for_event(self, timeout: float = 1.0) -> bool:
        """
        Wait until at least one trace event has been parsed since the last
        :meth:`arm_event_signal`.

        This is the real readiness proof: it returns True only once the kernel
        has actually delivered a trace event through the monitor process.
        """
        try:
            await asyncio.wait_for(self._event_signal.wait(), timeout=timeout)
            return True
        except asyncio.TimeoutError:
            return False

    def arm_event_signal(self) -> None:
        """Clear the event signal so the next :meth:`wait_for_event` waits for a *new* event."""
        self._event_signal.clear()

    def extend_deadline(self, seconds: float) -> None:
        """
        Push the read-loop deadline at least *seconds* into the future.

        Called after every packet injection so a long packet sequence cannot run
        past a deadline that was fixed when the loop started.
        """
        target = asyncio.get_event_loop().time() + seconds
        self._deadline = max(self._deadline, target)

    # ------------------------------------------------------------------
    # Lifecycle
    # ------------------------------------------------------------------

    async def start(
        self,
        netns_name: str,
        queue: asyncio.Queue[TraceEvent | None],
        timeout: float = 10.0,
        use_nsenter: bool = False,
        on_event: Callable[[TraceEvent], None] | None = None,
    ) -> None:
        """
        Launch `nft monitor trace` inside *netns_name* and begin streaming.

        Args:
            netns_name: The target network namespace.
            queue:      Output queue (TraceEvent or None sentinel on completion).
            timeout:    Initial seconds of inactivity before giving up. Extend it
                        with :meth:`extend_deadline` as work proceeds.
            use_nsenter: Use nsenter fallback in container environments.
            on_event:   Optional callback invoked synchronously on every parsed TraceEvent.
        """
        self._ready_event.clear()
        self._event_signal.clear()
        self._stop_requested = False
        self.error = None
        if use_nsenter:
            cmd = [
                resolve("nsenter"),
                f"--net=/var/run/netns/{netns_name}",
                "--",
                resolve("nft"),
                "monitor",
                "trace",
            ]
        else:
            cmd = [resolve("ip"), "netns", "exec", netns_name, resolve("nft"), "monitor", "trace"]
        logger.debug("Starting trace monitor: %s", " ".join(cmd))

        self._proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=None,
        )
        self.state = HarvestState.RUNNING
        self._deadline = asyncio.get_event_loop().time() + timeout
        self._task = asyncio.ensure_future(self._read_loop(queue=queue, on_event=on_event))

    async def _read_loop(
        self,
        queue: asyncio.Queue[TraceEvent | None],
        on_event: Callable[[TraceEvent], None] | None = None,
    ) -> None:
        """Read stdout line by line, parse, and push to queue."""

        assert self._proc is not None
        assert self._proc.stdout is not None

        self._ready_event.set()
        end_state = HarvestState.CLEAN_EOF

        try:
            while True:
                remaining = self._deadline - asyncio.get_event_loop().time()
                if remaining <= 0:
                    logger.warning("Trace monitor deadline reached with the monitor still open.")
                    end_state = (
                        HarvestState.STOPPED if self._stop_requested else HarvestState.TIMEOUT
                    )
                    break

                try:
                    line_bytes = await asyncio.wait_for(
                        self._proc.stdout.readline(),
                        timeout=min(remaining, _DEADLINE_POLL_INTERVAL),
                    )
                except asyncio.TimeoutError:
                    # Not necessarily the end: re-read the deadline, which an
                    # injection may have pushed further out while we waited.
                    continue

                if not line_bytes:
                    # EOF. Expected when we asked the monitor to stop; otherwise
                    # the monitor died on its own and the oracle went blind.
                    end_state = (
                        HarvestState.STOPPED if self._stop_requested else HarvestState.CLEAN_EOF
                    )
                    break

                line = line_bytes.decode(errors="replace").strip()
                if not line:
                    continue

                event = _parse_line(line)
                if event is None:
                    self._record_unparsed(line)
                    continue

                logger.debug("TraceEvent: %s", event)
                if event.trace_id and event.trace_id not in self.seen_trace_ids:
                    self.seen_trace_ids.append(event.trace_id)
                await queue.put(event)
                self._event_signal.set()
                if on_event is not None:
                    on_event(event)

        except asyncio.CancelledError:
            self.state = HarvestState.STOPPED
            raise
        except Exception as exc:
            logger.exception("Error in trace read loop")
            self.error = exc
            end_state = HarvestState.ERROR
        finally:
            if self.state is HarvestState.RUNNING:
                self.state = end_state
            with contextlib.suppress(asyncio.CancelledError):
                await queue.put(None)
            self._terminate_process()

    def _record_unparsed(self, line: str) -> None:
        """Account for a line the parser did not understand."""
        if _TRACE_LINE_RE.match(line):
            self.unparsed_trace_lines += 1
            logger.warning("Unparsed TRACE line (oracle blind spot): %r", line)
        else:
            self.unparsed_other_lines += 1
            logger.debug("Unparsed non-trace line: %r", line)
        if len(self.unparsed_samples) < _MAX_UNPARSED_SAMPLES:
            self.unparsed_samples.append(line)

    def _terminate_process(self) -> None:
        if self._proc and self._proc.returncode is None:
            with contextlib.suppress(ProcessLookupError, OSError):
                self._proc.terminate()

    def stop(self) -> None:
        """
        Ask the monitor to terminate. Does not cancel the read loop, so events
        already buffered are still parsed; prefer :meth:`aclose`, which waits.
        """
        self._stop_requested = True
        self._terminate_process()

    async def aclose(self, timeout: float = 2.0) -> HarvestState:
        """
        Stop the monitor, drain the read loop, and return the terminal state.

        Unlike the old ``stop()``, this never cancels the read loop out from
        under a partially-parsed line, so the state it returns is trustworthy.
        """
        self.stop()
        if self._task is not None and not self._task.done():
            try:
                await asyncio.wait_for(asyncio.shield(self._task), timeout=timeout)
            except asyncio.TimeoutError:
                self._task.cancel()
                with contextlib.suppress(asyncio.CancelledError, Exception):
                    await self._task
            except asyncio.CancelledError:
                self.state = HarvestState.STOPPED
        if self.state is HarvestState.RUNNING:
            self.state = HarvestState.STOPPED
        return self.state

    # ------------------------------------------------------------------
    # Diagnostics
    # ------------------------------------------------------------------

    def health_errors(self) -> list[str]:
        """
        Return the reasons this harvester cannot be trusted, empty if it can.

        The pipeline turns each of these into an ``error`` TraceEvent, which the
        CLI runner already fails on.
        """
        problems: list[str] = []
        if self.state is HarvestState.TIMEOUT:
            problems.append(
                "trace monitor timed out while still running: the verdict stream is truncated"
            )
        elif self.state is HarvestState.CLEAN_EOF:
            problems.append(
                "trace monitor exited on its own before the run finished "
                "(is `nft monitor trace` available inside the namespace?)"
            )
        elif self.state is HarvestState.ERROR:
            problems.append(f"trace read loop crashed: {self.error!r}")
        elif self.state is HarvestState.NOT_STARTED:
            problems.append("trace monitor was never started")

        if self.unparsed_trace_lines:
            sample = "; ".join(repr(s) for s in self.unparsed_samples[:3])
            problems.append(
                f"{self.unparsed_trace_lines} trace line(s) could not be parsed - the "
                f"parser does not understand this nftables/kernel output: {sample}"
            )
        return problems

aclose(timeout=2.0) async

Stop the monitor, drain the read loop, and return the terminal state.

Unlike the old stop(), this never cancels the read loop out from under a partially-parsed line, so the state it returns is trustworthy.

Source code in nse/core/trace_harvester.py

async def aclose(self, timeout: float = 2.0) -> HarvestState:
    """
    Stop the monitor, drain the read loop, and return the terminal state.

    Unlike the old ``stop()``, this never cancels the read loop out from
    under a partially-parsed line, so the state it returns is trustworthy.
    """
    self.stop()
    if self._task is not None and not self._task.done():
        try:
            await asyncio.wait_for(asyncio.shield(self._task), timeout=timeout)
        except asyncio.TimeoutError:
            self._task.cancel()
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await self._task
        except asyncio.CancelledError:
            self.state = HarvestState.STOPPED
    if self.state is HarvestState.RUNNING:
        self.state = HarvestState.STOPPED
    return self.state

arm_event_signal()

Clear the event signal so the next :meth:wait_for_event waits for a new event.

Source code in nse/core/trace_harvester.py

def arm_event_signal(self) -> None:
    """Clear the event signal so the next :meth:`wait_for_event` waits for a *new* event."""
    self._event_signal.clear()

extend_deadline(seconds)

Push the read-loop deadline at least seconds into the future.

Called after every packet injection so a long packet sequence cannot run past a deadline that was fixed when the loop started.

Source code in nse/core/trace_harvester.py

def extend_deadline(self, seconds: float) -> None:
    """
    Push the read-loop deadline at least *seconds* into the future.

    Called after every packet injection so a long packet sequence cannot run
    past a deadline that was fixed when the loop started.
    """
    target = asyncio.get_event_loop().time() + seconds
    self._deadline = max(self._deadline, target)

health_errors()

Return the reasons this harvester cannot be trusted, empty if it can.

The pipeline turns each of these into an error TraceEvent, which the CLI runner already fails on.

Source code in nse/core/trace_harvester.py

def health_errors(self) -> list[str]:
    """
    Return the reasons this harvester cannot be trusted, empty if it can.

    The pipeline turns each of these into an ``error`` TraceEvent, which the
    CLI runner already fails on.
    """
    problems: list[str] = []
    if self.state is HarvestState.TIMEOUT:
        problems.append(
            "trace monitor timed out while still running: the verdict stream is truncated"
        )
    elif self.state is HarvestState.CLEAN_EOF:
        problems.append(
            "trace monitor exited on its own before the run finished "
            "(is `nft monitor trace` available inside the namespace?)"
        )
    elif self.state is HarvestState.ERROR:
        problems.append(f"trace read loop crashed: {self.error!r}")
    elif self.state is HarvestState.NOT_STARTED:
        problems.append("trace monitor was never started")

    if self.unparsed_trace_lines:
        sample = "; ".join(repr(s) for s in self.unparsed_samples[:3])
        problems.append(
            f"{self.unparsed_trace_lines} trace line(s) could not be parsed - the "
            f"parser does not understand this nftables/kernel output: {sample}"
        )
    return problems

start(netns_name, queue, timeout=10.0, use_nsenter=False, on_event=None) async

Launch nft monitor trace inside netns_name and begin streaming.

Parameters:

Name	Type	Description	Default
netns_name	str	The target network namespace.	required
queue	Queue[TraceEvent | None]	Output queue (TraceEvent or None sentinel on completion).	required
timeout	float	Initial seconds of inactivity before giving up. Extend it with :meth:extend_deadline as work proceeds.	10.0
use_nsenter	bool	Use nsenter fallback in container environments.	False
on_event	Callable[[TraceEvent], None] | None	Optional callback invoked synchronously on every parsed TraceEvent.	None

Source code in nse/core/trace_harvester.py

async def start(
    self,
    netns_name: str,
    queue: asyncio.Queue[TraceEvent | None],
    timeout: float = 10.0,
    use_nsenter: bool = False,
    on_event: Callable[[TraceEvent], None] | None = None,
) -> None:
    """
    Launch `nft monitor trace` inside *netns_name* and begin streaming.

    Args:
        netns_name: The target network namespace.
        queue:      Output queue (TraceEvent or None sentinel on completion).
        timeout:    Initial seconds of inactivity before giving up. Extend it
                    with :meth:`extend_deadline` as work proceeds.
        use_nsenter: Use nsenter fallback in container environments.
        on_event:   Optional callback invoked synchronously on every parsed TraceEvent.
    """
    self._ready_event.clear()
    self._event_signal.clear()
    self._stop_requested = False
    self.error = None
    if use_nsenter:
        cmd = [
            resolve("nsenter"),
            f"--net=/var/run/netns/{netns_name}",
            "--",
            resolve("nft"),
            "monitor",
            "trace",
        ]
    else:
        cmd = [resolve("ip"), "netns", "exec", netns_name, resolve("nft"), "monitor", "trace"]
    logger.debug("Starting trace monitor: %s", " ".join(cmd))

    self._proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=None,
    )
    self.state = HarvestState.RUNNING
    self._deadline = asyncio.get_event_loop().time() + timeout
    self._task = asyncio.ensure_future(self._read_loop(queue=queue, on_event=on_event))

stop()

Ask the monitor to terminate. Does not cancel the read loop, so events already buffered are still parsed; prefer :meth:aclose, which waits.

Source code in nse/core/trace_harvester.py

def stop(self) -> None:
    """
    Ask the monitor to terminate. Does not cancel the read loop, so events
    already buffered are still parsed; prefer :meth:`aclose`, which waits.
    """
    self._stop_requested = True
    self._terminate_process()

wait_for_event(timeout=1.0) async

Wait until at least one trace event has been parsed since the last :meth:arm_event_signal.

This is the real readiness proof: it returns True only once the kernel has actually delivered a trace event through the monitor process.

Source code in nse/core/trace_harvester.py

async def wait_for_event(self, timeout: float = 1.0) -> bool:
    """
    Wait until at least one trace event has been parsed since the last
    :meth:`arm_event_signal`.

    This is the real readiness proof: it returns True only once the kernel
    has actually delivered a trace event through the monitor process.
    """
    try:
        await asyncio.wait_for(self._event_signal.wait(), timeout=timeout)
        return True
    except asyncio.TimeoutError:
        return False

wait_ready(timeout=2.0) async

Wait until the read loop is scheduled and reading.

.. warning:: This proves only that this process is ready to read. It does not prove that nft monitor trace has subscribed to the kernel's netlink group, so it is not sufficient on its own to conclude that a later packet will be observed. Use :meth:wait_for_event with a canary packet for that.

Source code in nse/core/trace_harvester.py

async def wait_ready(self, timeout: float = 2.0) -> bool:
    """
    Wait until the read loop is scheduled and reading.

    .. warning::
       This proves only that *this process* is ready to read. It does **not**
       prove that ``nft monitor trace`` has subscribed to the kernel's
       netlink group, so it is not sufficient on its own to conclude that a
       later packet will be observed. Use :meth:`wait_for_event` with a
       canary packet for that.
    """
    try:
        await asyncio.wait_for(self._ready_event.wait(), timeout=timeout)
        return True
    except asyncio.TimeoutError:
        return False

---

Scapy Packet Injector

nse.core.scapy_injector

ScapyInjector: forge and inject raw packets into a network namespace.

ScapyInjector

Build and inject a packet described by a PacketSpec into a netns.

Source code in nse/core/scapy_injector.py

class ScapyInjector:
    """Build and inject a packet described by a PacketSpec into a netns."""

    def __init__(self, use_nsenter: bool = False) -> None:
        self.use_nsenter = use_nsenter

    def inject(
        self,
        spec: PacketSpec,
        netns_name: str,
        veth_host: str,
        veth_peer: str,
        host_netns: str | None = None,
    ) -> None:
        """
        Construct and send a packet matching *spec* inside or into *netns_name*.

        Args:
            spec:        The packet to forge.
            netns_name:  Namespace owning *veth_peer*.
            veth_host:   Interface the packet is sent from.
            veth_peer:   Interface inside *netns_name*.
            host_netns:  Namespace owning *veth_host*, or None when it is in the
                         root namespace. The caller knows this; it used to be
                         guessed from the interface name and the guess produced a
                         namespace ("nse_router_<suffix>") that never existed, so
                         the MAC lookup could only ever fail.
        """
        logger.info(
            "Injecting %s packet: %s -> %s (netns=%s)",
            spec.protocol.upper(),
            spec.src_ip,
            spec.dst_ip,
            netns_name,
        )

        try:
            host_ns = host_netns
            host_mac = _get_mac_address(veth_host, host_ns, self.use_nsenter)
            peer_mac = _get_mac_address(veth_peer, netns_name, self.use_nsenter)
        except Exception as exc:
            logger.error("Failed to retrieve MAC addresses: %s", exc)
            raise RuntimeError(f"Failed to retrieve MAC addresses: {exc}") from exc

        # Determine injection direction:
        # If the source IP matches the sandbox IP (default 10.0.0.2), it is outgoing (Netns -> Host).
        # Otherwise, it is incoming (Host -> Netns).
        is_incoming = True
        if spec.src_ip in ("10.0.0.2", "fd00::2", "10.0.2.2", "fd00:2::2"):
            is_incoming = False

        if is_incoming:
            # Incoming: Host -> Netns.
            # Src MAC = host, Dst MAC = peer. Send on host interface.
            src_mac = host_mac
            dst_mac = peer_mac
            inject_interface = veth_host

            try:
                logger.debug(
                    "Performing in-process L2 injection on host interface %s",
                    inject_interface,
                )
                from scapy.all import (  # type: ignore[attr-defined]
                    ICMP,
                    IP,
                    TCP,
                    UDP,
                    Ether,
                    ICMPv6EchoRequest,
                    IPv6,
                    conf,
                    sendp,
                )

                conf.verb = 0

                proto = spec.protocol.lower()
                src_port = spec.src_port or 12345
                dst_port = spec.dst_port or 80

                # Formulate Layer 4 packet payload
                is_ipv6 = ":" in spec.src_ip
                if proto == "tcp":
                    tcp_flags_str = "".join(spec.tcp_flags) if spec.tcp_flags else ""
                    l4 = TCP(sport=src_port, dport=dst_port, flags=tcp_flags_str)
                elif proto == "udp":
                    l4 = UDP(sport=src_port, dport=dst_port)
                elif proto == "icmp":
                    l4 = ICMPv6EchoRequest() if is_ipv6 else ICMP()
                else:
                    l4 = TCP(sport=src_port, dport=dst_port)

                if is_ipv6:
                    l3 = IPv6(src=spec.src_ip, dst=spec.dst_ip)
                else:
                    l3 = IP(src=spec.src_ip, dst=spec.dst_ip)

                pkt = Ether(src=src_mac, dst=dst_mac) / l3 / l4
                sendp(pkt, iface=inject_interface, verbose=False)
                logger.debug("In-process packet sent successfully: %s", pkt.summary())
            except Exception as exc:
                logger.error("In-process Scapy injection failed: %s", exc)
                raise RuntimeError(f"Packet injection failed: {exc}") from exc

        else:
            # Outgoing: Netns -> Host.
            # Src MAC = peer, Dst MAC = host. Send on peer interface from inside netns context.
            # Fallback to ip netns exec / nsenter since we need to change namespace context.
            src_mac = peer_mac
            dst_mac = host_mac
            inject_interface = veth_peer

            script = _build_scapy_script(
                spec=spec,
                interface=inject_interface,
                src_mac=src_mac,
                dst_mac=dst_mac,
            )

            logger.debug(
                "Running injection inside netns %s on interface %s",
                netns_name,
                inject_interface,
            )
            cmd = []
            if self.use_nsenter:
                cmd += [resolve("nsenter"), f"--net=/var/run/netns/{netns_name}", "--"]
            else:
                cmd += [resolve("ip"), "netns", "exec", netns_name]
            cmd += [
                sys.executable,
                "-c",
                script,
            ]

            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                check=False,
                timeout=10.0,
            )

            if result.returncode != 0:
                logger.error("Scapy injection failed:\n%s", result.stderr)
                raise RuntimeError(f"Packet injection failed: {result.stderr.strip()}")

            logger.debug("Injection stdout: %s", result.stdout.strip())

inject(spec, netns_name, veth_host, veth_peer, host_netns=None)

Construct and send a packet matching spec inside or into netns_name.

Parameters:

Name	Type	Description	Default
spec	PacketSpec	The packet to forge.	required
netns_name	str	Namespace owning veth_peer.	required
veth_host	str	Interface the packet is sent from.	required
veth_peer	str	Interface inside netns_name.	required
host_netns	str | None	Namespace owning veth_host, or None when it is in the root namespace. The caller knows this; it used to be guessed from the interface name and the guess produced a namespace ("nse_router_") that never existed, so the MAC lookup could only ever fail.	None

Source code in nse/core/scapy_injector.py

def inject(
    self,
    spec: PacketSpec,
    netns_name: str,
    veth_host: str,
    veth_peer: str,
    host_netns: str | None = None,
) -> None:
    """
    Construct and send a packet matching *spec* inside or into *netns_name*.

    Args:
        spec:        The packet to forge.
        netns_name:  Namespace owning *veth_peer*.
        veth_host:   Interface the packet is sent from.
        veth_peer:   Interface inside *netns_name*.
        host_netns:  Namespace owning *veth_host*, or None when it is in the
                     root namespace. The caller knows this; it used to be
                     guessed from the interface name and the guess produced a
                     namespace ("nse_router_<suffix>") that never existed, so
                     the MAC lookup could only ever fail.
    """
    logger.info(
        "Injecting %s packet: %s -> %s (netns=%s)",
        spec.protocol.upper(),
        spec.src_ip,
        spec.dst_ip,
        netns_name,
    )

    try:
        host_ns = host_netns
        host_mac = _get_mac_address(veth_host, host_ns, self.use_nsenter)
        peer_mac = _get_mac_address(veth_peer, netns_name, self.use_nsenter)
    except Exception as exc:
        logger.error("Failed to retrieve MAC addresses: %s", exc)
        raise RuntimeError(f"Failed to retrieve MAC addresses: {exc}") from exc

    # Determine injection direction:
    # If the source IP matches the sandbox IP (default 10.0.0.2), it is outgoing (Netns -> Host).
    # Otherwise, it is incoming (Host -> Netns).
    is_incoming = True
    if spec.src_ip in ("10.0.0.2", "fd00::2", "10.0.2.2", "fd00:2::2"):
        is_incoming = False

    if is_incoming:
        # Incoming: Host -> Netns.
        # Src MAC = host, Dst MAC = peer. Send on host interface.
        src_mac = host_mac
        dst_mac = peer_mac
        inject_interface = veth_host

        try:
            logger.debug(
                "Performing in-process L2 injection on host interface %s",
                inject_interface,
            )
            from scapy.all import (  # type: ignore[attr-defined]
                ICMP,
                IP,
                TCP,
                UDP,
                Ether,
                ICMPv6EchoRequest,
                IPv6,
                conf,
                sendp,
            )

            conf.verb = 0

            proto = spec.protocol.lower()
            src_port = spec.src_port or 12345
            dst_port = spec.dst_port or 80

            # Formulate Layer 4 packet payload
            is_ipv6 = ":" in spec.src_ip
            if proto == "tcp":
                tcp_flags_str = "".join(spec.tcp_flags) if spec.tcp_flags else ""
                l4 = TCP(sport=src_port, dport=dst_port, flags=tcp_flags_str)
            elif proto == "udp":
                l4 = UDP(sport=src_port, dport=dst_port)
            elif proto == "icmp":
                l4 = ICMPv6EchoRequest() if is_ipv6 else ICMP()
            else:
                l4 = TCP(sport=src_port, dport=dst_port)

            if is_ipv6:
                l3 = IPv6(src=spec.src_ip, dst=spec.dst_ip)
            else:
                l3 = IP(src=spec.src_ip, dst=spec.dst_ip)

            pkt = Ether(src=src_mac, dst=dst_mac) / l3 / l4
            sendp(pkt, iface=inject_interface, verbose=False)
            logger.debug("In-process packet sent successfully: %s", pkt.summary())
        except Exception as exc:
            logger.error("In-process Scapy injection failed: %s", exc)
            raise RuntimeError(f"Packet injection failed: {exc}") from exc

    else:
        # Outgoing: Netns -> Host.
        # Src MAC = peer, Dst MAC = host. Send on peer interface from inside netns context.
        # Fallback to ip netns exec / nsenter since we need to change namespace context.
        src_mac = peer_mac
        dst_mac = host_mac
        inject_interface = veth_peer

        script = _build_scapy_script(
            spec=spec,
            interface=inject_interface,
            src_mac=src_mac,
            dst_mac=dst_mac,
        )

        logger.debug(
            "Running injection inside netns %s on interface %s",
            netns_name,
            inject_interface,
        )
        cmd = []
        if self.use_nsenter:
            cmd += [resolve("nsenter"), f"--net=/var/run/netns/{netns_name}", "--"]
        else:
            cmd += [resolve("ip"), "netns", "exec", netns_name]
        cmd += [
            sys.executable,
            "-c",
            script,
        ]

        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            check=False,
            timeout=10.0,
        )

        if result.returncode != 0:
            logger.error("Scapy injection failed:\n%s", result.stderr)
            raise RuntimeError(f"Packet injection failed: {result.stderr.strip()}")

        logger.debug("Injection stdout: %s", result.stdout.strip())

---

Data Models

nse.models.test_request

Pydantic models for incoming test requests.

PacketSpec

Bases: BaseModel

Describes the packet to forge and inject.

Source code in nse/models/test_request.py

class PacketSpec(BaseModel):
    """Describes the packet to forge and inject."""

    protocol: Literal["tcp", "udp", "icmp"] = Field(
        description="Layer 4 protocol.",
        examples=["tcp"],
    )
    src_ip: str = Field(
        default="10.0.0.1",
        description="Source IP address (IPv4 or IPv6).",
        examples=["192.168.1.10", "fd00::1"],
    )
    dst_ip: str = Field(
        default="10.0.0.2",
        description="Destination IP address (IPv4 or IPv6).",
        examples=["192.168.1.1", "fd00::2"],
    )
    src_port: int | None = Field(
        default=None,
        ge=1,
        le=65535,
        description="Source port (TCP/UDP only).",
        examples=[54321],
    )
    dst_port: int | None = Field(
        default=None,
        ge=1,
        le=65535,
        description="Destination port (TCP/UDP only).",
        examples=[80],
    )
    tcp_flags: list[str] = Field(
        default_factory=list,
        description="List of TCP flag names to set (e.g. ['S'] for SYN).",
        examples=[["S"]],
    )

    @field_validator("src_ip", "dst_ip")
    @classmethod
    def validate_ip(cls, v: str) -> str:
        try:
            ipaddress.ip_address(v)
        except ValueError as exc:
            raise ValueError(f"Invalid IP address (must be IPv4 or IPv6): {v!r}") from exc
        return v

    @field_validator("tcp_flags")
    @classmethod
    def validate_tcp_flags(cls, flags: list[str]) -> list[str]:
        valid = {"F", "S", "R", "P", "A", "U", "E", "C"}
        for f in flags:
            if f.upper() not in valid:
                raise ValueError(f"Invalid TCP flag: {f!r}. Valid: {valid}")
        return [f.upper() for f in flags]

TestRequest

Bases: BaseModel

Top-level body for POST /api/test.

Source code in nse/models/test_request.py

class TestRequest(BaseModel):
    """Top-level body for POST /api/test."""

    __test__ = False

    rules: str = Field(
        description="Raw nftables ruleset text (passed verbatim to `nft -f`).",
        min_length=1,
        examples=[
            "table ip filter {\n  chain input {\n    type filter hook input priority 0;\n    tcp dport 22 accept\n    drop\n  }\n}"
        ],
    )
    packets: list[PacketSpec] = Field(
        description="Sequence of packets to forge and inject into the sandboxed namespace.",
        min_length=1,
    )
    topology: TopologyType = Field(
        default=TopologyType.SIMPLE,
        description="Sandbox network topology configuration.",
        examples=[TopologyType.SIMPLE],
    )

nse.models.trace_event

Pydantic models for trace events streamed over WebSocket.

TestStatusResponse

Bases: BaseModel

Response body for GET /api/test/{test_id}.

Source code in nse/models/trace_event.py

class TestStatusResponse(BaseModel):
    """Response body for GET /api/test/{test_id}."""

    __test__ = False

    test_id: str
    status: Literal["pending", "running", "done", "error"]

TraceEvent

Bases: BaseModel

A single event emitted by nft monitor trace (or the pipeline itself).

Source code in nse/models/trace_event.py

class TraceEvent(BaseModel):
    """A single event emitted by `nft monitor trace` (or the pipeline itself)."""

    type: Literal["hook", "match", "verdict", "error", "ping", "conntrack"] = Field(
        description="Event category."
    )
    trace_id: str | None = Field(
        default=None,
        description="nft internal trace identifier (hex string).",
    )
    family: str | None = Field(
        default=None,
        description="Address family (ip, ip6, inet…).",
    )
    table: str | None = Field(
        default=None,
        description="nftables table name.",
    )
    chain: str | None = Field(
        default=None,
        description="nftables chain name.",
    )
    hook: str | None = Field(
        default=None,
        description="Hook name or incoming interface name (for 'hook' events).",
    )
    rule_handle: int | None = Field(
        default=None,
        description="Rule handle number (for 'match' events).",
    )
    rule_text: str | None = Field(
        default=None,
        description="Partial rule text as printed by nft (for 'match' events).",
    )
    verdict: str | None = Field(
        default=None,
        description="Verdict string: ACCEPT, DROP, REJECT, CONTINUE… (for 'match'/'verdict' events).",
    )
    raw_message: str | None = Field(
        default=None,
        description="Free-form error message (for 'error' events).",
    )
    timestamp: float | None = Field(
        default=None,
        description="Unix timestamp of the event (seconds).",
    )
    # Conntrack table state fields
    ct_proto: str | None = Field(default=None, description="Conntrack layer 4 protocol.")
    ct_state: str | None = Field(default=None, description="Conntrack connection state.")
    ct_src: str | None = Field(default=None, description="Conntrack source IP.")
    ct_dst: str | None = Field(default=None, description="Conntrack destination IP.")
    ct_sport: int | None = Field(default=None, description="Conntrack source port.")
    ct_dport: int | None = Field(default=None, description="Conntrack destination port.")