peter/infix
1
0
Fork 0
mirror of https://github.com/kernelkit/infix.git synced 2026-08-03 14:23:02 +02:00
Code Issues Packages Projects Releases Wiki Activity

test: new test, lan-wan gateway with snat

Browse Source 
Signed-off-by: Joachim Wiberg <troglobit@gmail.com>
This commit is contained in:
Joachim Wiberg
2025-10-10 15:14:14 +02:00
parent 47c4ddfb7d
commit a81f7c82e9
10 changed files with 488 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/case/infix_firewall/Readme.adoc
+4
View File
@@ -4,3 +4,7 @@
<<<
include::basic/Readme.adoc[]
<<<
include::lan-wan/Readme.adoc[]
test/case/infix_firewall/all.yaml
+3
View File
@@ -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
test/case/infix_firewall/lan-wan/Readme.adoc
Symlink
+1
View File
@@ -0,0 +1 @@
test.adoc
test/case/infix_firewall/lan-wan/lan-wan.svg
+4
View File
File diff suppressed because one or more lines are too long
Side by Side

After

Width:  |  Height:  |  Size: 18 KiB

test/case/infix_firewall/lan-wan/test.adoc
+33
View File
@@ -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)
test/case/infix_firewall/lan-wan/test.py
Executable
+176
View File
@@ -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()
test/case/infix_firewall/lan-wan/topology.dot
+24
View File
@@ -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> mgmt | <lan> lan | <wan> wan }",
pos="1,1!",
requires="controller"
];
gateway [
label="{ <mgmt> mgmt | <lan> lan | <wan> 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"]
}
test/case/infix_firewall/lan-wan/topology.svg
+53
View File
@@ -0,0 +1,53 @@
<?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: 1x3 Pages: 1 -->
<svg width="432pt" height="78pt"
viewBox="0.00 0.00 432.03 78.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 74)">
<title>1x3</title>
<polygon fill="white" stroke="transparent" points="-4,4 -4,-74 428.03,-74 428.03,4 -4,4"/>
<!-- host -->
<g id="node1" class="node">
<title>host</title>
<polygon fill="none" stroke="black" points="0,-0.5 0,-69.5 100,-69.5 100,-0.5 0,-0.5"/>
<text text-anchor="middle" x="25" y="-31.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">host</text>
<polyline fill="none" stroke="black" points="50,-0.5 50,-69.5 "/>
<text text-anchor="middle" x="75" y="-54.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">mgmt</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">lan</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">wan</text>
</g>
<!-- gateway -->
<g id="node2" class="node">
<title>gateway</title>
<polygon fill="none" stroke="black" points="300.03,-0.5 300.03,-69.5 424.03,-69.5 424.03,-0.5 300.03,-0.5"/>
<text text-anchor="middle" x="325.03" y="-54.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">mgmt</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">lan</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">wan</text>
<polyline fill="none" stroke="black" points="350.03,-0.5 350.03,-69.5 "/>
<text text-anchor="middle" x="387.03" y="-31.3" 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,-58C100,-58 300.03,-58 300.03,-58"/>
</g>
<!-- host&#45;&#45;gateway -->
<g id="edge2" class="edge">
<title>host:lan&#45;&#45;gateway:lan</title>
<path fill="none" stroke="black" stroke-width="2" d="M100,-35C100,-35 300.03,-35 300.03,-35"/>
<text text-anchor="middle" x="154.5" y="-38.8" font-family="DejaVu Serif, Book" font-size="14.00">192.168.1.0/24</text>
</g>
<!-- host&#45;&#45;gateway -->
<g id="edge3" class="edge">
<title>host:wan&#45;&#45;gateway:wan</title>
<path fill="none" stroke="red" stroke-width="2" d="M100,-12C100,-12 300.03,-12 300.03,-12"/>
<text text-anchor="middle" x="154.5" y="-15.8" font-family="DejaVu Serif, Book" font-size="14.00" fill="red">203.0.113.0/24</text>
</g>
</g>
</svg>
Side by Side

After

Width:  |  Height:  |  Size: 3.0 KiB

test/infamy/__init__.py
+1
View File
@@ -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
test/infamy/firewall.py
+189
View File
@@ -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