From 47c4ddfb7d439140db3b7161eaab40282bb0e653 Mon Sep 17 00:00:00 2001 From: Joachim Wiberg Date: Thu, 21 Aug 2025 11:32:02 +0200 Subject: [PATCH] test: new test, basic firewall zone verification Signed-off-by: Joachim Wiberg --- test/case/all.yaml | 3 + test/case/infix_firewall/Readme.adoc | 6 + test/case/infix_firewall/all.yaml | 3 + test/case/infix_firewall/basic/Readme.adoc | 1 + test/case/infix_firewall/basic/basic.svg | 4 + test/case/infix_firewall/basic/test.adoc | 31 ++++ test/case/infix_firewall/basic/test.py | 177 ++++++++++++++++++++ test/case/infix_firewall/basic/topology.dot | 30 ++++ test/case/infix_firewall/basic/topology.svg | 45 +++++ test/infamy/__init__.py | 1 + test/infamy/portscanner.py | 153 +++++++++++++++++ test/spec/Readme.adoc.in | 4 + 12 files changed, 458 insertions(+) create mode 100644 test/case/infix_firewall/Readme.adoc create mode 100644 test/case/infix_firewall/all.yaml create mode 120000 test/case/infix_firewall/basic/Readme.adoc create mode 100644 test/case/infix_firewall/basic/basic.svg create mode 100644 test/case/infix_firewall/basic/test.adoc create mode 100755 test/case/infix_firewall/basic/test.py create mode 100644 test/case/infix_firewall/basic/topology.dot create mode 100644 test/case/infix_firewall/basic/topology.svg create mode 100644 test/infamy/portscanner.py 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 @@ + + + + + + +1x3 + + + +host + +host + +mgmt + +data + + + +target + +mgmt + +data + +unused + +target + + + +host:mgmt--target:mgmt + + + + +host:data--target:data + +192.168.1.42/24 + + + 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[] <<<