nixos-router/tests/vlans.nix

# VLANs and zones: one trunk carrying an untagged LAN plus two VLANs, two of
# the three LANs sharing a zone.
#
#   zone trusted: main (untagged, internet)
#   zone iot:     iot (VLAN 20) + cams (VLAN 30), no internet, no management
#
#   rules: trusted -> iot (anything)
#          iot -> trusted, tcp 8080, only to 192.168.10.100
#          iot -> trusted, tcp 9090, only from source ports 40000-40010
#          "lab" (no interface yet) -> trusted: matches nothing
#
#   nat:   trusted -> router:7070 is forwarded to iot:8081 (source kept)
#          trusted -> iot tcp 8082 is masqueraded (iot sees the router)
#          trusted -> iot tcp 8083 is source-translated to 192.168.20.1
#
#   upstream (ISP) -- vlan 1 -- router -- vlan 2 (trunk) -- main / iot / cams
{ self, pkgs }:
let
  inherit (pkgs) lib;
  common = import ./common.nix;

  # A host on the trunk, either untagged (id = null) or on a VLAN.
  host =
    id:
    { lib, ... }:
    let
      dev = if id == null then "eth1" else "eth1.${toString id}";
    in
    lib.mkMerge [
      {
        virtualisation.vlans = [ 2 ];
        networking.useNetworkd = true;
        networking.useDHCP = false;
        networking.firewall.enable = false;
        networking.interfaces.eth1.ipv4.addresses = lib.mkForce [ ];
        systemd.network.networks."20-lan" = {
          matchConfig.Name = dev;
          networkConfig.DHCP = "ipv4";
        };
      }
      (lib.mkIf (id != null) {
        systemd.network.netdevs."10-vlan" = {
          netdevConfig = {
            Kind = "vlan";
            Name = dev;
          };
          vlanConfig.Id = id;
        };
        systemd.network.networks."10-trunk" = {
          matchConfig.Name = "eth1";
          networkConfig = {
            VLAN = [ dev ];
            LinkLocalAddressing = "no";
          };
        };
      })
    ];
