diff --git a/test/case/all.yaml b/test/case/all.yaml
index 2f5b728d..b3b02552 100644
--- a/test/case/all.yaml
+++ b/test/case/all.yaml
@@ -34,6 +34,9 @@
- name: "IETF Routing"
suite: ietf_routing/all.yaml
+- name: "Infix Firewall"
+ suite: infix_firewall/all.yaml
+
- name: "Infix Containers"
suite: infix_containers/all.yaml
diff --git a/test/case/infix_firewall/Readme.adoc b/test/case/infix_firewall/Readme.adoc
new file mode 100644
index 00000000..e0b8cc21
--- /dev/null
+++ b/test/case/infix_firewall/Readme.adoc
@@ -0,0 +1,6 @@
+:testgroup:
+== infix-firewall
+
+<<<
+
+include::basic/Readme.adoc[]
diff --git a/test/case/infix_firewall/all.yaml b/test/case/infix_firewall/all.yaml
new file mode 100644
index 00000000..49f22424
--- /dev/null
+++ b/test/case/infix_firewall/all.yaml
@@ -0,0 +1,3 @@
+---
+- name: Basic Firewall for End Devices
+ case: basic/test.py
diff --git a/test/case/infix_firewall/basic/Readme.adoc b/test/case/infix_firewall/basic/Readme.adoc
new file mode 120000
index 00000000..ae32c841
--- /dev/null
+++ b/test/case/infix_firewall/basic/Readme.adoc
@@ -0,0 +1 @@
+test.adoc
\ No newline at end of file
diff --git a/test/case/infix_firewall/basic/basic.svg b/test/case/infix_firewall/basic/basic.svg
new file mode 100644
index 00000000..07180716
--- /dev/null
+++ b/test/case/infix_firewall/basic/basic.svg
@@ -0,0 +1,4 @@
+
+
+
+
\ No newline at end of file
diff --git a/test/case/infix_firewall/basic/test.adoc b/test/case/infix_firewall/basic/test.adoc
new file mode 100644
index 00000000..bf6aef8b
--- /dev/null
+++ b/test/case/infix_firewall/basic/test.adoc
@@ -0,0 +1,31 @@
+=== Basic Firewall for End Devices
+
+ifdef::topdoc[:imagesdir: {topdoc}../../test/case/infix_firewall/basic]
+
+==== Description
+
+Firewall configuration suitable for end devices on untrusted networks.
+
+image::basic.svg[align=center, scaledwidth=50%]
+
+- Single zone configuration, "public", with action=drop
+- Allowed services: SSH (port 22), DHCPv6-client, mySSH (custom, port 222)
+- All other ports (HTTP, HTTPS, Telnet, etc.) blocked
+- Verifies unused interfaces automatically assigned to default zone
+
+==== Topology
+
+image::topology.svg[Basic Firewall for End Devices topology, align=center, scaledwidth=75%]
+
+==== Sequence
+
+. Set up topology and attach to target
+. Configure basic end-device firewall
+. Verify unused interface assigned to default zone
+. Verify ICMP is dropped
+. Verify ICMPv6 is dropped
+. Verify SSH service is allowed
+. Verify custom mySSH service is allowed
+. Verify other ports are blocked
+
+
diff --git a/test/case/infix_firewall/basic/test.py b/test/case/infix_firewall/basic/test.py
new file mode 100755
index 00000000..c93cc341
--- /dev/null
+++ b/test/case/infix_firewall/basic/test.py
@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+"""Basic Firewall for End Devices
+
+Firewall configuration suitable for end devices on untrusted networks.
+
+image::basic.svg[align=center, scaledwidth=50%]
+
+- Single zone configuration, "public", with action=drop
+- Allowed services: SSH (port 22), DHCPv6-client, mySSH (custom, port 222)
+- All other ports (HTTP, HTTPS, Telnet, etc.) blocked
+- Check that unused interfaces are automatically assigned to default zone
+"""
+
+import time
+import infamy
+from infamy.util import until
+
+
+with infamy.Test() as test:
+ with test.step("Set up topology and attach to target"):
+ env = infamy.Env()
+ target = env.attach("target", "mgmt")
+ _, data_if = env.ltop.xlate("target", "data")
+ _, mgmt_if = env.ltop.xlate("target", "mgmt")
+ _, unused_if = env.ltop.xlate("target", "unused")
+ _, host_data = env.ltop.xlate("host", "data")
+ TARGET_IP = "192.168.1.1"
+ HOST_IP = "192.168.1.42"
+
+ with test.step("Configure basic end-device firewall"):
+ target.put_config_dict("ietf-interfaces", {
+ "interfaces": {
+ "interface": [
+ {
+ "name": data_if,
+ "enabled": True,
+ "ipv4": {
+ "address": [{
+ "ip": TARGET_IP,
+ "prefix-length": 24
+ }]
+ }
+ }
+ ]
+ }
+ })
+
+ target.put_config_dict("infix-firewall", {
+ "firewall": {
+ "default": "public",
+ "logging": "all",
+ "service": [{
+ "name": "mySSH",
+ "port": [{
+ "lower": 222,
+ "proto": "tcp"
+ }]
+ }, {
+ "name": "http",
+ "port": [{
+ "lower": 8080,
+ "proto": "tcp"
+ }]
+ }],
+ "zone": [{
+ "name": "mgmt",
+ "description": "Management network - for test automation",
+ "action": "accept",
+ "interface": [mgmt_if],
+ "service": ["ssh", "netconf", "restconf"]
+ }, {
+ "name": "public",
+ "description": "Public untrusted network",
+ "action": "drop",
+ "interface": [data_if],
+ "service": ["ssh", "dhcpv6-client", "mySSH", "http"]
+ }]
+ }
+ })
+
+ # Wait for configuration to be activated
+ infamy.Firewall.wait_for_operational(target, {
+ "public": {"action": "drop"},
+ "mgmt": {"action": "accept"}
+ })
+
+ # Verify firewall operational state
+ data = target.get_data("/infix-firewall:firewall")
+ fw = data["firewall"]
+
+ assert fw["default"] == "public"
+
+ services = {svc["name"]: svc for svc in fw.get("service", [])}
+ assert "mySSH" in services, "Custom service mySSH not found"
+ custom_service = services["mySSH"]
+ assert len(custom_service["port"]) == 1
+ port_entry = next(iter(custom_service["port"]))
+ assert port_entry["proto"] == "tcp"
+ assert int(port_entry["lower"]) == 222
+
+ assert "http" in services, "HTTP service override not found"
+ http_service = services["http"]
+ assert len(http_service["port"]) == 1
+ port_entry = next(iter(http_service["port"]))
+ assert port_entry["proto"] == "tcp"
+ assert int(port_entry["lower"]) == 8080
+
+ zones = {zone["name"]: zone for zone in fw["zone"]}
+ assert "public" in zones, "Public zone not found in configuration"
+ public_zone = zones["public"]
+ assert public_zone["action"] == "drop"
+ assert data_if in public_zone["interface"]
+ assert "ssh" in public_zone["service"]
+ assert "dhcpv6-client" in public_zone["service"]
+ assert "mySSH" in public_zone["service"]
+ assert "http" in public_zone["service"]
+
+ with test.step("Verify unused interface assigned to default zone"):
+ data = target.get_data("/infix-firewall:firewall")
+ fw = data["firewall"]
+
+ assert fw["default"] == "public", "Default zone should be 'public'"
+
+ zones = {zone["name"]: zone for zone in fw["zone"]}
+ public_zone = zones["public"]
+
+ assert unused_if in public_zone["interface"], \
+ f"Unused interface {unused_if} should be in default zone 'public', got interfaces: {public_zone['interface']}"
+
+ with infamy.IsolatedMacVlan(host_data) as ns:
+ ns.addip(HOST_IP)
+
+ with test.step("Verify ICMP is dropped"):
+ ns.must_not_reach(TARGET_IP, timeout=2)
+
+ with test.step("Verify ICMPv6 is dropped"):
+ ns.must_not_reach("fe80::1%iface", timeout=2)
+
+ with test.step("Verify SSH service is allowed"):
+ scanner = infamy.PortScanner(ns)
+ ssh_result = scanner.scan_port(TARGET_IP, 22, timeout=2)
+ assert ssh_result["status"] in ["open", "closed"], \
+ f"SSH port should be allowed, got: {ssh_result['status']}"
+
+ with test.step("Verify custom mySSH service is allowed"):
+ scanner = infamy.PortScanner(ns)
+ myssh_result = scanner.scan_port(TARGET_IP, 222, timeout=2)
+ assert myssh_result["status"] in ["open", "closed"], \
+ f"mySSH port 222 should be allowed, got: {myssh_result['status']}"
+
+ with test.step("Verify HTTP service override (8080 allowed, 80 blocked)"):
+ scanner = infamy.PortScanner(ns)
+
+ # Custom HTTP on port 8080 should be allowed
+ http_custom_result = scanner.scan_port(TARGET_IP, 8080, timeout=2)
+ assert http_custom_result["status"] in ["open", "closed"], \
+ f"Custom HTTP port 8080 should be allowed, got: {http_custom_result['status']}"
+
+ # Built-in HTTP on port 80 should be blocked (filtered)
+ http_builtin_result = scanner.scan_port(TARGET_IP, 80, timeout=2)
+ assert http_builtin_result["status"] == "filtered", \
+ f"Built-in HTTP port 80 should be blocked, got: {http_builtin_result['status']}"
+
+ with test.step("Verify other ports are blocked"):
+ firewall = infamy.Firewall(ns, None)
+ allowed = [22, 222, 8080]
+
+ ok, open_ports, filtered = \
+ firewall.verify_blocked(TARGET_IP, exempt=allowed)
+ if not ok:
+ if open_ports:
+ print(f"Unexpected open ports: {', '.join(open_ports)}")
+ if filtered:
+ print(f"Unexpected, filtered ports: {', '.join(filtered)}")
+ test.fail()
+
+ test.succeed()
diff --git a/test/case/infix_firewall/basic/topology.dot b/test/case/infix_firewall/basic/topology.dot
new file mode 100644
index 00000000..c75e3bbf
--- /dev/null
+++ b/test/case/infix_firewall/basic/topology.dot
@@ -0,0 +1,30 @@
+graph "1x3" {
+ layout = "neato";
+ overlap = false;
+ esep = "+80";
+
+ node [shape=record, fontname="DejaVu Sans Mono, Book"];
+ edge [color="cornflowerblue", penwidth="2", fontname="DejaVu Serif, Book"];
+
+ host [
+ label="host | { mgmt | data }",
+ pos="1,1!",
+ requires="controller"
+ ];
+
+ target [
+ label="{ mgmt | data | unused } | target",
+ pos="3,1!",
+ requires="infix",
+ ];
+
+ dummy [
+ label="{ link } | dummy",
+ pos="5,1!",
+ requires="infix",
+ ];
+
+ host:mgmt -- target:mgmt [requires="mgmt", color="lightgray"]
+ host:data -- target:data [color=black, fontcolor=black, taillabel="192.168.1.42/24"]
+ target:unused -- dummy:link [color="gray", style="dashed"]
+}
diff --git a/test/case/infix_firewall/basic/topology.svg b/test/case/infix_firewall/basic/topology.svg
new file mode 100644
index 00000000..644d700d
--- /dev/null
+++ b/test/case/infix_firewall/basic/topology.svg
@@ -0,0 +1,45 @@
+
+
+
+
+
diff --git a/test/infamy/__init__.py b/test/infamy/__init__.py
index 4c44ec8c..7d2f56e4 100644
--- a/test/infamy/__init__.py
+++ b/test/infamy/__init__.py
@@ -6,6 +6,7 @@ from .env import ArgumentParser
from .env import test_argument
from .furl import Furl
from .netns import IsolatedMacVlan,IsolatedMacVlans
+from .portscanner import PortScanner
from .sniffer import Sniffer
from .tap import Test
from .util import parallel, until
diff --git a/test/infamy/portscanner.py b/test/infamy/portscanner.py
new file mode 100644
index 00000000..26f56349
--- /dev/null
+++ b/test/infamy/portscanner.py
@@ -0,0 +1,153 @@
+"""
+Port scanning utilities
+
+Lightweight wrapper around nmap for automated firewall testing.
+Supports:
+
+- Scanning individual TCP/UDP ports with detailed state detection
+- Parallel scanning of multiple ports using threads
+- Predefined set of common service ports for convenience
+"""
+import threading
+from typing import List, Dict, Union, Tuple
+
+
+class PortScanner:
+ """Simple port scanner using netcat for firewall testing"""
+
+ # Well-known ports for testing common services
+ WELL_KNOWN_PORTS = [
+ (22, "tcp", "ssh"),
+ (53, "udp", "dns"),
+ (67, "udp", "dhcp"),
+ (69, "udp", "tftp"),
+ (80, "tcp", "http"),
+ (443, "tcp", "https"),
+ (5353, "udp", "mdns"),
+ (7681, "tcp", "ttyd"),
+ (8080, "tcp", "http-alt"),
+ (8443, "tcp", "https-alt"),
+ (7, "tcp", "echo"),
+ (1234, "tcp", "test-mid"),
+ (9999, "tcp", "test-high"),
+ ]
+
+ def __init__(self, netns=None):
+ """
+ Initialize port scanner
+ Args:
+ netns: Network namespace object (IsolatedMacVlan) or netns name
+ """
+ self.netns = netns
+
+ def scan_port(self, host: str, port: int, protocol: str = "tcp",
+ timeout: int = 3) -> Dict[str, Union[str, bool]]:
+ """
+ Scan a single port using nmap for accurate firewall state detection
+ Args:
+ host: Target hostname or IP address
+ port: Port number to scan
+ protocol: 'tcp' or 'udp'
+ timeout: Connection timeout in seconds
+ Returns:
+ Dict with keys: 'open' (bool), 'status' (str), 'response' (str)
+ """
+ # Build optimized nmap command based on protocol
+ if protocol.lower() == "tcp":
+ proto = "-sT"
+ elif protocol.lower() == "udp":
+ proto = "-sU"
+ else:
+ raise ValueError(f"Unsupported protocol: {protocol}")
+
+ cmd = f"nmap -n {proto} -Pn -p {port} --host-timeout={timeout} " \
+ f"--min-rate=1000 --max-retries=1 --disable-arp-ping {host}"
+ result = self.netns.runsh(cmd)
+
+ # Parse nmap output to determine port state
+ output = result.stdout
+
+ # Look for the specific port line in nmap output
+ # Format: "PORT STATE SERVICE" or "22/tcp open ssh"
+ port_line = None
+ for line in output.split('\n'):
+ if f"{port}/" in line and protocol in line:
+ port_line = line.strip()
+ break
+
+ if port_line:
+ if "open|filtered" in port_line:
+ # No ICMP port unreachable, likely firewall dropping silently
+ status = "open|filtered"
+ is_open = False
+ elif "closed|filtered" in port_line:
+ # No service running, or firewall dropping
+ status = "closed|filtered"
+ is_open = False
+ elif "open" in port_line:
+ status = "open"
+ is_open = True
+ elif "filtered" in port_line:
+ # Blocked/Rejected by firewall
+ status = "filtered"
+ is_open = False
+ elif "closed" in port_line:
+ # Port reachable but service not running
+ status = "closed"
+ is_open = False
+ else:
+ # Unknown state
+ status = "unknown"
+ is_open = False
+ else:
+ # No port line found - likely filtered or error
+ status = "filtered"
+ is_open = False
+
+ return {
+ "open": is_open,
+ "status": status,
+ "response": output.strip()
+ }
+
+ def scan_ports(self, host: str,
+ port_specs: List[Tuple[int, str, str]],
+ timeout: int = 3) -> List[Tuple[int, str, Dict]]:
+ """
+ Scan multiple ports in parallel using threads
+ Args:
+ host: Target hostname or IP address
+ port_specs: List (port, protocol, name) e.g.
+ [(80, "tcp", "http"), (53, "udp", "dns")]
+ timeout: Connection timeout per port
+ Returns:
+ List of (port, name, result) scan results.
+ """
+ results = []
+ threads = []
+ lock = threading.Lock()
+
+ def scan_worker(port: int, protocol: str, name: str):
+ try:
+ result = self.scan_port(host, port, protocol, timeout)
+ with lock:
+ results.append((port, name, result))
+ except Exception as e:
+ with lock:
+ results.append((port, name, {
+ "open": False,
+ "status": "error",
+ "response": str(e)
+ }))
+
+ for port, protocol, name in port_specs:
+ thread = threading.Thread(target=scan_worker, args=(port, protocol, name))
+ threads.append(thread)
+ thread.start()
+
+ for thread in threads:
+ thread.join()
+
+ # Sort results by port number for consistent output
+ results.sort(key=lambda x: x[0])
+ return results
diff --git a/test/spec/Readme.adoc.in b/test/spec/Readme.adoc.in
index 90a9cf4b..c1c1d9cf 100644
--- a/test/spec/Readme.adoc.in
+++ b/test/spec/Readme.adoc.in
@@ -40,6 +40,10 @@ include::../case/ietf_routing/Readme.adoc[]
<<<
+include::../case/infix_firewall/Readme.adoc[]
+
+<<<
+
include::../case/infix_containers/Readme.adoc[]
<<<