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.

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.

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.
    """
    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] = []

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

    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

    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 & wait for readiness probe
        # ------------------------------------------------------------------
        logger.info("[%s] Starting nft monitor trace", run.test_id)

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

        await harvester.start(
            netns_name=target_netns,
            queue=event_queue,
            timeout=_TRACE_TIMEOUT,
            use_nsenter=controller.use_nsenter,
            on_event=on_trace_event,
        )

        # Readiness probe replacement for hardcoded delay
        await harvester.wait_ready(timeout=2.0)

        # ------------------------------------------------------------------
        # 5. 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,
            )
            veth_peer_target = veth_router_host if is_gateway else _VETH_PEER
            await loop.run_in_executor(
                None,
                injector.inject,
                pkt,
                target_netns,
                veth_host,
                veth_peer_target,
            )
            # Short per-packet delay to let kernel trace process the verdict deterministically
            await asyncio.sleep(0.15)

        # ------------------------------------------------------------------
        # 6. Collect conntrack entries & finish
        # ------------------------------------------------------------------
        await asyncio.sleep(0.3)

        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']}",
                )
            )

        harvester.stop()
        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)
        harvester.stop()

    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)
        harvester.stop()

    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(
                ["ip", "link", "del", veth_host],
                capture_output=True,
                check=False,
            ),
        )

    return collected_events

---

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(
                ["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(
                            ["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(
                ["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(
                                ["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 ["nsenter", f"--net=/var/run/netns/{name}", "--"]
        else:
            return ["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(["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,
                [
                    "sysctl",
                    "-w",
                    "net.ipv6.conf.all.accept_dad=0",
                ],
            )
            self._run_in_netns(
                name,
                [
                    "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(["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, ["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(["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(["ip", "link", "add", veth_host, "type", "veth", "peer", "name", veth_peer])
        # Move the peer end into the target namespace
        _run(["ip", "link", "set", veth_peer, "netns", netns_name])

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

        # --- Namespace side ---
        if v4_peer:
            self._run_in_netns(
                netns_name,
                [
                    "ip",
                    "addr",
                    "add",
                    v4_peer,
                    "dev",
                    veth_peer,
                ],
            )
        if v6_peer:
            self._run_in_netns(
                netns_name,
                [
                    "ip",
                    "addr",
                    "add",
                    v6_peer,
                    "dev",
                    veth_peer,
                ],
            )
        self._run_in_netns(netns_name, ["ip", "link", "set", veth_peer, "up"])
        self._run_in_netns(netns_name, ["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, ["sysctl", "-w", "net.ipv4.ip_forward=1"])
        self._run_in_netns(
            router_ns,
            [
                "sysctl",
                "-w",
                "net.ipv6.conf.all.forwarding=1",
            ],
        )

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

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

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

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

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

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

        # Bring up loopbacks
        self._run_in_netns(router_ns, ["ip", "link", "set", "lo", "up"])
        self._run_in_netns(server_ns, ["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(["ip", "route", "add", "10.0.2.0/24", "via", host_rt_via, "dev", veth_host])
        _run(
            [
                "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,
            [
                "ip",
                "route",
                "add",
                "default",
                "via",
                srv_rt_via,
                "dev",
                veth_server,
            ],
        )
        self._run_in_netns(
            server_ns,
            [
                "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, ["sysctl", "-w", "net.ipv4.ip_forward=1"])
    self._run_in_netns(
        router_ns,
        [
            "sysctl",
            "-w",
            "net.ipv6.conf.all.forwarding=1",
        ],
    )

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

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

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

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

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

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

    # Bring up loopbacks
    self._run_in_netns(router_ns, ["ip", "link", "set", "lo", "up"])
    self._run_in_netns(server_ns, ["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(["ip", "route", "add", "10.0.2.0/24", "via", host_rt_via, "dev", veth_host])
    _run(
        [
            "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,
        [
            "ip",
            "route",
            "add",
            "default",
            "via",
            srv_rt_via,
            "dev",
            veth_server,
        ],
    )
    self._run_in_netns(
        server_ns,
        [
            "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, ["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(["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(["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,
            [
                "sysctl",
                "-w",
                "net.ipv6.conf.all.accept_dad=0",
            ],
        )
        self._run_in_netns(
            name,
            [
                "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(["ip", "link", "add", veth_host, "type", "veth", "peer", "name", veth_peer])
    # Move the peer end into the target namespace
    _run(["ip", "link", "set", veth_peer, "netns", netns_name])

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

    # --- Namespace side ---
    if v4_peer:
        self._run_in_netns(
            netns_name,
            [
                "ip",
                "addr",
                "add",
                v4_peer,
                "dev",
                veth_peer,
            ],
        )
    if v6_peer:
        self._run_in_netns(
            netns_name,
            [
                "ip",
                "addr",
                "add",
                v6_peer,
                "dev",
                veth_peer,
            ],
        )
    self._run_in_netns(netns_name, ["ip", "link", "set", veth_peer, "up"])
    self._run_in_netns(netns_name, ["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(["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(
            ["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(
                        ["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(
            ["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(
                            ["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 ["nsenter", f"--net=/var/run/netns/{name}", "--"]
        else:
            return ["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(
                ["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), "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), "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), "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), "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(
            ["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 that the WebSocket handler reads from.

The sentinel value None is pushed when monitoring ends (process exits or times out) to signal the WebSocket to send a "done" event and close.

TraceHarvester

Async subprocess wrapper for nft monitor trace.

Usage::

harvester = TraceHarvester()
await harvester.start(netns_name="nse_abc", queue=event_queue)
# later…
harvester.stop()

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)
        # later…
        harvester.stop()
    """

    def __init__(self) -> None:
        self._proc: asyncio.subprocess.Process | None = None
        self._task: asyncio.Task[None] | None = None
        self._ready_event = asyncio.Event()

    async def wait_ready(self, timeout: float = 2.0) -> bool:
        """Wait until the trace harvester process is up and reading."""
        try:
            await asyncio.wait_for(self._ready_event.wait(), timeout=timeout)
            return True
        except asyncio.TimeoutError:
            return False

    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:    Max seconds to wait for trace events before giving up.
            use_nsenter: Use nsenter fallback in container environments.
            on_event:   Optional callback invoked synchronously on every parsed TraceEvent.
        """
        self._ready_event.clear()
        if use_nsenter:
            cmd = ["nsenter", f"--net=/var/run/netns/{netns_name}", "--", "nft", "monitor", "trace"]
        else:
            cmd = ["ip", "netns", "exec", netns_name, "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._task = asyncio.ensure_future(
            self._read_loop(queue=queue, timeout=timeout, on_event=on_event)
        )

    async def _read_loop(
        self,
        queue: asyncio.Queue[TraceEvent | None],
        timeout: float,
        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

        deadline = asyncio.get_event_loop().time() + timeout
        self._ready_event.set()

        try:
            while True:
                remaining = deadline - asyncio.get_event_loop().time()
                if remaining <= 0:
                    logger.debug("Trace monitor timeout reached.")
                    break

                try:
                    line_bytes = await asyncio.wait_for(
                        self._proc.stdout.readline(), timeout=remaining
                    )
                except asyncio.TimeoutError:
                    break

                if not line_bytes:
                    break  # EOF

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

                event = _parse_line(line)
                if event is not None:
                    logger.debug("TraceEvent: %s", event)
                    await queue.put(event)
                    if on_event is not None:
                        on_event(event)

        except Exception:
            logger.exception("Error in trace read loop")
        finally:
            await queue.put(None)  # Sentinel → WebSocket sends "done"
            self.stop()

    def stop(self) -> None:
        """Terminate the monitor process."""
        if self._proc and self._proc.returncode is None:
            with contextlib.suppress(ProcessLookupError):
                self._proc.terminate()
        if self._task and not self._task.done():
            try:
                current = asyncio.current_task()
            except RuntimeError:
                current = None
            if self._task is not current:
                self._task.cancel()

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	Max seconds to wait for trace events before giving up.	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:    Max seconds to wait for trace events before giving up.
        use_nsenter: Use nsenter fallback in container environments.
        on_event:   Optional callback invoked synchronously on every parsed TraceEvent.
    """
    self._ready_event.clear()
    if use_nsenter:
        cmd = ["nsenter", f"--net=/var/run/netns/{netns_name}", "--", "nft", "monitor", "trace"]
    else:
        cmd = ["ip", "netns", "exec", netns_name, "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._task = asyncio.ensure_future(
        self._read_loop(queue=queue, timeout=timeout, on_event=on_event)
    )

