Instrument Control

Enforcing Instrument VLAN Firewall Rules from Python

A default-deny host firewall is only as trustworthy as the process that writes its rules. This guide covers compiling a declarative instrument allowlist into nftables from Python and re-asserting it under drift, pinning LXI, VXI-11, and HiSLIP traffic to its control VLAN inside Security Boundaries & Network Isolation. It complements the client-side TCP control layer in securing lab networks for instrument control systems, which assumes the boundary is already correct — this page is the code that makes it correct and keeps it that way.

Baseline Constraints & Scope

The target is a Linux control host that sits on the instrument VLAN and needs its own packet filter, not just a switch ACL upstream. Managed switches and appliance firewalls enforce segmentation, but a scope on 10.20.0.0/24 can still reach a mistyped 8.8.8.8 if the host itself has a default route and no egress filter — and an LXI instrument with a curl-capable embedded Linux is a genuine exfiltration and lateral-movement surface. The scope here is narrow: generate host-side firewall rules from a Python policy object, apply them idempotently, and detect and repair drift. It assumes nftables on Python 3.11+ (the same shape ports to iptables or, via netsh advfirewall, to Windows), a control host that talks to instruments pinned by literal IP through a VISA Resource Manager Setup, and instruments that speak cleartext SCPI on TCP 5025, HiSLIP on 4880, or VXI-11 via the RPC portmapper on 111.

The design contract is fail closed. LXI, VXI-11, and HiSLIP instruments have no business reaching the internet: they pull no updates you have vetted, and a compromised or merely buggy embedded stack that can open an outbound socket is a data-integrity and safety hazard on the same network as a physical actuator. So egress from the instrument VLAN defaults to DROP, and the only flows that survive are the control-host↔instrument tuples the policy names explicitly. Every rule the host runs is derived from one source of truth, so a review of the policy object is a review of the firewall.

Modeling the Policy as an Allowlist Set

The policy is an allowlist set $A$ of permitted 4-tuples. Let a packet be characterized by $p = (\text{src}, \text{dst}, \text{proto}, \text{port})$. The filter admits $p$ if and only if its tuple is a member of $A$, and drops everything else:

Two properties make this safe to compile mechanically. First, membership is order-independent — set inclusion has no notion of “which rule came first” — so the compiler must preserve that by never emitting a broad ACCEPT that a later DROP is supposed to override. In an ordered ruleset (nftables chains, iptables) the first match wins, so the only faithful encoding of default-deny is: emit every allowlist accept first, then a single terminal drop as the chain policy, and never emit a general accept above the drop. Second, the desired state is a pure function of $A$, so drift detection reduces to a set difference between the desired accept-tuples and the tuples currently live in the kernel. If $A_{\text{active}} \neq A$, the host has drifted and must re-assert.

Default-deny host firewall pinning instrument traffic to its VLAN A control host in the centre carries a firewall block whose allowlist holds three permitted tuples: control-host to scope on HiSLIP 4880, to power supply on SCPI 5025, and to a VXI-11 instrument on RPC 111. Solid accent arrows run from the firewall to the instrument control VLAN on the right, which contains an LXI oscilloscope, a SCPI power supply, and a VXI-11 instrument. A dashed amber arrow from the firewall toward an internet cloud is cut by a DROP badge, showing default-deny egress. Any tuple not in the allowlist is dropped. DEFAULT-DENY EGRESS, ALLOWLIST PINS INSTRUMENT FLOWS Control host nftables allowlist A host → scope :4880 HiSLIP host → PSU :5025 SCPI host → DMM :111 VXI-11 Instrument control VLAN LXI oscilloscope HiSLIP 4880 SCPI power supply raw socket 5025 VXI-11 instrument RPC 111 Internet updates, C2, exfil DROP

Every accept the kernel enforces is compiled from the allowlist; all other egress — including any instrument’s attempt to reach the internet — hits the terminal default-deny.

Compiling a FirewallPolicy to nftables from Python

The FirewallPolicy dataclass below is the single source of truth. It renders a complete nft -f ruleset into a dedicated table so the apply is atomic and idempotent — nft replaces the whole table in one transaction, so re-running the same policy is a no-op at the kernel level, and there is no half-applied window. Egress from the instrument subnet defaults to drop; only the named tuples and the return traffic of established connections survive. The verify step re-parses the live ruleset as JSON and diffs the active accept-tuples against the desired set, so drift is a structured finding, not a guess.

