test: new test, wan-dmz-lan firewall with snat and dnat

Signed-off-by: Joachim Wiberg <troglobit@gmail.com>
2026-07-22 01:13:00 +02:00 · 2025-10-10 15:14:14 +02:00
parent a81f7c82e9
commit c069308c27
9 changed files with 460 additions and 0 deletions
@@ -8,3 +8,7 @@ include::basic/Readme.adoc[]
 <<<

 include::lan-wan/Readme.adoc[]
+
+<<<
+
+include::wan-dmz-lan/Readme.adoc[]
@@ -4,3 +4,6 @@

 - name: LAN-WAN Firewall with Masquerading
  case: lan-wan/test.py
+
+- name: WAN-DMZ-LAN Firewall with Port Forwarding
+  case: wan-dmz-lan/test.py
@@ -0,0 +1 @@
+test.adoc
@@ -0,0 +1,34 @@
+=== WAN-DMZ-LAN Firewall with Port Forwarding
+
+ifdef::topdoc[:imagesdir: {topdoc}../../test/case/infix_firewall/wan-dmz-lan]
+
+==== Description
+
+Multi-zone firewall setup with port forwarding (DNAT) to a DMZ server,
+and masquerading (SNAT) of WAN-bound traffic.
+
+image::wan-dmz-lan.svg[align=center, scaledwidth=50%]
+
+- DUT/Gateway with WAN/DMZ/LAN zones and NAT
+- Test host's WAN interface acts as external Internet client
+- Test host's DMZ interface acts as internal server (HTTP on port 80)
+- Test host's LAN interface acts as internal LAN client
+
+==== Topology
+
+image::topology.svg[WAN-DMZ-LAN Firewall with Port Forwarding topology, align=center, scaledwidth=75%]
+
+==== Sequence
+
+. Set up topology and attach to gateway
+. Configure gateway with multi-zone firewall and NAT
+. Verify basic connectivity within zones
+. Verify WAN to DMZ port forwarding (DNAT)
+. Verify LAN to DMZ connectivity
+. Verify DMZ to LAN blocking
+. Verify WAN isolation
+. Verify LAN to WAN connectivity with SNAT
+. Verify DMZ to WAN connectivity with SNAT
+. Verify zone default actions/services
+
+
@@ -0,0 +1,284 @@
+#!/usr/bin/env python3
+"""WAN-DMZ-LAN Firewall with Port Forwarding
+
+Multi-zone firewall setup with port forwarding (DNAT) to a DMZ server,
+and masquerading (SNAT) of WAN-bound traffic.
+
+image::wan-dmz-lan.svg[align=center, scaledwidth=50%]
+
+- DUT/Gateway with WAN/DMZ/LAN zones and NAT
+- Test host's WAN interface acts as external Internet client
+- Test host's DMZ interface acts as internal server (HTTP on port 80)
+- Test host's LAN interface acts as internal LAN client
+"""
+
+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")
+        _, wan_if = env.ltop.xlate("gateway", "wan")
+        _, dmz_if = env.ltop.xlate("gateway", "dmz")
+        _, lan_if = env.ltop.xlate("gateway", "lan")
+        _, mgmt_if = env.ltop.xlate("gateway", "mgmt")
+        _, host_wan = env.ltop.xlate("host", "wan")
+        _, host_dmz = env.ltop.xlate("host", "dmz")
+        _, host_lan = env.ltop.xlate("host", "lan")
+
+        WAN_NET = "203.0.113.0/24"        # RFC 5737 test network
+        WAN_ROUTER_IP = "203.0.113.1"     # Gateway WAN interface
+        WAN_CLIENT_IP = "203.0.113.100"   # Host WAN interface
+
+        DMZ_NET = "10.0.1.0/24"
+        DMZ_ROUTER_IP = "10.0.1.1"        # Gateway DMZ interface
+        DMZ_SERVER_IP = "10.0.1.100"      # Host DMZ interface
+
+        LAN_NET = "192.168.1.0/24"
+        LAN_ROUTER_IP = "192.168.1.1"     # Gateway LAN interface
+        LAN_CLIENT_IP = "192.168.1.100"   # Host LAN interface
+
+    with test.step("Configure gateway with multi-zone firewall and NAT"):
+        gateway.put_config_dict("ietf-interfaces", {
+            "interfaces": {
+                "interface": [
+                    {
+                        "name": wan_if,
+                        "enabled": True,
+                        "ipv4": {
+                            "forwarding": True,
+                            "address": [{
+                                "ip": WAN_ROUTER_IP,
+                                "prefix-length": 24
+                            }]
+                        }
+                    },
+                    {
+                        "name": dmz_if,
+                        "enabled": True,
+                        "ipv4": {
+                            "forwarding": True,
+                            "address": [{
+                                "ip": DMZ_ROUTER_IP,
+                                "prefix-length": 24
+                            }]
+                        }
+                    },
+                    {
+                        "name": lan_if,
+                        "enabled": True,
+                        "ipv4": {
+                            "forwarding": True,
+                            "address": [{
+                                "ip": LAN_ROUTER_IP,
+                                "prefix-length": 24
+                            }]
+                        }
+                    }
+                ]
+            }
+        })
+
+        gateway.put_config_dict("infix-firewall", {
+            "firewall": {
+                "default": "wan",
+                "logging": "all",
+                "zone": [
+                    {
+                        "name": "wan",
+                        "description": "External WAN interface - untrusted",
+                        "action": "drop",
+                        "interface": [wan_if],
+                        "port-forward": [{
+                            "lower": 8080,
+                            "proto": "tcp",
+                            "to": {
+                                "addr": DMZ_SERVER_IP,
+                                "port": 80
+                            }
+                        }]
+                    },
+                    {
+                        "name": "dmz",
+                        "description": "DMZ network - limited trust",
+                        "action": "reject",
+                        "network": [DMZ_NET],
+                        "service": ["http"]
+                    },
+                    {
+                        "name": "lan",
+                        "description": "Internal LAN network - trusted",
+                        "action": "accept",
+                        "interface": [lan_if, mgmt_if]
+                    }
+                ],
+                "policy": [
+                    {
+                        "name": "loc-to-wan",
+                        "description": "Allow local networks to WAN with SNAT",
+                        "ingress": ["lan", "dmz"],
+                        "egress": ["wan"],
+                        "action": "accept",
+                        "masquerade": True
+                    }, {
+                        "name": "lan-to-dmz",
+                        "description": "Allow LAN access to DMZ services",
+                        "ingress": ["lan"],
+                        "egress": ["dmz"],
+                        "action": "accept",
+                        "service": ["ssh", "http"]
+                    }
+                ]
+            }
+        })
+
+        # Wait for configuration to be activated
+        infamy.Firewall.wait_for_operational(gateway, {
+            "wan": {"action": "drop"},
+            "dmz": {"action": "reject"},
+            "lan": {"action": "accept"}
+        })
+
+        # Verify firewall operational state
+        data = gateway.get_data("/infix-firewall:firewall")
+        fw = data["firewall"]
+        zones = {z["name"]: z for z in fw["zone"]}
+
+        # Verify WAN zone with port forwarding
+        wan_zone = zones["wan"]
+        assert wan_zone["action"] == "drop"
+        assert wan_if in wan_zone["interface"]
+        assert len(wan_zone["port-forward"]) == 1
+        pf = next(iter(wan_zone["port-forward"]))
+        assert pf["lower"] == 8080
+        assert pf["to"]["addr"] == DMZ_SERVER_IP
+        assert pf["to"]["port"] == 80
+
+        # Verify DMZ zone
+        dmz_zone = zones["dmz"]
+        assert dmz_zone["action"] == "reject"
+        assert DMZ_NET in dmz_zone["network"]
+        assert "http" in dmz_zone["service"]
+
+        # Verify LAN zone
+        lan_zone = zones["lan"]
+        assert lan_zone["action"] == "accept"
+        assert lan_if in lan_zone["interface"]
+
+        # Check policies
+        policies = {p["name"]: p for p in fw["policy"]}
+
+        # Verify loc-to-wan policy
+        loc_wan_policy = policies["loc-to-wan"]
+        assert set(loc_wan_policy["ingress"]) == {"lan", "dmz"}
+        assert loc_wan_policy["egress"] == ["wan"]
+        assert loc_wan_policy["masquerade"] is True
+
+        # Verify lan-to-dmz policy
+        lan_dmz_policy = policies["lan-to-dmz"]
+        assert lan_dmz_policy["ingress"] == ["lan"]
+        assert lan_dmz_policy["egress"] == ["dmz"]
+        assert "ssh" in lan_dmz_policy["service"]
+        assert "http" in lan_dmz_policy["service"]
+
+    with infamy.IsolatedMacVlan(host_wan) as wan_client:
+        wan_client.addip(WAN_CLIENT_IP)
+
+        with infamy.IsolatedMacVlan(host_dmz) as dmz_server:
+            dmz_server.addip(DMZ_SERVER_IP)
+            dmz_server.addroute("0.0.0.0", DMZ_ROUTER_IP, prefix_length="0")
+
+            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 test.step("Verify basic connectivity within zones"):
+                    lan_client.must_reach(LAN_ROUTER_IP, timeout=3)
+                    dmz_server.must_not_reach(DMZ_ROUTER_IP, timeout=3)
+
+                with test.step("Verify WAN to DMZ port forwarding (DNAT)"):
+                    firewall = infamy.Firewall(wan_client, dmz_server)
+
+                    # Test port forwarding: WAN:8080 → DMZ:80
+                    ok, info = firewall.verify_dnat(
+                        WAN_ROUTER_IP, forward_port=8080, target_port=80)
+
+                    if not ok:
+                        print(f"   ⚠ {info}")
+                        test.fail()
+
+                with test.step("Verify LAN to DMZ connectivity"):
+                    lan_client.must_reach(DMZ_SERVER_IP, timeout=3)
+                    firewall = infamy.Firewall(lan_client, None)
+                    svc = [
+                        (22, "tcp", "ssh"),
+                        (80, "tcp", "http"),
+                    ]
+
+                    ok, ports = firewall.verify_allowed(DMZ_SERVER_IP, svc)
+                    if not ok:
+                        print(f"   ⚠ Some DMZ services filtered from LAN: {', '.join(ports)}")
+                        test.fail()
+
+                with test.step("Verify DMZ to LAN blocking"):
+                    dmz_server.must_not_reach(LAN_CLIENT_IP, timeout=3)
+
+                with test.step("Verify WAN isolation"):
+                    firewall = infamy.Firewall(wan_client, None)
+
+                    ok, ports, _ = firewall.verify_blocked(LAN_ROUTER_IP)
+                    if not ok:
+                        print(f"   ⚠ WAN can access LAN ports: {', '.join(ports)}")
+                        test.fail()
+
+                    ok, ports, _ = firewall.verify_blocked(DMZ_ROUTER_IP)
+                    if not ok:
+                        print(f"   ⚠ WAN can access DMZ ports: {', '.join(ports)}")
+
+                with test.step("Verify LAN to WAN connectivity with SNAT"):
+                    firewall = infamy.Firewall(lan_client, wan_client)
+
+                    lan_client.must_reach(WAN_CLIENT_IP, timeout=3)
+
+                    ok, info = firewall.verify_snat(WAN_CLIENT_IP, WAN_ROUTER_IP)
+                    if not ok:
+                        print(f"   ⚠ LAN to WAN SNAT: {info}")
+                        test.fail()
+
+                with test.step("Verify DMZ to WAN connectivity with SNAT"):
+                    firewall = infamy.Firewall(dmz_server, wan_client)
+
+                    dmz_server.must_reach(WAN_CLIENT_IP, timeout=3)
+
+                    ok, info = firewall.verify_snat(WAN_CLIENT_IP, WAN_ROUTER_IP)
+                    if not ok:
+                        print(f"   ⚠ DMZ to WAN SNAT: {info}")
+                        test.fail()
+
+                with test.step("Verify zone default actions/services"):
+                    firewall_lan = infamy.Firewall(lan_client, None)
+                    firewall_dmz = infamy.Firewall(dmz_server, None)
+                    firewall_wan = infamy.Firewall(wan_client, None)
+
+                    svc = [
+                        (22, "tcp", "ssh"),
+                        (53, "udp", "dns"),
+                        (67, "udp", "dhcp")
+                    ]
+                    ok, ports = firewall_lan.verify_allowed(LAN_ROUTER_IP, svc)
+                    if not ok:
+                        print(f"   ⚠ LAN services not properly accessible: {', '.join(ports)}")
+
+                    svc = [(80, "tcp", "http")]
+                    ok, ports = firewall_dmz.verify_allowed(DMZ_ROUTER_IP, svc)
+                    if not ok:
+                        print(f"   ⚠ DMZ HTTP service not accessible: {', '.join(ports)}")
+
+                    ok, ports, _ = firewall_wan.verify_blocked(WAN_ROUTER_IP)
+                    if not ok:
+                        print(f"   ⚠ WAN has unexpected open ports: {', '.join(ports)}")
+
+    test.succeed()
@@ -0,0 +1,25 @@
+graph "1x4" {
+    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> mgmt | <wan> wan | <dmz> dmz | <lan> lan }",
+        pos="1,1!",
+        requires="controller"  
+    ];
+
+    gateway [
+        label="{ <mgmt> mgmt | <wan> wan | <dmz> dmz | <lan> lan } | gateway",
+        pos="3,1!",
+        requires="infix",
+    ];
+
+    host:mgmt -- gateway:mgmt [requires="mgmt", color="lightgray"]
+    host:wan  -- gateway:wan  [color=red, fontcolor=red, taillabel="203.0.113.0/24"]
+    host:dmz  -- gateway:dmz  [color=orange, fontcolor=orange, taillabel="10.0.1.0/24"]
+    host:lan  -- gateway:lan  [color=black, fontcolor=black, taillabel="192.168.1.0/24"]
+}
@@ -0,0 +1,63 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN"
+ "http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd">
+
+<!-- Title: 1x4 Pages: 1 -->
+<svg width="432pt" height="101pt"
+ viewBox="0.00 0.00 432.03 101.00" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink">
+<g id="graph0" class="graph" transform="scale(1 1) rotate(0) translate(4 97)">
+<title>1x4</title>
+<polygon fill="white" stroke="transparent" points="-4,4 -4,-97 428.03,-97 428.03,4 -4,4"/>
+<!-- host -->
+<g id="node1" class="node">
+<title>host</title>
+<polygon fill="none" stroke="black" points="0,-0.5 0,-92.5 100,-92.5 100,-0.5 0,-0.5"/>
+<text text-anchor="middle" x="25" y="-42.8" font-family="DejaVu Sans Mono, Book" font-size="14.00">host</text>
+<polyline fill="none" stroke="black" points="50,-0.5 50,-92.5 "/>
+<text text-anchor="middle" x="75" y="-77.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">mgmt</text>
+<polyline fill="none" stroke="black" points="50,-69.5 100,-69.5 "/>
+<text text-anchor="middle" x="75" y="-54.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">wan</text>
+<polyline fill="none" stroke="black" points="50,-46.5 100,-46.5 "/>
+<text text-anchor="middle" x="75" y="-31.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">dmz</text>
+<polyline fill="none" stroke="black" points="50,-23.5 100,-23.5 "/>
+<text text-anchor="middle" x="75" y="-8.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">lan</text>
+</g>
+<!-- gateway -->
+<g id="node2" class="node">
+<title>gateway</title>
+<polygon fill="none" stroke="black" points="300.03,-0.5 300.03,-92.5 424.03,-92.5 424.03,-0.5 300.03,-0.5"/>
+<text text-anchor="middle" x="325.03" y="-77.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">mgmt</text>
+<polyline fill="none" stroke="black" points="300.03,-69.5 350.03,-69.5 "/>
+<text text-anchor="middle" x="325.03" y="-54.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">wan</text>
+<polyline fill="none" stroke="black" points="300.03,-46.5 350.03,-46.5 "/>
+<text text-anchor="middle" x="325.03" y="-31.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">dmz</text>
+<polyline fill="none" stroke="black" points="300.03,-23.5 350.03,-23.5 "/>
+<text text-anchor="middle" x="325.03" y="-8.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">lan</text>
+<polyline fill="none" stroke="black" points="350.03,-0.5 350.03,-92.5 "/>
+<text text-anchor="middle" x="387.03" y="-42.8" font-family="DejaVu Sans Mono, Book" font-size="14.00">gateway</text>
+</g>
+<!-- host&#45;&#45;gateway -->
+<g id="edge1" class="edge">
+<title>host:mgmt&#45;&#45;gateway:mgmt</title>
+<path fill="none" stroke="lightgray" stroke-width="2" d="M100,-81.5C100,-81.5 300.03,-81.5 300.03,-81.5"/>
+</g>
+<!-- host&#45;&#45;gateway -->
+<g id="edge2" class="edge">
+<title>host:wan&#45;&#45;gateway:wan</title>
+<path fill="none" stroke="red" stroke-width="2" d="M100,-58.5C100,-58.5 300.03,-58.5 300.03,-58.5"/>
+<text text-anchor="middle" x="154.5" y="-62.3" font-family="DejaVu Serif, Book" font-size="14.00" fill="red">203.0.113.0/24</text>
+</g>
+<!-- host&#45;&#45;gateway -->
+<g id="edge3" class="edge">
+<title>host:dmz&#45;&#45;gateway:dmz</title>
+<path fill="none" stroke="orange" stroke-width="2" d="M100,-34.5C100,-34.5 300.03,-34.5 300.03,-34.5"/>
+<text text-anchor="middle" x="141" y="-38.3" font-family="DejaVu Serif, Book" font-size="14.00" fill="orange">10.0.1.0/24</text>
+</g>
+<!-- host&#45;&#45;gateway -->
+<g id="edge4" class="edge">
+<title>host:lan&#45;&#45;gateway:lan</title>
+<path fill="none" stroke="black" stroke-width="2" d="M100,-11.5C100,-11.5 300.03,-11.5 300.03,-11.5"/>
+<text text-anchor="middle" x="154.5" y="-15.3" font-family="DejaVu Serif, Book" font-size="14.00">192.168.1.0/24</text>
+</g>
+</g>
+</svg>
@@ -8,6 +8,7 @@ behavior in automated tests. Supports:
 - Zone policy checks using targeted port scans
 - Positive and negative policy validation (allowed vs. blocked ports)
 """
+import subprocess
 import time
 from typing import Tuple, List
 from .sniffer import Sniffer
@@ -187,3 +188,44 @@ class Firewall:
                filtered_ports.append(f"{name}({port})")

        return len(filtered_ports) == 0, filtered_ports
+
+    def verify_dnat(self, gateway_ip: str, forward_port: int, target_port: int,
+                    timeout: int = 5) -> Tuple[bool, str]:
+        """
+        Verify DNAT (port forwarding) by testing end-to-end connectivity
+
+        Args:
+            gateway_ip:   Gateway IP where port forwarding is configured
+            forward_port: External port being forwarded (e.g., 8080)
+            target_port:  Internal target port (e.g., 80)
+            timeout:      Connection timeout
+        Returns:
+            Tuple of (dnat_working: bool, details: str)
+        """
+        try:
+            # Use netcat to simulate a simple service on target port
+            cmd = f"nc -l -p {target_port} -e /bin/echo 'DNAT-TEST-OK'"
+            pid = self.dstns.popen(cmd.split(), stdout=subprocess.PIPE,
+                                   stderr=subprocess.PIPE)
+            time.sleep(1)  # Give server time to start
+
+            # Test connection from source to gateway:forward_port
+            cmd = f"nc -w {timeout} {gateway_ip} {forward_port}"
+            result = self.srcns.runsh(cmd)
+
+            try:
+                pid.terminate()
+                pid.wait(timeout=1)
+            except:
+                pid.kill()
+
+            # Check if we got the expected response
+            if "DNAT-TEST-OK" in result.stdout:
+                return True, f"DNAT working: {gateway_ip}:{forward_port} → target:{target_port}"
+            if result.returncode == 0:
+                return True, f"DNAT working: connection successful to {gateway_ip}:{forward_port}"
+
+            return False, f"DNAT failed: no response from {gateway_ip}:{forward_port}"
+
+        except Exception as e:
+            return False, f"DNAT verification failed with error: {e}"