stop()

Terminate the monitor process.

Source code in nse/core/trace_harvester.py

def stop(self) -> None:
    """Terminate the monitor process."""
    if self._proc and self._proc.returncode is None:
        with contextlib.suppress(ProcessLookupError):
            self._proc.terminate()
    if self._task and not self._task.done():
        try:
            current = asyncio.current_task()
        except RuntimeError:
            current = None
        if self._task is not current:
            self._task.cancel()

wait_ready(timeout=2.0) async

Wait until the trace harvester process is up and reading.

Source code in nse/core/trace_harvester.py

async def wait_ready(self, timeout: float = 2.0) -> bool:
    """Wait until the trace harvester process is up and reading."""
    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,
    ) -> None:
        """
        Construct and send a packet matching *spec* inside or into *netns_name*.
        """
        logger.info(
            "Injecting %s packet: %s -> %s (netns=%s)",
            spec.protocol.upper(),
            spec.src_ip,
            spec.dst_ip,
            netns_name,
        )

        try:
            # Detect if host interface lives in a router namespace (gateway topology)
            host_ns = None
            if veth_host.startswith(("vrs-", "vrh-")):
                suffix = veth_host.split("-")[1]
                host_ns = f"nse_router_{suffix}"
            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 += ["nsenter", f"--net=/var/run/netns/{netns_name}", "--"]
            else:
                cmd += ["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)

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

Source code in nse/core/scapy_injector.py

def inject(
    self,
    spec: PacketSpec,
    netns_name: str,
    veth_host: str,
    veth_peer: str,
) -> None:
    """
    Construct and send a packet matching *spec* inside or into *netns_name*.
    """
    logger.info(
        "Injecting %s packet: %s -> %s (netns=%s)",
        spec.protocol.upper(),
        spec.src_ip,
        spec.dst_ip,
        netns_name,
    )

    try:
        # Detect if host interface lives in a router namespace (gateway topology)
        host_ns = None
        if veth_host.startswith(("vrs-", "vrh-")):
            suffix = veth_host.split("-")[1]
            host_ns = f"nse_router_{suffix}"
        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 += ["nsenter", f"--net=/var/run/netns/{netns_name}", "--"]
        else:
            cmd += ["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.")