from __future__ import annotations

import ipaddress
import json
import logging
import subprocess
from dataclasses import dataclass, field

logger = logging.getLogger("instrument_fw")

TABLE = "inet instrument_guard"


class FirewallError(RuntimeError):
    """nft invocation failed or the ruleset could not be asserted."""


@dataclass(frozen=True)
class AllowRule:
    """One permitted control-host -> instrument flow."""

    control_host: ipaddress.IPv4Address
    instrument: ipaddress.IPv4Address
    port: int
    proto: str = "tcp"          # tcp | udp (udp for VXI-11 portmap)

    def tuple(self) -> tuple[str, str, str, int]:
        return (str(self.control_host), str(self.instrument), self.proto, self.port)


@dataclass(frozen=True)
class FirewallPolicy:
    """Declarative allowlist that compiles to a default-deny nftables table."""

    vlan: ipaddress.IPv4Network      # e.g. 10.20.0.0/24 instrument VLAN
    rules: tuple[AllowRule, ...] = field(default_factory=tuple)
    mgmt_cidr: str | None = None     # management plane kept reachable, always

    def desired_tuples(self) -> frozenset[tuple[str, str, str, int]]:
        return frozenset(r.tuple() for r in self.rules)

    def render(self) -> str:
        """Emit a full `nft -f` ruleset. Accepts first, then default-deny."""
        for r in self.rules:
            if r.instrument not in self.vlan:
                raise FirewallError(f"{r.instrument} is outside VLAN {self.vlan}")
        lines = ["table " + TABLE + " {",
                 "  chain egress {",
                 "    type filter hook output priority 0; policy drop;",
                 "    ct state established,related accept",
                 "    oifname \"lo\" accept"]
        if self.mgmt_cidr:                      # never lock out the management plane
            lines.append(f"    ip daddr {self.mgmt_cidr} tcp dport 22 accept")
        for r in self.rules:                    # explicit allowlist accepts
            lines.append(
                f"    ip saddr {r.control_host} ip daddr {r.instrument} "
                f"{r.proto} dport {r.port} accept"
            )
        lines += ["    counter drop", "  }", "}"]
        return "\n".join(lines)

    def apply(self) -> None:
        """Atomically replace the table with the rendered ruleset (idempotent)."""
        ruleset = f"flush table {TABLE}\n{self.render()}\n"
        try:
            subprocess.run(["nft", "-f", "-"], input=ruleset, text=True,
                           check=True, capture_output=True)
        except FileNotFoundError as exc:
            raise FirewallError("nft binary not found") from exc
        except subprocess.CalledProcessError as exc:
            raise FirewallError(f"nft apply failed: {exc.stderr.strip()}") from exc
        logger.info("applied %d allow rules to %s", len(self.rules), TABLE)

    def active_tuples(self) -> frozenset[tuple[str, str, str, int]]:
        """Parse the live ruleset and extract enforced accept-tuples."""
        try:
            out = subprocess.run(["nft", "-j", "list", "table", TABLE],
                                 check=True, text=True, capture_output=True).stdout
        except subprocess.CalledProcessError as exc:
            raise FirewallError(f"table {TABLE} not present: {exc.stderr}") from exc
        live: set[tuple[str, str, str, int]] = set()
        for obj in json.loads(out).get("nftables", []):
            expr = obj.get("rule", {}).get("expr", [])
            saddr = daddr = proto = None
            port: int | None = None
            accept = any("accept" in e for e in expr)
            for e in expr:
                match = e.get("match", {})
                left = match.get("left", {}).get("payload", {})
                if left.get("field") == "saddr":
                    saddr = match["right"]
                elif left.get("field") == "daddr":
                    daddr = match["right"]
                elif left.get("field") == "dport":
                    proto = left.get("protocol")
                    port = match["right"]
            if accept and saddr and daddr and port is not None:
                live.add((saddr, daddr, proto or "tcp", int(port)))
        return frozenset(live)

    def drift(self) -> tuple[frozenset, frozenset]:
        """Return (missing, unexpected) tuples: desired minus active, and extra."""
        active = self.active_tuples()
        desired = self.desired_tuples()
        return (desired - active, active - desired)

    def reconcile(self) -> bool:
        """Re-assert the policy if the kernel has drifted. True if repaired."""
        missing, unexpected = self.drift()
        if missing or unexpected:
            logger.warning("drift detected: missing=%s unexpected=%s",
                           sorted(missing), sorted(unexpected))
            self.apply()
            return True
        return False

