infix/test/case/firewall/basic/test.py

#!/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-untrusted-net", 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")
        _, host_data = env.ltop.xlate("host", "data")
        TARGET_IP = "192.168.1.1"
        HOST_IP = "192.168.1.42"
        # A dummy interface stands in for an unused port: it is not placed
        # in any zone, so it must fall back to the default zone.
        UNUSED_IF = "dummy0"

    with test.step("Configure basic end-device firewall"):
        target.put_config_dicts({
            "ietf-interfaces": {
                "interfaces": {
                    "interface": [
                        {
                            "name": data_if,
                            "enabled": True,
                            "ipv4": {
                                "address": [{
                                    "ip": TARGET_IP,
                                    "prefix-length": 24
                                }]
                            }
                        },
                        {
                            "name": UNUSED_IF,
                            "type": "infix-if-type:dummy",
                            "enabled": True
                        }
                    ]
                }
            },
            "infix-firewall": {
                "firewall": {
                    "default": "public-untrusted-net",
                    "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"]
                    }, {
                        # 20-char name, exceeds old iptables-derived 17-char limit
                        # Verifies we allow long names with nftables, issue #1389
                        "name": "public-untrusted-net",
                        "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-untrusted-net": {"action": "drop"},
            "mgmt": {"action": "accept"}
        })

        # Verify firewall operational state
        data = target.get_data("/infix-firewall:firewall")
        fw = data["firewall"]

        assert fw["default"] == "public-untrusted-net"

        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-untrusted-net" in zones, "public-untrusted-net zone not found in configuration"
        public_zone = zones["public-untrusted-net"]
        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"):
        def unused_in_default_zone():
            data = target.get_data("/infix-firewall:firewall")
            fw = data["firewall"]
            assert fw["default"] == "public-untrusted-net", "Default zone should be 'public-untrusted-net'"
            zones = {zone["name"]: zone for zone in fw["zone"]}
            return UNUSED_IF in zones["public-untrusted-net"].get("interface", [])

        until(unused_in_default_zone, attempts=10)

    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()