From a81f7c82e9ed1d414034387e079e4e168ace346b Mon Sep 17 00:00:00 2001 From: Joachim Wiberg Date: Fri, 22 Aug 2025 07:19:25 +0200 Subject: [PATCH] test: new test, lan-wan gateway with snat Signed-off-by: Joachim Wiberg --- test/case/infix_firewall/Readme.adoc | 4 + test/case/infix_firewall/all.yaml | 3 + test/case/infix_firewall/lan-wan/Readme.adoc | 1 + test/case/infix_firewall/lan-wan/lan-wan.svg | 4 + test/case/infix_firewall/lan-wan/test.adoc | 33 +++ test/case/infix_firewall/lan-wan/test.py | 176 ++++++++++++++++ test/case/infix_firewall/lan-wan/topology.dot | 24 +++ test/case/infix_firewall/lan-wan/topology.svg | 53 +++++ test/infamy/__init__.py | 1 + test/infamy/firewall.py | 189 ++++++++++++++++++ 10 files changed, 488 insertions(+) create mode 120000 test/case/infix_firewall/lan-wan/Readme.adoc create mode 100644 test/case/infix_firewall/lan-wan/lan-wan.svg create mode 100644 test/case/infix_firewall/lan-wan/test.adoc create mode 100755 test/case/infix_firewall/lan-wan/test.py create mode 100644 test/case/infix_firewall/lan-wan/topology.dot create mode 100644 test/case/infix_firewall/lan-wan/topology.svg create mode 100644 test/infamy/firewall.py 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 @@ + + + + + + +1x3 + + + +host + +host + +mgmt + +lan + +wan + + + +gateway + +mgmt + +lan + +wan + +gateway + + + +host:mgmt--gateway:mgmt + + + + +host:lan--gateway:lan + +192.168.1.0/24 + + + +host:wan--gateway:wan + +203.0.113.0/24 + + + 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