cnx-network-clan/modules/clan/router/tests/vm/default.nix

# End-to-end VM test of the router service: a PPPoE access concentrator plays
# the ISP on the WAN port, a trunk carries tagged lan/iot VLANs to `client`,
# and an untagged access port carries mgmt to `admin`.
#
#   isp ---(vlan 1: PPPoE)--- wan [gw] trunk ---(vlan 2: tagged 20/40)--- client
#                                    access --(vlan 3: untagged mgmt)--- admin
#
# What is proven: PPPoE dial-in with the vars-provided credentials, bridge
# VLAN tagging/untagging, Kea leases and reservations per VLAN, Blocky
# answering on the VLAN with the blocklist active, NAT to the WAN, and the
# firewall trust model (allowWan, mgmt-only SSH, no inter-VLAN forwarding).
{ pkgs, lib, ... }:
let
  # The vars mock answers every prompt with "mock-prompt-value-<name>"; the
  # ISP side must accept exactly those.
  chapSecrets = ''"mock-prompt-value-username" * "mock-prompt-value-password" *'';
  ispAddress = "192.0.2.1";

  clientMac = "02:00:00:00:00:20";
  clientAddress = "10.9.20.50";
  adminMac = "02:00:00:00:00:10";
  adminAddress = "10.9.10.50";
in
{
  name = "router";

  clan = {
    directory = ./.;
    # Bridges, VLAN netdevs, PPPoE and nftables need a real kernel.
    test.useContainers = false;
    inventory = {
      # Every node is a clan machine (the test framework's defaults require
      # it); only gw gets the router role.
      machines = {
        gw = { };
        isp = { };
        client = { };
        admin = { };
      };

      instances.router = {
        module.name = "router";
        module.input = "self";
        roles.default.machines.gw.settings = {
          site = "tst";
          siteId = 9;
          mesh.subnet = "fd00:7e57:c1a1:c0de::/64";
          wan.interface = "wan";
          trunkPorts = [ "trunk" ];
          accessPorts.access = "mgmt";
          vlans = {
            mgmt = {
              id = 10;
              dhcp.reservations.admin = {
                hwAddress = adminMac;
                ipAddress = adminAddress;
              };
            };
            lan = {
              id = 20;
              dhcp.reservations.client = {
                hwAddress = clientMac;
                ipAddress = clientAddress;
              };
            };
            iot = {
              id = 40;
              allowWan = false;
            };
          };
        };
      };
    };
  };

  nodes = {
    gw = {
      virtualisation.interfaces = {
        wan = {
          vlan = 1;
          assignIP = false;
        };
        trunk = {
          vlan = 2;
          assignIP = false;
        };
        access = {
          vlan = 3;
          assignIP = false;
        };
      };

      # Something must listen on 22 for the mgmt-only SSH rule to be observable
      # (a refused and a dropped connection look the same to the client).
      services.openssh.enable = true;

      # The sandbox has no internet: serve the blocklist from a local file
      # instead of GitHub, and skip CrowdSec, whose hub sync needs the network
      # (it is not what this test exercises).
      services.blocky.settings.blocking.denylists.ads = lib.mkForce [
        (toString (pkgs.writeText "ads.hosts" "0.0.0.0 ads.example.com\n"))
      ];
      services.crowdsec.enable = lib.mkForce false;
      services.crowdsec-firewall-bouncer.enable = lib.mkForce false;
    };

    isp = {
      virtualisation.interfaces.wan = {
        vlan = 1;
        assignIP = false;
      };
      # PPPoE access concentrator: one session, peer gets 192.0.2.10.
      systemd.services.pppoe-server = {
        wantedBy = [ "multi-user.target" ];
        after = [ "network.target" ];
        serviceConfig.ExecStart =
          "${pkgs.rp-pppoe}/sbin/pppoe-server -F -O /etc/ppp/pppoe-server-options"
          + " -q ${pkgs.ppp}/sbin/pppd -I wan -L ${ispAddress} -R 192.0.2.10";
      };
      environment.etc = {
        "ppp/pppoe-server-options".text = ''
          plugin pppoe.so
          require-chap
          lcp-echo-interval 10
          lcp-echo-failure 2
          nobsdcomp
          noccp
          novj
        '';
        "ppp/chap-secrets" = {
          text = chapSecrets;
          mode = "0640";
        };
      };
    };

    client = {
      virtualisation.interfaces.trunk = {
        vlan = 2;
        assignIP = false;
      };
      networking.useDHCP = false;
      networking.vlans = {
        lan0 = {
          id = 20;
          interface = "trunk";
        };
        iot0 = {
          id = 40;
          interface = "trunk";
        };
      };
      networking.interfaces.lan0 = {
        useDHCP = true;
        macAddress = clientMac;
      };
      networking.interfaces.iot0.useDHCP = true;
      environment.systemPackages = [
        pkgs.dnsutils
        pkgs.netcat
      ];
    };

    admin = {
      virtualisation.interfaces.access = {
        vlan = 3;
        assignIP = false;
      };
      networking.useDHCP = false;
      networking.interfaces.access = {
        useDHCP = true;
        macAddress = adminMac;
      };
      environment.systemPackages = [ pkgs.netcat ];
    };
  };

  testScript = ''
    start_all()

    with subtest("PPPoE session comes up with the vars credentials"):
        gw.wait_for_unit("pppd-wan.service")
        gw.wait_until_succeeds("ping -c1 -W1 ${ispAddress}")

    with subtest("DHCP hands out reserved leases per VLAN"):
        gw.wait_for_unit("kea-dhcp4-server.service")
        client.wait_until_succeeds("ip -4 addr show lan0 | grep -q 'inet ${clientAddress}/24'")
        client.wait_until_succeeds("ip -4 addr show iot0 | grep -q 'inet 10.9.40.1[0-9][0-9]/24'")
        admin.wait_until_succeeds("ip -4 addr show access | grep -q 'inet ${adminAddress}/24'")

    with subtest("Blocky serves the VLAN and blocks the denylist"):
        gw.wait_for_unit("blocky.service")
        answer = client.wait_until_succeeds("dig +short +time=2 @10.9.20.1 ads.example.com")
        assert answer.strip() == "0.0.0.0", f"expected blocked answer, got: {answer!r}"

    with subtest("NAT to the WAN only for VLANs with allowWan"):
        client.wait_until_succeeds("ping -c1 -W1 -I lan0 ${ispAddress}")
        client.fail("ping -c1 -W2 -I iot0 ${ispAddress}")

    with subtest("mgmt reaches other VLANs, other VLANs do not"):
        admin.succeed("ping -c1 -W2 ${clientAddress}")
        client.fail("ping -c1 -W2 -I lan0 ${adminAddress}")

    with subtest("SSH on the router only from mgmt"):
        gw.wait_for_open_port(22)
        admin.succeed("nc -z -w2 10.9.10.1 22")
        client.fail("nc -z -w2 10.9.20.1 22")
  '';
}