Validation Against a Test Chain and a Denied Flow

Never validate a fail-closed policy by trusting that apply() returned zero. Prove that a tuple outside the allowlist is actually dropped, and that a tuple inside it passes. The cleanest way is to apply the policy, then assert on the kernel: a denied flow must fail to connect within a short timeout, while an allowed flow to a live instrument returns its *IDN? string. Pair a live check with a drift assertion so the test also confirms reconcile() is a no-op on a clean host.

import socket
import pytest


def _reaches(host: str, port: int, timeout: float = 1.0) -> bool:
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


def test_default_deny_blocks_unlisted_egress(policy: FirewallPolicy) -> None:
    """A destination not in the allowlist must not be reachable once applied."""
    policy.apply()
    # 1.1.1.1:443 is deliberately absent from A -> must be dropped
    assert not _reaches("1.1.1.1", 443), "default-deny egress leaked"
    # counter on the drop rule should be advancing
    out = subprocess.run(["nft", "list", "table", TABLE],
                         text=True, capture_output=True).stdout
    assert "counter" in out and "drop" in out


def test_allowlisted_instrument_is_reachable(policy: FirewallPolicy) -> None:
    """A permitted control-host -> instrument tuple survives the filter."""
    policy.apply()
    rule = policy.rules[0]
    assert _reaches(str(rule.instrument), rule.port), "allowlist blocked a valid flow"
    assert policy.reconcile() is False, "clean host should not report drift"

Between test cases, delete the table with nft delete table inet instrument_guard in a fixture teardown so a failed assertion never leaves a host firewalled. On a live rig, watch the drop counter climb under watch -n1 'nft list table inet instrument_guard' while an instrument’s embedded stack tries and fails to reach a vendor telemetry endpoint — a steadily incrementing counter with no operational impact is the signal the boundary is doing its job. Validated commands crossing that boundary should already conform to Command Set Standardization, and the audit record of every apply and every drift repair belongs alongside the run’s Metadata Injection Workflows so an access-control trail exists for 21 CFR Part 11 review.

Failure Modes & Edge Cases

Four failure modes are specific to compiling and enforcing an instrument firewall from code, and each has a concrete diagnosis:

  • A broad ACCEPT shadowing the DROP. In an ordered chain the first match wins, so a well-meaning oifname "eth0" accept emitted above the allowlist silently defeats default-deny — every packet matches the general rule and the drop is dead code. The compiler here only ever emits narrow accepts and one terminal counter drop, but if you extend render(), keep specific accepts first and never add a general one. Diagnose by reading nft -a list table inet instrument_guard and confirming the drop rule’s counter is nonzero under real traffic; a flatlined counter means something upstream is accepting everything.
  • Rules lost after reboot. nftables is not persistent by default. A host that reboots mid-campaign comes back with an empty table and no egress filter at all — the worst possible failure for a fail-closed design, because it fails open. Render the policy to /etc/nftables.conf and enable nftables.service, and treat reconcile() as a boot-time and periodic task (a systemd timer) so the running kernel is continuously re-driven to $A$. Detect the gap with systemctl is-enabled nftables in your pre-run checklist.
  • DHCP moving an instrument out from under a pinned rule. The allowlist pins a literal IP; if an instrument’s lease renews to a new address, the pinned accept now points at nothing and the device is unreachable while the drop silently blackholes it. Give every instrument a static DHCP reservation or a static IP, and have reconcile() alert when a configured instrument stops answering on its pinned tuple. A sudden VI_ERROR_TMO right after a lease window is the tell; classify it through Error Code Categorization so a moved-IP fault is not retried as a transient timeout.
  • An over-broad deny locking out the management plane. A policy that drops all egress can strand the very SSH session you are editing it from, or cut the host off from NTP and the log collector — a remote host becomes unrecoverable without console access. The mgmt_cidr accept exists precisely to keep the management path open; always render it, and stage risky changes behind a nft rollback timer (apply, sleep, auto-revert unless confirmed) so a mistake self-heals instead of bricking the host.

References

← Back to Security Boundaries & Network Isolation