webnsupdate/flake-modules/tests.nix

{ self, ... }:
{
  perSystem =
    { pkgs, self', ... }:
    {
      checks =
        let
          testDomain = "webnstest.example";
          lastIPPath = "/var/lib/webnsupdate/last-ip.json";

          zoneFile = pkgs.writeText "${testDomain}.zoneinfo" ''
            $TTL 60 ; 1 minute
            $ORIGIN ${testDomain}.
            @         IN SOA    ns1.${testDomain}. admin.${testDomain}. (
                          1            ; serial
                          6h           ; refresh
                          1h           ; retry
                          1w           ; expire
                          1d)          ; negative caching TTL

                      IN  NS    ns1.${testDomain}.
            @         IN  A     127.0.0.1
            ns1       IN  A     127.0.0.1
            nsupdate  IN  A     127.0.0.1
            @         IN  AAAA  ::1
            ns1       IN  AAAA  ::1
            nsupdate  IN  AAAA  ::1
          '';

          bindDynamicZone =
            { config, ... }:
            let
              bindCfg = config.services.bind;
              bindData = bindCfg.directory;
              dynamicZonesDir = "${bindData}/zones";
            in
            {
              services.bind.zones.${testDomain} = {
                master = true;
                file = "${dynamicZonesDir}/${testDomain}";
                extraConfig = ''
                  allow-update { key rndc-key; };
                '';
              };

              systemd.services.bind.preStart = ''
                # shellcheck disable=SC2211,SC1127
                rm -f ${dynamicZonesDir}/* # reset dynamic zones

                # create a dynamic zones dir
                mkdir -m 0755 -p ${dynamicZonesDir}
                # copy dynamic zone's file to the dynamic zones dir
                cp ${zoneFile} ${dynamicZonesDir}/${testDomain}
              '';
            };

          webnsupdate-ipv4-machine =
            { lib, ... }:
            {
              imports = [
                bindDynamicZone
                self.nixosModules.webnsupdate
              ];

              config = {
                environment.systemPackages = [
                  pkgs.dig
                  pkgs.curl
                ];

                services = {
                  bind.enable = true;

                  webnsupdate = {
                    enable = true;
                    bindIp = lib.mkDefault "127.0.0.1";
                    keyFile = "/etc/bind/rndc.key";
                    # test:test (user:password)
                    passwordFile = pkgs.writeText "webnsupdate.pass" "FQoNmuU1BKfg8qsU96F6bK5ykp2b0SLe3ZpB3nbtfZA";
                    package = self'.packages.webnsupdate;
                    extraArgs = [
                      "-vvv" # debug messages
                      "--ip-source=ConnectInfo"
                    ];
                    records = ''
                      test1.${testDomain}.
                      test2.${testDomain}.
                      test3.${testDomain}.
                    '';
                  };
                };
              };
            };

          webnsupdate-ipv6-machine = {
            imports = [
              webnsupdate-ipv4-machine
            ];

            config.services.webnsupdate.bindIp = "::1";
          };

          webnsupdate-nginx-machine =
            { lib, config, ... }:
            {
              imports = [
                webnsupdate-ipv4-machine
              ];

              config.services = {
                # Use default IP Source
                webnsupdate.extraArgs = lib.mkForce [ "-vvv" ]; # debug messages

                nginx = {
                  enable = true;
                  recommendedProxySettings = true;

                  virtualHosts.webnsupdate.locations."/".proxyPass =
                    "http://${config.services.webnsupdate.bindIp}:${builtins.toString config.services.webnsupdate.bindPort}";
                };
              };
            };

          webnsupdate-ipv4-only-machine = {
            imports = [ webnsupdate-nginx-machine ];
            config.services.webnsupdate.allowedIPVersion = "ipv4-only";
          };

          webnsupdate-ipv6-only-machine = {
            imports = [ webnsupdate-nginx-machine ];
            config.services.webnsupdate.allowedIPVersion = "ipv6-only";
          };

          # "A" for IPv4, "AAAA" for IPv6, "ANY" for any
          testTemplate =
            {
              ipv4 ? false,
              ipv6 ? false,
              nginx ? false,
              exclusive ? false,
            }:
            if exclusive && (ipv4 == ipv6) then
              builtins.throw "exclusive means one of ipv4 or ipv6 must be set, but not both"
            else
              ''
                IPV4: bool = ${if ipv4 then "True" else "False"}
                IPV6: bool = ${if ipv6 then "True" else "False"}
                NGINX: bool = ${if nginx then "True" else "False"}
                EXCLUSIVE: bool = ${if exclusive then "True" else "False"}
                print(f"{IPV4=} {IPV6=} {EXCLUSIVE=}")

                CURL: str = "curl --fail --no-progress-meter --show-error"

                machine.start(allow_reboot=True)
                machine.wait_for_unit("bind.service")
                machine.wait_for_unit("webnsupdate.service")

                STATIC_DOMAINS: list[str] = ["${testDomain}", "ns1.${testDomain}", "nsupdate.${testDomain}"]
                DYNAMIC_DOMAINS: list[str] = ["test1.${testDomain}", "test2.${testDomain}", "test3.${testDomain}"]

                def dig_cmd(domain: str, record: str, ip: str | None) -> str:
                    match_ip = "" if ip is None else f"\\s\\+60\\s\\+IN\\s\\+{record}\\s\\+{ip}$"
                    return f"dig @localhost {record} {domain} +noall +answer | grep '^{domain}.{match_ip}'"

                def curl_cmd(domain: str, identity: str, path: str, query: dict[str, str]) -> str:
                    from urllib.parse import urlencode
                    q= f"?{urlencode(query)}" if query else ""
                    return f"{CURL} -u {identity} -X GET 'http://{domain}{"" if NGINX else ":5353"}/{path}{q}'"

                def domain_available(domain: str, record: str, ip: str | None=None):
                    machine.succeed(dig_cmd(domain, record, ip))

                def domain_missing(domain: str, record: str, ip: str | None=None):
                    machine.fail(dig_cmd(domain, record, ip))

                def update_records(domain: str="localhost", /, *, path: str="update", **kwargs):
                    machine.succeed(curl_cmd(domain, "test:test", path, kwargs))
                    machine.succeed("cat ${lastIPPath}")

                def update_records_fail(domain: str="localhost", /, *, identity: str="test:test", path: str="update", **kwargs):
                    machine.fail(curl_cmd(domain, identity, path, kwargs))
                    machine.fail("cat ${lastIPPath}")

                def invalid_update(domain: str="localhost"):
                    update_records_fail(domain, identity="bad_user:test")
                    update_records_fail(domain, identity="test:bad_pass")

                # Tests

                with subtest("static DNS records are available"):
                    print(f"{IPV4=} {IPV6=} {EXCLUSIVE=}")
                    for domain in STATIC_DOMAINS:
                        domain_available(domain, "A", "127.0.0.1") # IPv4
                        domain_available(domain, "AAAA", "::1")    # IPv6

                with subtest("dynamic DNS records are missing"):
                    print(f"{IPV4=} {IPV6=} {EXCLUSIVE=}")
                    for domain in DYNAMIC_DOMAINS:
                        domain_missing(domain, "A")    # IPv4
                        domain_missing(domain, "AAAA") # IPv6

                with subtest("invalid auth fails to update records"):
                    print(f"{IPV4=} {IPV6=} {EXCLUSIVE=}")
                    invalid_update()
                    for domain in DYNAMIC_DOMAINS:
                        domain_missing(domain, "A")    # IPv4
                        domain_missing(domain, "AAAA") # IPv6

                if EXCLUSIVE:
                    with subtest("exclusive IP version fails to update with invalid version"):
                        print(f"{IPV4=} {IPV6=} {EXCLUSIVE=}")
                        if IPV6:
                            update_records_fail("127.0.0.1")
                        if IPV4:
                            update_records_fail("[::1]")

                with subtest("valid auth updates records"):
                    print(f"{IPV4=} {IPV6=} {EXCLUSIVE=}")
                    if IPV4:
                        update_records("127.0.0.1")
                    if IPV6:
                        update_records("[::1]")

                    for domain in DYNAMIC_DOMAINS:
                        if IPV4:
                            domain_available(domain, "A", "127.0.0.1")
                        elif IPV6 and EXCLUSIVE:
                            domain_missing(domain, "A")

                        if IPV6:
                            domain_available(domain, "AAAA", "::1")
                        elif IPV4 and EXCLUSIVE:
                            domain_missing(domain, "AAAA")

                with subtest("valid auth fritzbox compatible updates records"):
                    print(f"{IPV4=} {IPV6=} {EXCLUSIVE=}")
                    if IPV4 and IPV6:
                        update_records("127.0.0.1", domain="test", ipv4="1.2.3.4", ipv6="::1234")
                    elif IPV4:
                        update_records("127.0.0.1", ipv4="1.2.3.4")
                    elif IPV6:
                        update_records("[::1]", ipv6="::1234")

                    for domain in DYNAMIC_DOMAINS:
                        if IPV4:
                            domain_available(domain, "A", "1.2.3.4")
                        elif IPV6 and EXCLUSIVE:
                            domain_missing(domain, "A")

                        if IPV6:
                            domain_available(domain, "AAAA", "::1234")
                        elif IPV4 and EXCLUSIVE:
                            domain_missing(domain, "AAAA")

                with subtest("valid auth replaces records"):
                    print(f"{IPV4=} {IPV6=} {EXCLUSIVE=}")
                    if IPV4:
                        update_records("127.0.0.1")
                    if IPV6:
                        update_records("[::1]")

                    for domain in DYNAMIC_DOMAINS:
                        if IPV4:
                            domain_available(domain, "A", "127.0.0.1")
                        elif IPV6 and EXCLUSIVE:
                            domain_missing(domain, "A")

                        if IPV6:
                            domain_available(domain, "AAAA", "::1")
                        elif IPV4 and EXCLUSIVE:
                            domain_missing(domain, "AAAA")

                machine.reboot()
                machine.succeed("cat ${lastIPPath}")
                machine.wait_for_unit("webnsupdate.service")
                machine.succeed("cat ${lastIPPath}")

                with subtest("static DNS records are available after reboot"):
                    print(f"{IPV4=} {IPV6=} {EXCLUSIVE=}")
                    for domain in STATIC_DOMAINS:
                        domain_available(domain, "A", "127.0.0.1") # IPv4
                        domain_available(domain, "AAAA", "::1")    # IPv6

                with subtest("dynamic DNS records are available after reboot"):
                    print(f"{IPV4=} {IPV6=} {EXCLUSIVE=}")
                    for domain in DYNAMIC_DOMAINS:
                        if IPV4:
                            domain_available(domain, "A", "127.0.0.1")
                        elif IPV6 and EXCLUSIVE:
                            domain_missing(domain, "A")

                        if IPV6:
                            domain_available(domain, "AAAA", "::1")
                        elif IPV4 and EXCLUSIVE:
                            domain_missing(domain, "AAAA")
              '';
        in
        {
          module-ipv4-test = pkgs.testers.nixosTest {
            name = "webnsupdate-ipv4-module";
            nodes.machine = webnsupdate-ipv4-machine;
            testScript = testTemplate { ipv4 = true; };
          };
          module-ipv6-test = pkgs.testers.nixosTest {
            name = "webnsupdate-ipv6-module";
            nodes.machine = webnsupdate-ipv6-machine;
            testScript = testTemplate { ipv6 = true; };
          };
          module-nginx-test = pkgs.testers.nixosTest {
            name = "webnsupdate-nginx-module";
            nodes.machine = webnsupdate-nginx-machine;
            testScript = testTemplate {
              ipv4 = true;
              ipv6 = true;
              nginx = true;
            };
          };
          module-ipv4-only-test = pkgs.testers.nixosTest {
            name = "webnsupdate-ipv4-only-module";
            nodes.machine = webnsupdate-ipv4-only-machine;
            testScript = testTemplate {
              ipv4 = true;
              nginx = true;
              exclusive = true;
            };
          };
          module-ipv6-only-test = pkgs.testers.nixosTest {
            name = "webnsupdate-ipv6-only-module";
            nodes.machine = webnsupdate-ipv6-only-machine;
            testScript = testTemplate {
              ipv6 = true;
              nginx = true;
              exclusive = true;
            };
          };
        };
    };
}