nixpkgs/nixos/tests/knot.nix

import ./make-test-python.nix (
  { pkgs, lib, ... }:
  let
    common = {
      networking.firewall.enable = false;
      networking.useDHCP = false;
    };
    exampleZone = pkgs.writeTextDir "example.com.zone" ''
      @ SOA ns.example.com. noc.example.com. 2019031301 86400 7200 3600000 172800
      @       NS      ns1
      @       NS      ns2
      ns1     A       192.168.0.1
      ns1     AAAA    fd00::1
      ns2     A       192.168.0.2
      ns2     AAAA    fd00::2
      www     A       192.0.2.1
      www     AAAA    2001:DB8::1
      sub     NS      ns.example.com.
    '';
    delegatedZone = pkgs.writeTextDir "sub.example.com.zone" ''
      @ SOA ns.example.com. noc.example.com. 2019031301 86400 7200 3600000 172800
      @       NS      ns1.example.com.
      @       NS      ns2.example.com.
      @       A       192.0.2.2
      @       AAAA    2001:DB8::2
    '';

    knotZonesEnv = pkgs.buildEnv {
      name = "knot-zones";
      paths = [
        exampleZone
        delegatedZone
      ];
    };
    # DO NOT USE pkgs.writeText IN PRODUCTION. This put secrets in the nix store!
    tsigFile = pkgs.writeText "tsig.conf" ''
      key:
        - id: xfr_key
          algorithm: hmac-sha256
          secret: zOYgOgnzx3TGe5J5I/0kxd7gTcxXhLYMEq3Ek3fY37s=
    '';
  in
  {
    name = "knot";
    meta = with pkgs.lib.maintainers; {
      maintainers = [ hexa ];
    };

    nodes = {
      primary =
        { lib, ... }:
        {
          imports = [ common ];

          # trigger sched_setaffinity syscall
          virtualisation.cores = 2;

          networking.interfaces.eth1 = {
            ipv4.addresses = lib.mkForce [
              {
                address = "192.168.0.1";
                prefixLength = 24;
              }
            ];
            ipv6.addresses = lib.mkForce [
              {
                address = "fd00::1";
                prefixLength = 64;
              }
            ];
          };
          services.knot.enable = true;
          services.knot.extraArgs = [ "-v" ];
          services.knot.keyFiles = [ tsigFile ];
          services.knot.settings = {
            server = {
              listen = [
                "0.0.0.0@53"
                "::@53"
              ];
              listen-quic = [
                "0.0.0.0@853"
                "::@853"
              ];
              automatic-acl = true;
            };

            acl.secondary_acl = {
              address = "192.168.0.2";
              key = "xfr_key";
              action = "transfer";
            };

            remote.secondary.address = "192.168.0.2@53";

            template.default = {
              storage = knotZonesEnv;
              notify = [ "secondary" ];
              acl = [ "secondary_acl" ];
              dnssec-signing = true;
              # Input-only zone files
              # https://www.knot-dns.cz/docs/2.8/html/operation.html#example-3
              # prevents modification of the zonefiles, since the zonefiles are immutable
              zonefile-sync = -1;
              zonefile-load = "difference";
              journal-content = "changes";
            };

            zone = {
              "example.com".file = "example.com.zone";
              "sub.example.com".file = "sub.example.com.zone";
            };

            log.syslog.any = "info";
          };
        };

      secondary =
        { lib, ... }:
        {
          imports = [ common ];
          networking.interfaces.eth1 = {
            ipv4.addresses = lib.mkForce [
              {
                address = "192.168.0.2";
                prefixLength = 24;
              }
            ];
            ipv6.addresses = lib.mkForce [
              {
                address = "fd00::2";
                prefixLength = 64;
              }
            ];
          };
          services.knot.enable = true;
          services.knot.keyFiles = [ tsigFile ];
          services.knot.extraArgs = [ "-v" ];
          services.knot.settings = {
            server = {
              automatic-acl = true;
            };

            xdp = {
              listen = [
                "eth1"
              ];
              tcp = true;
            };

            remote.primary = {
              address = "192.168.0.1@53";
              key = "xfr_key";
            };

            remote.primary-quic = {
              address = "192.168.0.1@853";
              key = "xfr_key";
              quic = true;
            };

            template.default = {
              # zonefileless setup
              # https://www.knot-dns.cz/docs/2.8/html/operation.html#example-2
              zonefile-sync = "-1";
              zonefile-load = "none";
              journal-content = "all";
            };

            zone = {
              "example.com" = {
                master = "primary";
                file = "example.com.zone";
              };
              "sub.example.com" = {
                master = "primary-quic";
                file = "sub.example.com.zone";
              };
            };

            log.syslog.any = "debug";
          };
        };
      client =
        { lib, nodes, ... }:
        {
          imports = [ common ];
          networking.interfaces.eth1 = {
            ipv4.addresses = [
              {
                address = "192.168.0.3";
                prefixLength = 24;
              }
            ];
            ipv6.addresses = [
              {
                address = "fd00::3";
                prefixLength = 64;
              }
            ];
          };
          environment.systemPackages = [ pkgs.knot-dns ];
        };
    };

    testScript =
      { nodes, ... }:
      let
        primary4 = (lib.head nodes.primary.config.networking.interfaces.eth1.ipv4.addresses).address;
        primary6 = (lib.head nodes.primary.config.networking.interfaces.eth1.ipv6.addresses).address;

        secondary4 = (lib.head nodes.secondary.config.networking.interfaces.eth1.ipv4.addresses).address;
        secondary6 = (lib.head nodes.secondary.config.networking.interfaces.eth1.ipv6.addresses).address;
      in
      ''
        import re

        start_all()

        client.wait_for_unit("network.target")
        primary.wait_for_unit("knot.service")
        secondary.wait_for_unit("knot.service")

        for zone in ("example.com.", "sub.example.com."):
            secondary.wait_until_succeeds(
              f"knotc zone-status {zone} | grep -q 'serial: 2019031302'"
            )

        def test(host, query_type, query, pattern):
            out = client.succeed(f"khost -t {query_type} {query} {host}").strip()
            client.log(f"{host} replied with: {out}")
            assert re.search(pattern, out), f'Did not match "{pattern}"'


        for host in ("${primary4}", "${primary6}", "${secondary4}", "${secondary6}"):
            with subtest(f"Interrogate {host}"):
                test(host, "SOA", "example.com", r"start of authority.*noc\.example\.com\.")
                test(host, "A", "example.com", r"has no [^ ]+ record")
                test(host, "AAAA", "example.com", r"has no [^ ]+ record")

                test(host, "A", "www.example.com", r"address 192.0.2.1$")
                test(host, "AAAA", "www.example.com", r"address 2001:db8::1$")

                test(host, "NS", "sub.example.com", r"nameserver is ns\d\.example\.com.$")
                test(host, "A", "sub.example.com", r"address 192.0.2.2$")
                test(host, "AAAA", "sub.example.com", r"address 2001:db8::2$")

                test(host, "RRSIG", "www.example.com", r"RR set signature is")
                test(host, "DNSKEY", "example.com", r"DNSSEC key is")

        primary.log(primary.succeed("systemd-analyze security knot.service | grep -v '✓'"))
      '';
  }
)