mirror of
https://github.com/kernelkit/infix.git
synced 2026-07-22 01:13:00 +02:00
232 lines
8.9 KiB
Python
232 lines
8.9 KiB
Python
"""
|
|
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 subprocess
|
|
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
|
|
|
|
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}"
|