in
pkgs.testers.runNixOSTest {
  name = "router-vlans";

  defaults.imports = [ common.tools ];
  interactive.defaults = common.interactiveDefaults;

  nodes = {
    # The ISP: one DHCP lease for the router, DNS, and a page echoing the
    # address it sees (to prove NAT).
    upstream =
      { lib, ... }:
      {
        virtualisation.vlans = [ 1 ];
        networking.useNetworkd = true;
        networking.useDHCP = false;
        networking.firewall.enable = false;
        networking.interfaces.eth1.ipv4.addresses = lib.mkForce [ ];
        systemd.network.networks."10-wan" = {
          matchConfig.Name = "eth1";
          address = [ "203.0.113.1/24" ];
        };

        services.dnsmasq = {
          enable = true;
          settings = {
            interface = "eth1";
            bind-interfaces = true;
            dhcp-range = "203.0.113.100,203.0.113.100,255.255.255.0,1h";
            dhcp-option = [ "option:router,203.0.113.1" ];
            address = "/example.test/203.0.113.1";
          };
        };
        services.nginx = {
          enable = true;
          virtualHosts."example.test".locations."/".return = ''200 "$remote_addr"'';
        };
      };

    router =
      { lib, ... }:
      {
        imports = [ self.nixosModules.router ];
        virtualisation.vlans = [
          1
          2
        ];
        networking.interfaces.eth1.ipv4.addresses = lib.mkForce [ ];
        networking.interfaces.eth2.ipv4.addresses = lib.mkForce [ ];

        router = {
          enable = true;
          wan.interface = "eth1";
          lans = {
            main = {
              interface = "eth2";
              zone = "trusted";
              internet = true;
              ipv4.address = "192.168.10.1";
              dhcp.pool = {
                start = "192.168.10.100";
                end = "192.168.10.199";
              };
            };
            iot = {
              interface = "eth2";
              vlan = 20;
              zone = "iot";
              allowManagement = false;
              ipv4.address = "192.168.20.1";
              dhcp.pool = {
                start = "192.168.20.100";
                end = "192.168.20.199";
              };
            };
            cams = {
              interface = "eth2";
              vlan = 30;
              zone = "iot";
              allowManagement = false;
              ipv4.address = "192.168.30.1";
              dhcp.pool = {
                start = "192.168.30.100";
                end = "192.168.30.199";
              };
            };
          };
          firewall.rules = [
            {
              from = "trusted";
              to = "iot";
            }
            {
              from = "iot";
              to = "trusted";
              protocol = "tcp";
              ports = [ 8080 ];
              destination = "192.168.10.100";
            }
            {
              from = "iot";
              to = "trusted";
              protocol = "tcp";
              ports = [ 9090 ];
              sourcePorts = [ "40000-40010" ];
              destination = "192.168.10.100";
            }
            {
              from = "lab";
              to = "trusted";
            }
          ];
          firewall.zones = [ "lab" ];
          firewall.nat = [
            {
              type = "dnat";
              from = "trusted";
              to = "iot";
              protocol = "tcp";
              ports = [ 7070 ];
              toAddress = "192.168.20.100";
              toPort = 8081;
            }
            {
              type = "masquerade";
              from = "trusted";
              to = "iot";
              protocol = "tcp";
              ports = [ 8082 ];
            }
            {
              type = "snat";
              from = "trusted";
              to = "iot";
              protocol = "tcp";
              ports = [ 8083 ];
              toAddress = "192.168.20.1";
            }
          ];
          dns.forwarders = [ "203.0.113.1" ];
        };

        # See basic.nix: relax Unbound for the made-up test domain only.
        services.unbound.settings.server = {
          local-zone = [ "test. transparent" ];
          domain-insecure = [ "example.test" ];
        };
      };

    # Untagged host in the trusted zone; also owns a second address, so the
    # destination restriction in the iot -> trusted rule can be told apart
    # from a zone-wide opening.
    main =
      { lib, ... }:
      {
        imports = [ (host null) ];
        systemd.network.networks."20-lan".address = [ "192.168.10.150/24" ];
        services.nginx = {
          enable = true;
          virtualHosts."ports".listen = [
            {
              addr = "0.0.0.0";
              port = 8080;
            }
            {
              addr = "0.0.0.0";
              port = 9090;
            }
          ];
          virtualHosts."ports".locations."/".return = ''200 "ok"'';
        };
      };

    # Echoes the address it sees, to tell translated connections from plain ones.
    iot =
      { ... }:
      {
        imports = [ (host 20) ];
        services.nginx = {
          enable = true;
          virtualHosts."ports" = {
            listen = map (port: {
              addr = "0.0.0.0";
              inherit port;
            }) [ 8081 8082 8083 ];
            locations."/".return = ''200 "$remote_addr"'';
          };
        };
      };
    cams = host 30;
  };

  testScript = ''
    start_all()

    upstream.wait_for_unit("dnsmasq.service")
    upstream.wait_for_unit("nginx.service")
    main.wait_for_unit("nginx.service")
    iot.wait_for_unit("nginx.service")

    with subtest("router is up with VLAN devices on the trunk"):
        router.wait_for_unit("network-online.target")
        router.wait_until_succeeds("ip -4 addr show dev eth1 | grep -q 203.0.113.100")
        router.wait_for_unit("nftables.service")
        router.wait_for_unit("kea-dhcp4-server.service")
        router.wait_for_unit("unbound.service")
        for vid in (20, 30):
            out = router.succeed(f"ip -d link show eth2.{vid}")
            assert f"vlan protocol 802.1Q id {vid}" in out, out
        router.succeed("ip -4 addr show dev eth2 | grep -q 192.168.10.1")
        router.succeed("ip -4 addr show dev eth2.20 | grep -q 192.168.20.1")
        router.succeed("ip -4 addr show dev eth2.30 | grep -q 192.168.30.1")

    with subtest("each host gets an address and gateway from its own VLAN"):
        main.wait_until_succeeds("ip -4 addr show dev eth1 | grep -q 192.168.10.100")
        iot.wait_until_succeeds("ip -4 addr show dev eth1.20 | grep -q 192.168.20.100")
        cams.wait_until_succeeds("ip -4 addr show dev eth1.30 | grep -q 192.168.30.100")
        main.succeed("ip route show default | grep -q 'via 192.168.10.1'")
        iot.succeed("ip route show default | grep -q 'via 192.168.20.1'")
        cams.succeed("ip route show default | grep -q 'via 192.168.30.1'")

    with subtest("internet is opt-in per LAN"):
        # Positive control: the ISP is reachable from the LAN that opted in.
        out = main.wait_until_succeeds("curl -sf --max-time 5 http://203.0.113.1/ -H 'Host: example.test'")
        assert out.strip() == "203.0.113.100", f"expected NAT to 203.0.113.100, got {out!r}"
        # The VLAN hosts have a default route, so this is the firewall at work.
        iot.fail("curl -sf --max-time 3 http://203.0.113.1/ -H 'Host: example.test'")
        cams.fail("curl -sf --max-time 3 http://203.0.113.1/ -H 'Host: example.test'")

    with subtest("router services per VLAN: DNS yes, SSH only where allowed"):
        iot.succeed("dig +short +time=2 +tries=1 @192.168.20.1 example.test | grep -qx 203.0.113.1")
        main.succeed("nc -z -w 2 192.168.10.1 22")
        iot.fail("nc -z -w 2 192.168.20.1 22")
        cams.fail("nc -z -w 2 192.168.30.1 22")
        iot.fail("nc -z -w 2 192.168.10.1 22")

    with subtest("hosts in one zone reach each other across VLANs"):
        iot.succeed("ping -c 1 -W 2 192.168.30.100")
        cams.succeed("ping -c 1 -W 2 192.168.20.100")

    with subtest("trusted -> iot is allowed, but not the other way round"):
        main.succeed("ping -c 1 -W 2 192.168.20.100")
        main.succeed("ping -c 1 -W 2 192.168.30.100")
        iot.fail("ping -c 1 -W 2 192.168.10.100")

    with subtest("iot -> trusted is limited to tcp 8080 on one host"):
        # Positive controls: the listeners work from inside the trusted zone.
        main.succeed("nc -z -w 2 192.168.10.100 9090")
        main.succeed("nc -z -w 2 192.168.10.150 8080")
        for host in (iot, cams):
            host.succeed("nc -z -w 2 192.168.10.100 8080")
            host.fail("nc -z -w 2 192.168.10.100 9090")
            host.fail("nc -z -w 2 192.168.10.150 8080")

    with subtest("a rule can match the source port too"):
        for host in (iot, cams):
            host.succeed("nc -z -w 2 -p 40005 192.168.10.100 9090")
            host.fail("nc -z -w 2 -p 50000 192.168.10.100 9090")
            host.fail("nc -z -w 2 -p 40005 192.168.10.150 9090")

    with subtest("a port forward from a LAN keeps the source address"):
        out = main.succeed("curl -sf --max-time 5 http://192.168.10.1:7070/").strip()
        assert out == "192.168.10.100", out
        # Not addressed to the router: not forwarded.
        main.fail("curl -sf --max-time 3 http://192.168.10.1:7071/")

    with subtest("masquerade and source NAT rewrite the source between zones"):
        assert main.succeed("curl -sf --max-time 5 http://192.168.20.100:8081/").strip() == "192.168.10.100"
        assert main.succeed("curl -sf --max-time 5 http://192.168.20.100:8082/").strip() == "192.168.20.1"
        assert main.succeed("curl -sf --max-time 5 http://192.168.20.100:8083/").strip() == "192.168.20.1"
        router.succeed("nft list ruleset | grep -q 'snat to 192.168.20.1'")

    with subtest("a zone without interfaces does not break the ruleset"):
        router.succeed("nft list ruleset | grep -q 'sport 40000-40010'")
        router.succeed("systemctl is-active nftables.service")
  '';
}