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

test: new test, basic firewall zone verification

Browse Source 
Signed-off-by: Joachim Wiberg <troglobit@gmail.com>
This commit is contained in:
Joachim Wiberg
2025-10-10 15:14:13 +02:00
parent 1735a97dce
commit 47c4ddfb7d
12 changed files with 458 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/case/all.yaml
+3
View File
@@ -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
test/case/infix_firewall/Readme.adoc
+6
View File
@@ -0,0 +1,6 @@
:testgroup:
== infix-firewall
<<<
include::basic/Readme.adoc[]
test/case/infix_firewall/all.yaml
+3
View File
@@ -0,0 +1,3 @@
---
- name: Basic Firewall for End Devices
case: basic/test.py
test/case/infix_firewall/basic/Readme.adoc
Symlink
+1
View File
@@ -0,0 +1 @@
test.adoc
test/case/infix_firewall/basic/basic.svg
+4
View File
File diff suppressed because one or more lines are too long
Side by Side

After

Width:  |  Height:  |  Size: 12 KiB

test/case/infix_firewall/basic/test.adoc
+31
View File
@@ -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
test/case/infix_firewall/basic/test.py
Executable
+177
View File
@@ -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()
test/case/infix_firewall/basic/topology.dot
+30
View File
@@ -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> mgmt | <data> data }",
pos="1,1!",
requires="controller"
];
target [
label="{ <mgmt> mgmt | <data> data | <unused> unused } | target",
pos="3,1!",
requires="infix",
];
dummy [
label="{ <link> 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"]
}
test/case/infix_firewall/basic/topology.svg
+45
View File
@@ -0,0 +1,45 @@
<?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="440pt" height="78pt"
viewBox="0.00 0.00 440.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 436.03,-74 436.03,4 -4,4"/>
<!-- host -->
<g id="node1" class="node">
<title>host</title>
<polygon fill="none" stroke="black" points="0,-12 0,-58 100,-58 100,-12 0,-12"/>
<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,-12 50,-58 "/>
<text text-anchor="middle" x="75" y="-42.8" font-family="DejaVu Sans Mono, Book" font-size="14.00">mgmt</text>
<polyline fill="none" stroke="black" points="50,-35 100,-35 "/>
<text text-anchor="middle" x="75" y="-19.8" font-family="DejaVu Sans Mono, Book" font-size="14.00">data</text>
</g>
<!-- target -->
<g id="node2" class="node">
<title>target</title>
<polygon fill="none" stroke="black" points="300.03,-0.5 300.03,-69.5 432.03,-69.5 432.03,-0.5 300.03,-0.5"/>
<text text-anchor="middle" x="333.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 366.03,-46.5 "/>
<text text-anchor="middle" x="333.03" y="-31.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">data</text>
<polyline fill="none" stroke="black" points="300.03,-23.5 366.03,-23.5 "/>
<text text-anchor="middle" x="333.03" y="-8.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">unused</text>
<polyline fill="none" stroke="black" points="366.03,-0.5 366.03,-69.5 "/>
<text text-anchor="middle" x="399.03" y="-31.3" font-family="DejaVu Sans Mono, Book" font-size="14.00">target</text>
</g>
<!-- host&#45;&#45;target -->
<g id="edge1" class="edge">
<title>host:mgmt&#45;&#45;target:mgmt</title>
<path fill="none" stroke="lightgray" stroke-width="2" d="M100,-47C100,-47 300.03,-58 300.03,-58"/>
</g>
<!-- host&#45;&#45;target -->
<g id="edge2" class="edge">
<title>host:data&#45;&#45;target:data</title>
<path fill="none" stroke="black" stroke-width="2" d="M100,-23C100,-23 300.03,-35 300.03,-35"/>
<text text-anchor="middle" x="159" y="-26.8" font-family="DejaVu Serif, Book" font-size="14.00">192.168.1.42/24</text>
</g>
</g>
</svg>
Side by Side

After

Width:  |  Height:  |  Size: 2.5 KiB

test/infamy/__init__.py
+1
View File
@@ -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
test/infamy/portscanner.py
+153
View File
@@ -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
test/spec/Readme.adoc.in
+4
View File
@@ -40,6 +40,10 @@ include::../case/ietf_routing/Readme.adoc[]
<<<
include::../case/infix_firewall/Readme.adoc[]
<<<
include::../case/infix_containers/Readme.adoc[]
<<<