diff --git a/test/case/infix_firewall/Readme.adoc b/test/case/infix_firewall/Readme.adoc
index e0b8cc21..b20fb711 100644
--- a/test/case/infix_firewall/Readme.adoc
+++ b/test/case/infix_firewall/Readme.adoc
@@ -4,3 +4,7 @@
<<<
include::basic/Readme.adoc[]
+
+<<<
+
+include::lan-wan/Readme.adoc[]
diff --git a/test/case/infix_firewall/all.yaml b/test/case/infix_firewall/all.yaml
index 49f22424..7cf796c6 100644
--- a/test/case/infix_firewall/all.yaml
+++ b/test/case/infix_firewall/all.yaml
@@ -1,3 +1,6 @@
---
- name: Basic Firewall for End Devices
case: basic/test.py
+
+- name: LAN-WAN Firewall with Masquerading
+ case: lan-wan/test.py
diff --git a/test/case/infix_firewall/lan-wan/Readme.adoc b/test/case/infix_firewall/lan-wan/Readme.adoc
new file mode 120000
index 00000000..ae32c841
--- /dev/null
+++ b/test/case/infix_firewall/lan-wan/Readme.adoc
@@ -0,0 +1 @@
+test.adoc
\ No newline at end of file
diff --git a/test/case/infix_firewall/lan-wan/lan-wan.svg b/test/case/infix_firewall/lan-wan/lan-wan.svg
new file mode 100644
index 00000000..0dcda1e8
--- /dev/null
+++ b/test/case/infix_firewall/lan-wan/lan-wan.svg
@@ -0,0 +1,4 @@
+
+
+
+
\ No newline at end of file
diff --git a/test/case/infix_firewall/lan-wan/test.adoc b/test/case/infix_firewall/lan-wan/test.adoc
new file mode 100644
index 00000000..666cd496
--- /dev/null
+++ b/test/case/infix_firewall/lan-wan/test.adoc
@@ -0,0 +1,33 @@
+=== LAN-WAN Firewall with Masquerading
+
+ifdef::topdoc[:imagesdir: {topdoc}../../test/case/infix_firewall/lan-wan]
+
+==== Description
+
+Typical home/office router scenario where the DUT acts as a gateway with
+LAN-to-WAN traffic forwarding and masquerading (SNAT).
+
+image::lan-wan.svg[align=center, scaledwidth=50%]
+
+- DUT/Gateway with firewall and NAT
+- Test host has two interfaces: a LAN-side and a WAN-side (Internet)
+- Test host's LAN interface acts as a client behind the router
+- Test host's WAN interface acts as an Internet server/destination
+
+==== Topology
+
+image::topology.svg[LAN-WAN Firewall with Masquerading topology, align=center, scaledwidth=75%]
+
+==== Sequence
+
+. Set up topology and attach to gateway
+. Configure gateway with firewall and SNAT
+. Verify LAN access to router
+. Verify LAN services accessibility
+. Verify WAN access to router is blocked
+. Verify WAN blocks all well-known ports
+. Verify LAN-to-WAN connectivity (outbound)
+. Verify LAN-to-WAN masquerading
+. Verify WAN-to-LAN blocking (inbound)
+
+
diff --git a/test/case/infix_firewall/lan-wan/test.py b/test/case/infix_firewall/lan-wan/test.py
new file mode 100755
index 00000000..17edf6fa
--- /dev/null
+++ b/test/case/infix_firewall/lan-wan/test.py
@@ -0,0 +1,176 @@
+#!/usr/bin/env python3
+"""LAN-WAN Firewall with Masquerading
+
+Typical home/office router scenario where the DUT acts as a gateway with
+LAN-to-WAN traffic forwarding and masquerading (SNAT).
+
+image::lan-wan.svg[align=center, scaledwidth=50%]
+
+- DUT/Gateway with firewall and NAT
+- Test host has two interfaces: a LAN-side and a WAN-side (Internet)
+- Test host's LAN interface acts as a client behind the router
+- Test host's WAN interface acts as an Internet server/destination
+"""
+
+import time
+import infamy
+from infamy.util import until
+
+
+with infamy.Test() as test:
+ with test.step("Set up topology and attach to gateway"):
+ env = infamy.Env()
+ gateway = env.attach("gateway", "mgmt")
+ _, lan_if = env.ltop.xlate("gateway", "lan")
+ _, wan_if = env.ltop.xlate("gateway", "wan")
+ _, mgmt_if = env.ltop.xlate("gateway", "mgmt")
+ _, host_lan = env.ltop.xlate("host", "lan") # Host LAN-side interface
+ _, host_wan = env.ltop.xlate("host", "wan") # Host WAN-side interface
+
+ LAN_NET = "192.168.1.0/24"
+ LAN_ROUTER_IP = "192.168.1.1" # Router's LAN interface
+ LAN_CLIENT_IP = "192.168.1.100" # Client on LAN side
+
+ WAN_NET = "203.0.113.0/24" # RFC 5737 test network
+ WAN_ROUTER_IP = "203.0.113.1" # Router's WAN interface
+ WAN_SERVER_IP = "203.0.113.100" # Server on WAN side
+
+ with test.step("Configure gateway with firewall and SNAT"):
+ gateway.put_config_dict("ietf-interfaces", {
+ "interfaces": {
+ "interface": [
+ {
+ "name": lan_if,
+ "enabled": True,
+ "ipv4": {
+ "forwarding": True,
+ "address": [{
+ "ip": LAN_ROUTER_IP,
+ "prefix-length": 24
+ }]
+ }
+ },
+ {
+ "name": wan_if,
+ "enabled": True,
+ "ipv4": {
+ "forwarding": True,
+ "address": [{
+ "ip": WAN_ROUTER_IP,
+ "prefix-length": 24
+ }]
+ }
+ }
+ ]
+ }
+ })
+
+ gateway.put_config_dict("infix-firewall", {
+ "firewall": {
+ "default": "wan",
+ "logging": "all",
+ "zone": [
+ {
+ "name": "lan",
+ "description": "Internal LAN network - trusted",
+ "action": "accept",
+ "interface": [lan_if, mgmt_if],
+ "service": ["ssh", "dhcp", "dns"]
+ }, {
+ "name": "wan",
+ "description": "External WAN interface - untrusted",
+ "action": "drop",
+ "interface": [wan_if]
+ }
+ ],
+ "policy": [
+ {
+ "name": "lan-to-wan",
+ "description": "Allow LAN to WAN traffic with SNAT",
+ "ingress": ["lan"],
+ "egress": ["wan"],
+ "action": "accept",
+ "masquerade": True
+ }
+ ]
+ }
+ })
+
+ # Wait for configuration to be activated
+ infamy.Firewall.wait_for_operational(gateway, {
+ "lan": {"action": "accept"},
+ "wan": {"action": "drop"}
+ })
+
+ # Verify firewall operational state
+ data = gateway.get_data("/infix-firewall:firewall")
+ fw = data["firewall"]
+ zones = {z["name"]: z for z in fw["zone"]}
+
+ # Verify LAN zone
+ lan_zone = zones["lan"]
+ assert lan_zone["action"] == "accept"
+ assert lan_if in lan_zone["interface"]
+
+ # Verify WAN zone
+ wan_zone = zones["wan"]
+ assert wan_zone["action"] == "drop"
+ assert wan_if in wan_zone["interface"]
+
+ # Verify policy
+ policies = {p["name"]: p for p in fw["policy"]}
+ lan_wan_policy = policies["lan-to-wan"]
+ assert lan_wan_policy["ingress"] == ["lan"]
+ assert lan_wan_policy["egress"] == ["wan"]
+ assert lan_wan_policy["action"] == "accept"
+ assert lan_wan_policy["masquerade"] is True
+
+ with infamy.IsolatedMacVlan(host_lan) as lan_client:
+ lan_client.addip(LAN_CLIENT_IP)
+ lan_client.addroute("0.0.0.0", LAN_ROUTER_IP, prefix_length="0")
+
+ with infamy.IsolatedMacVlan(host_wan) as wan_server:
+ wan_server.addip(WAN_SERVER_IP)
+
+ with test.step("Verify LAN access to router"):
+ lan_client.must_reach(LAN_ROUTER_IP, timeout=3)
+
+ with test.step("Verify LAN services accessibility"):
+ firewall = infamy.Firewall(lan_client, None)
+ svc = [
+ (22, "tcp", "ssh"),
+ (53, "udp", "dns"),
+ (67, "udp", "dhcp"),
+ ]
+
+ ok, ports = firewall.verify_allowed(LAN_ROUTER_IP, svc)
+ if not ok:
+ print(f" ⚠ Some LAN services are filtered: {', '.join(ports)}")
+ test.fail()
+
+ with test.step("Verify WAN access to router is blocked"):
+ wan_server.must_not_reach(WAN_ROUTER_IP, timeout=3)
+
+ with test.step("Verify WAN blocks all well-known ports"):
+ firewall = infamy.Firewall(wan_server, None)
+
+ ok, ports, _ = firewall.verify_blocked(WAN_ROUTER_IP)
+ if not ok:
+ print(f" ⚠ Some ports are unexpectedly open from WAN: {', '.join(ports)}")
+ test.fail()
+
+ with test.step("Verify LAN-to-WAN connectivity (outbound)"):
+ lan_client.must_reach(WAN_SERVER_IP, timeout=3)
+
+ with test.step("Verify LAN-to-WAN masquerading"):
+ firewall = infamy.Firewall(lan_client, wan_server)
+
+ ok, info = firewall.verify_snat(WAN_SERVER_IP, WAN_ROUTER_IP)
+ if not ok:
+ print(f" ⚠ {info}")
+ test.fail()
+
+ with test.step("Verify WAN-to-LAN blocking (inbound)"):
+ wan_server.must_not_reach(LAN_CLIENT_IP, timeout=3)
+
+ test.succeed()
diff --git a/test/case/infix_firewall/lan-wan/topology.dot b/test/case/infix_firewall/lan-wan/topology.dot
new file mode 100644
index 00000000..8c0588ee
--- /dev/null
+++ b/test/case/infix_firewall/lan-wan/topology.dot
@@ -0,0 +1,24 @@
+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 | lan | wan }",
+ pos="1,1!",
+ requires="controller"
+ ];
+
+ gateway [
+ label="{ mgmt | lan | wan } | gateway",
+ pos="3,1!",
+ requires="infix",
+ ];
+
+ host:mgmt -- gateway:mgmt [requires="mgmt", color="lightgray"]
+ host:lan -- gateway:lan [color=black, fontcolor=black, taillabel="192.168.1.0/24"]
+ host:wan -- gateway:wan [color=red, fontcolor=red, taillabel="203.0.113.0/24"]
+}
diff --git a/test/case/infix_firewall/lan-wan/topology.svg b/test/case/infix_firewall/lan-wan/topology.svg
new file mode 100644
index 00000000..e5d25cfa
--- /dev/null
+++ b/test/case/infix_firewall/lan-wan/topology.svg
@@ -0,0 +1,53 @@
+
+
+
+
+
diff --git a/test/infamy/__init__.py b/test/infamy/__init__.py
index 7d2f56e4..f5397d93 100644
--- a/test/infamy/__init__.py
+++ b/test/infamy/__init__.py
@@ -4,6 +4,7 @@ from .container import Container
from .env import Env
from .env import ArgumentParser
from .env import test_argument
+from .firewall import Firewall
from .furl import Furl
from .netns import IsolatedMacVlan,IsolatedMacVlans
from .portscanner import PortScanner
diff --git a/test/infamy/firewall.py b/test/infamy/firewall.py
new file mode 100644
index 00000000..784cd8e1
--- /dev/null
+++ b/test/infamy/firewall.py
@@ -0,0 +1,189 @@
+"""
+Firewall testing utilities
+
+Provides a helper class and supporting tools to validate firewall
+behavior in automated tests. Supports:
+
+- SNAT verification by inspecting captured ICMP traffic
+- Zone policy checks using targeted port scans
+- Positive and negative policy validation (allowed vs. blocked ports)
+"""
+import time
+from typing import Tuple, List
+from .sniffer import Sniffer
+from .portscanner import PortScanner
+from .util import until
+
+
+class Firewall:
+ """Specialized utilities for testing firewall functionality"""
+
+ def __init__(self, source=None, dest=None):
+ """
+ Initialize firewall tester
+ Args:
+ source: Source network namespace (for traffic generation)
+ dest: Destination network namespace (for traffic capture)
+ """
+ self.srcns = source
+ self.dstns = dest
+
+ @staticmethod
+ def wait_for_operational(target, expected_zones, timeout=30):
+ """Wait for firewall config to be activated/available in operational"""
+ def check_operational():
+ try:
+ oper = target.get_data("/infix-firewall:firewall")
+ if not oper or "firewall" not in oper:
+ return False
+
+ if "zone" not in oper["firewall"]:
+ return False
+
+ zones = {z["name"]: z for z in oper["firewall"]["zone"]}
+
+ for zone_name, expected in expected_zones.items():
+ if zone_name not in zones:
+ return False
+ for key, value in expected.items():
+ if zones[zone_name].get(key) != value:
+ return False
+ return True
+ except:
+ return False
+
+ until(check_operational, attempts=timeout)
+
+ def verify_snat(self, dest_ip: str, snat_ip: str,
+ timeout: int = 3) -> Tuple[bool, str]:
+ """
+ Verify SNAT (masquerading) by analyzing source IP of ICMP traffic
+
+ Args:
+ dest_ip: Destination IP address to ping
+ snat_ip: Expected source IP after SNAT (router's WAN IP)
+ timeout: Test timeout in seconds
+
+ Returns:
+ Tuple of (snat_working: bool, details: str)
+ """
+
+ try:
+ sniffer = Sniffer(self.dstns, "icmp")
+ with sniffer:
+ time.sleep(0.5)
+ self.srcns.runsh(f"ping -c3 -W{timeout} {dest_ip}")
+ time.sleep(0.5)
+
+ rc = sniffer.output()
+ packets = rc.stdout
+ if rc.returncode or not packets.strip():
+ return False, "No packets captured — routing may be broken"
+
+ lines = packets.strip().split('\n')
+ snat_ip_found = False
+ lan_ip_found = False
+
+ for line in lines:
+ if not line.strip():
+ continue
+
+ # Check if we see the expected SNAT IP as source
+ if f"{snat_ip} > {dest_ip}" in line:
+ snat_ip_found = True
+
+ # Check if we see any other source IP (SNAT not working)
+ if f"> {dest_ip}" in line and snat_ip not in line:
+ parts = line.split()
+ for part in parts:
+ if f"> {dest_ip}" in part:
+ src_ip = part.split('>')[0].strip()
+ if '.' in src_ip and src_ip != snat_ip:
+ lan_ip_found = True
+ break
+
+ if snat_ip_found and not lan_ip_found:
+ return True, f"SNAT working: only traffic from {snat_ip}"
+ if lan_ip_found and not snat_ip_found:
+ return False, f"SNAT broken: LAN IPs visible, no {snat_ip}"
+ if snat_ip_found and lan_ip_found:
+ return False, f"SNAT broken: both {snat_ip} and LAN IPs on WAN"
+
+ return False, f"Unclear SNAT status, see capture:\n{packets}"
+
+ except Exception as e:
+ return False, f"SNAT verification failed with error: {e}"
+
+ def verify_blocked(self, dest_ip: str, ports: List[Tuple[int, str, str]] = None,
+ exempt: List[int] = None, timeout: int = 3) -> Tuple[bool, List[str], List[str]]:
+ """
+ Verify specified ports are blocked, with optional exceptions
+ Args:
+ dest_ip: Target hostname or IP address
+ ports: List of port tuples, defaults to
+ PortScanner.WELL_KNOWN_PORTS
+ exempt: List of ports that should be excempt
+ timeout: Connection timeout per port
+ Returns:
+ When exempt=None: Tuple of (all_blocked: bool, open_ports: List[str], [])
+ When exempt=[...]: Tuple of (policy_correct: bool, unexpected_open: List[str],
+ unexpected_filtered_allowed: List[str])
+ """
+ if ports is None:
+ ports = PortScanner.WELL_KNOWN_PORTS
+
+ scanner = PortScanner(self.srcns)
+ results = scanner.scan_ports(dest_ip, ports, timeout)
+
+ if exempt is None:
+ # Simple "all blocked" behavior - only "open" is bad
+ open_ports = []
+ for port, name, result in results:
+ if result["status"] == "open":
+ open_ports.append(f"{name}({port})")
+ return len(open_ports) == 0, open_ports, []
+
+ unexpected_open = []
+ unexpected_filtered_allowed = []
+
+ for port, name, result in results:
+ if port in exempt:
+ # This port should be allowed (not filtered by firewall)
+ status = result["status"]
+ if status in ["filtered", "open|filtered", "closed|filtered"]:
+ unexpected_filtered_allowed.append(f"{name}({port})")
+ else:
+ # This port should be blocked - only "open" is bad
+ if result["status"] == "open":
+ unexpected_open.append(f"{name}({port})")
+
+ policy_correct = (len(unexpected_open) == 0 and
+ len(unexpected_filtered_allowed) == 0)
+ return policy_correct, unexpected_open, unexpected_filtered_allowed
+
+ def verify_allowed(self, dest_ip: str, ports: List[Tuple[int, str, str]] = None,
+ timeout: int = 3) -> Tuple[bool, List[str]]:
+ """
+ Verify specified ports are allowed (open or closed, not filtered)
+ Args:
+ dest_ip: Target hostname or IP address
+ ports: List of port tuples, defaults to
+ PortScanner.WELL_KNOWN_PORTS
+ timeout: Connection timeout per port
+ Returns:
+ Tuple of (all_allowed: bool, filtered_ports: List[str])
+ """
+ if ports is None:
+ ports = PortScanner.WELL_KNOWN_PORTS
+
+ scanner = PortScanner(self.srcns)
+ results = scanner.scan_ports(dest_ip, ports, timeout)
+ filtered_ports = []
+
+ for port, name, result in results:
+ status = result["status"]
+ # Consider any form of filtering as "not allowed"
+ if status in ["filtered", "open|filtered", "closed|filtered"]:
+ filtered_ports.append(f"{name}({port})")
+
+ return len(filtered_ports) == 0, filtered_ports