nixos-router/tests/basic.nix

# Single WAN (DHCP) + single LAN: DHCP, DNS, NAT and the default firewall policy.
#
#   upstream (ISP, 203.0.113.1) -- vlan 1 -- router -- vlan 2 -- client
{ self, pkgs }:
let
  # `fail` in the test script also passes on "command not found", so make sure
  # the tools the script relies on are really there.
  tools = { pkgs, ... }: {
    environment.systemPackages = with pkgs; [ dnsutils netcat-openbsd curl tcpdump ];
  };
in
pkgs.testers.runNixOSTest {
  name = "router-basic";

  defaults.imports = [ tools ];

  nodes = {
    # Plays the ISP: hands out one WAN lease, answers DNS, and serves a page
    # that echoes the client address it sees (to prove NAT happens).
    upstream = { lib, ... }: {
      virtualisation.vlans = [ 1 ];
      networking.useDHCP = false;
      networking.firewall.enable = false;
      networking.interfaces.eth1.ipv4 = {
        addresses = lib.mkForce [ { address = "203.0.113.1"; prefixLength = 24; } ];
        # Route back to the LAN via the router so that "unreachable" below
        # is due to the firewall and not a missing route.
        routes = [ { address = "192.168.10.0"; prefixLength = 24; via = "203.0.113.100"; } ];
      };

      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 ];

      # The test driver preconfigures static addresses on every vlan interface.
      networking.interfaces.eth1.ipv4.addresses = lib.mkForce [ ];
      networking.interfaces.eth2.ipv4.addresses = lib.mkForce [ ];

      router = {
        enable = true;
        wan.interface = "eth1";
        lans.main = {
          interface = "eth2";
          ipv4 = { address = "192.168.10.1"; subnet = "192.168.10.0/24"; };
          dhcp.pool = { start = "192.168.10.100"; end = "192.168.10.199"; };
        };
        dns.forwarders = [ "203.0.113.1" ];
      };

      # Unbound answers `.test` locally (RFC 6761) and would validate DNSSEC
      # against a fake ISP resolver that cannot provide it; neither applies to
      # a real forwarder, so relax both for the made-up test domain only.
      services.unbound.settings.server = {
        local-zone = [ "test. transparent" ];
        domain-insecure = [ "example.test" ];
      };
    };

    client = { lib, ... }: {
      virtualisation.vlans = [ 2 ];
      networking.useNetworkd = true;
      networking.useDHCP = false;
      networking.interfaces.eth1.ipv4.addresses = lib.mkForce [ ];
      systemd.network.networks."10-lan" = {
        matchConfig.Name = "eth1";
        networkConfig.DHCP = "ipv4";
      };
      networking.firewall.enable = false;
    };
  };

  testScript = ''
    start_all()

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

    with subtest("router obtains its WAN lease and brings up services"):
        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")

    with subtest("client gets address, gateway and DNS from the router"):
        client.wait_until_succeeds("ip -4 addr show dev eth1 | grep -q 192.168.10.")
        client.succeed("ip route show default | grep -q 'via 192.168.10.1'")
        client.succeed("resolvectl status eth1 | grep -q 192.168.10.1")

    with subtest("client resolves names through the router"):
        client.wait_until_succeeds("dig +short +time=2 +tries=1 @192.168.10.1 example.test | grep -qx 203.0.113.1")

    with subtest("client reaches the 'internet' and is NATed"):
        out = client.wait_until_succeeds("curl -sf --max-time 5 --resolve example.test:80:203.0.113.1 http://example.test/")
        assert out.strip() == "203.0.113.100", f"expected NATed source 203.0.113.100, got {out!r}"

    with subtest("WAN cannot initiate to the router or the LAN"):
        # Positive control: the probe tool works against a port that is open.
        upstream.succeed("nc -z -w 2 203.0.113.1 80")
        upstream.fail("nc -z -w 2 203.0.113.100 22")
        upstream.fail("nc -z -w 2 203.0.113.100 53")
        upstream.fail("ping -c 1 -W 2 192.168.10.1")
        upstream.fail("ping -c 1 -W 2 203.0.113.100")
        client_ip = client.succeed("ip -4 -o addr show dev eth1 | awk '{print $4}' | cut -d/ -f1").strip()
        upstream.fail(f"ping -c 1 -W 2 {client_ip}")

    with subtest("LAN can reach SSH and ping on the router"):
        client.succeed("nc -z -w 2 192.168.10.1 22")
        client.succeed("ping -c 1 -W 2 192.168.10.1")
  '';
}