# `checks.vm-hive-squid`: runs ../host-modules/hive-squid.nix in a VM, which
# the module-eval suite can only read, and checks that an agent with the proxy
# variables of nix/agent-modules/egress-proxy.nix reaches squid and skips it
# where it should. Three nodes on two networks:
#
#   agent (vlan 1) --> proxy (vlan 1 = the stand-in bridge, vlan 2 = uplink) --> web (vlan 2)
#
# `proxy` imports the squid module against stub options for the few hive
# options it reads, so no hive-c0re build is pulled in; the module-eval suite
# covers the real option wiring.
{ pkgs }:
let
  bridgeIp = "192.168.1.1";

  hiveStubs =
    { lib, config, ... }:
    {
      options.services.hyperhive = {
        deploy.hive-controller.enable = lib.mkOption {
          type = lib.types.bool;
          default = true;
        };
        swarm.serviceDomains = lib.mkOption {
          type = lib.types.listOf lib.types.str;
          default = [ "forge.t.local" ];
        };
        swarm.forge.domain = lib.mkOption {
          type = lib.types.str;
          default = "forge.t.local";
        };
        network = {
          bridgeIp = lib.mkOption {
            type = lib.types.str;
            default = bridgeIp;
          };
          networkCidr = lib.mkOption {
            type = lib.types.str;
            default = "192.168.1.0/24";
          };
          exposeHostPorts = lib.mkOption {
            type = lib.types.listOf lib.types.port;
            default = [ ];
          };
          enable = lib.mkOption {
            type = lib.types.bool;
            default = true;
          };
        };
        otel = {
          scrapeTargets = lib.mkOption {
            type = lib.types.attrsOf lib.types.str;
            default = { };
          };
          journalUnits = lib.mkOption {
            type = lib.types.listOf lib.types.str;
            default = [ ];
          };
        };
      };
      # The user hive-c0re runs as, which owns the users file squid reads.
      config.users.users.hive-core = {
        isSystemUser = true;
        group = "hive-core";
      };
      config.users.groups.hive-core = { };
      # The firewall hole hive-network.nix opens for `exposeHostPorts`.
      config.networking.firewall.interfaces.eth1.allowedTCPPorts =
        config.services.hyperhive.network.exposeHostPorts;
    };

  # The uplink is a public range. The private ranges squid must refuse are
  # extra subnets on the same link, so the proxy can reach a listener in each:
  # a name that resolves there is refused only by squid's own rules.
  privateNets = [
    {
      name = "priv10.t.local";
      net = "10.99.0";
    }
    {
      name = "priv172.t.local";
      net = "172.16.99";
    }
    {
      name = "priv192.t.local";
      net = "192.168.99";
    }
    {
      name = "cgnat.t.local";
      net = "100.64.99";
    }
  ];
  privateAddrs =
    host:
    map (p: {
      address = "${p.net}.${toString host}";
      prefixLength = 24;
    }) privateNets;
  ulaAddr = host: {
    address = "fd00:99::${toString host}";
    prefixLength = 64;
  };

  # Plain TCP on 443: a CONNECT is answered before any TLS, which is all the
  # test reads.
  listen443 = addrs: {
    services.nginx = {
      enable = true;
      virtualHosts.default.listen = map (addr: {
        inherit addr;
        port = 443;
      }) addrs;
    };
  };
in
pkgs.testers.runNixOSTest {
  name = "hive-squid";

  nodes = {
    proxy =
      { lib, pkgs, ... }:
      {
        imports = [
          hiveStubs
          ../host-modules/hive-squid.nix
          # Listeners a leaked destination would reach: loopback (`::` lands on `::1`) and the bridge.
          (listen443 [
            "127.0.0.1"
            "[::1]"
            bridgeIp
          ])
        ];
        virtualisation.vlans = [
          1
          2
        ];
        networking.interfaces.eth1.ipv4.addresses = lib.mkForce [
          {
            address = bridgeIp;
            prefixLength = 24;
          }
        ];
        networking.interfaces.eth2.ipv4.addresses = lib.mkForce (
          [
            {
              address = "203.0.113.1";
              prefixLength = 24;
            }
          ]
          ++ privateAddrs 1
        );
        networking.interfaces.eth2.ipv6.addresses = [ (ulaAddr 1) ];
        # Names the proxy resolves, each a different destination class.
        networking.hosts = {
          # `other.t.local` and the self-check's deny canary resolve, are
          # reachable and are on no allow-list. The canary is in the hosts file
          # because the destination denies look every name up first, and the VM
          # has no resolver to answer for an unknown one before the client gives up.
          "203.0.113.2" = [
            "forge.t.local"
            "other.t.local"
            "deny-canary.invalid"
          ];
          "127.0.0.1" = [ "loop.t.local" ];
          ${bridgeIp} = [ "bridge.t.local" ];
          "169.254.169.254" = [ "meta.t.local" ];
          "fd00:99::2" = [ "ula.t.local" ];
          "::" = [ "any6.t.local" ];
        }
        // lib.listToAttrs (
          map (p: {
            name = "${p.net}.2";
            value = [ p.name ];
          }) privateNets
        );
        # The gateway: nginx answers on the bridge, which agents reach directly.
        networking.firewall.interfaces.eth1.allowedTCPPorts = [ 443 ];
        # Renders the users file the way hive-c0re does.
        environment.systemPackages = [ pkgs.mkpasswd ];
        services.hyperhive.squid.selfcheck.unrestrictedCanary = "other.t.local";
        services.hyperhive.squid.allowedDomains = [
          "forge.t.local"
          "loop.t.local"
          "bridge.t.local"
          "meta.t.local"
          "ula.t.local"
          "any6.t.local"
        ]
        ++ map (p: p.name) privateNets;
      };

    web =
      { lib, ... }:
      {
        imports = [
          (listen443 [
            "0.0.0.0"
            "[::]"
          ])
        ];
        virtualisation.vlans = [ 2 ];
        networking.interfaces.eth1.ipv4.addresses = lib.mkForce (
          [
            {
              address = "203.0.113.2";
              prefixLength = 24;
            }
          ]
          ++ privateAddrs 2
        );
        networking.interfaces.eth1.ipv6.addresses = [ (ulaAddr 2) ];
        networking.firewall.allowedTCPPorts = [ 443 ];
      };

    agent =
      { lib, ... }:
      {
        virtualisation.vlans = [ 1 ];
        # The bridge listener, by the name the proxy resolves to the same address.
        networking.hosts.${bridgeIp} = [ "bridge.t.local" ];
        networking.interfaces.eth1.ipv4.addresses = lib.mkForce [
          {
            address = "192.168.1.3";
            prefixLength = 24;
          }
        ];
      };
  };

  testScript = ''
    import re
    import secrets
    import time

    # Drawn fresh each run. The users file holds only their hashes, as
    # hive-c0re renders it. `atlas` is a restricted agent and `uma` an
    # unrestricted one; the two self-check users share one secret.
    PW = {
        u: secrets.token_urlsafe(24)
        for u in ("atlas", "uma", "selfcheck.squid", "ruth")
    }
    PW["selfcheck-unrestricted.squid"] = PW["selfcheck.squid"]
    ATLAS = "atlas:" + PW["atlas"]
    UMA = "uma:" + PW["uma"]
    SELFCHECK = "selfcheck.squid:" + PW["selfcheck.squid"]
    SELFCHECK_U = "selfcheck-unrestricted.squid:" + PW["selfcheck.squid"]
    RUTH = "ruth:" + PW["ruth"]
    USERS = "/var/lib/hive-squid/users"
    LIST = "/var/lib/hive-squid/unrestricted"
    UNRESTRICTED = ["uma", "selfcheck-unrestricted.squid"]

    start_all()
    proxy.wait_for_unit("squid.service")
    proxy.wait_for_unit("squid-exporter.service")
    proxy.wait_for_unit("nginx.service")
    web.wait_for_unit("nginx.service")
    proxy.wait_for_open_port(3128, "${bridgeIp}")

    PRIVATE = ["priv10.t.local", "priv172.t.local", "priv192.t.local", "cgnat.t.local", "ula.t.local", "any6.t.local"]

    def connect(host, cred=None):
        # The status squid answered the CONNECT with; 000 when none came back.
        auth = f"-U '{cred}' " if cred else ""
        return agent.succeed(
            "curl -sS -m 15 -p -x http://${bridgeIp}:3128 " + auth + "-o /dev/null "
            f"-w '%{{http_connect}}' https://{host}/ 2>/dev/null || true"
        ).strip()

    def write_users(*creds):
        # What hive-c0re does: hashes only, written aside and renamed into
        # place, in squid's group. The sleep keeps two writes in different
        # seconds, which is how the helper tells the file changed.
        lines = " ".join(
            f"printf '%s:%s\\n' {u} \"$(mkpasswd -m sha-512 -s <<< {p})\";"
            for u, p in (c.split(":", 1) for c in creds)
        )
        proxy.succeed(
            f"({lines}) > {USERS}.new && chown hive-core:squid {USERS}.new "
            f"&& chmod 640 {USERS}.new && mv {USERS}.new {USERS} && sleep 1"
        )

    def write_list(*names):
        # What hive-c0re does for the list. Squid reads it on a reload, which
        # squid-unrestricted-reload does when the file changes.
        body = "printf '%s\\n' " + " ".join(names) if names else ":"
        proxy.succeed(
            f"({body}) > {LIST}.new && chown hive-core:squid {LIST}.new "
            f"&& chmod 640 {LIST}.new && mv {LIST}.new {LIST}"
        )

    def wait_status(host, cred, want):
        # The reload that follows a list write is asynchronous.
        got = None
        for _ in range(30):
            got = connect(host, cred)
            if got == want:
                return
            time.sleep(1)
        raise AssertionError(f"{host} as {cred.split(':')[0]}: wanted {want}, last got {got}")

    with subtest("the private listeners answer the proxy itself, so a refusal below is squid's"):
        for addr in ["10.99.0.2", "172.16.99.2", "192.168.99.2", "100.64.99.2", "fd00:99::2"]:
            proxy.wait_until_succeeds(f"timeout 3 bash -c 'echo > /dev/tcp/{addr}/443'")
        assert connect("forge.t.local", None) == "407", "control: the public name is challenged, not refused"

    with subtest("squid starts on the empty users file and list and refuses every login"):
        assert proxy.succeed(f"stat -c '%U:%G %a' {USERS}").strip() == "hive-core:squid 640"
        assert proxy.succeed(f"stat -c '%U:%G %a' {LIST}").strip() == "hive-core:squid 640"
        assert proxy.succeed("stat -c %a /var/lib/hive-squid").strip() == "2750"
        assert proxy.succeed(f"stat -c %s {USERS}").strip() == "0"
        assert proxy.succeed(f"stat -c %s {LIST}").strip() == "0"
        assert connect("forge.t.local", ATLAS) == "407", "no such user"
        assert connect("other.t.local", UMA) == "407", "no such user, unlisted name"

    with subtest("the users file holds hashes, never a password"):
        write_users(ATLAS, UMA, SELFCHECK, SELFCHECK_U)
        users = proxy.succeed(f"cat {USERS}")
        rows = [l.split(":", 1) for l in users.splitlines()]
        assert sorted(r[0] for r in rows) == sorted(
            ["atlas", "uma", "selfcheck.squid", "selfcheck-unrestricted.squid"]
        ), users
        assert all(re.fullmatch(r"\$6\$\S+", r[1]) for r in rows), users
        assert not any(p in users for p in PW.values()), users

    with subtest("an allowed name is tunnelled with a credential and challenged without one"):
        assert connect("forge.t.local", ATLAS) == "200", "allow canary"
        assert connect("forge.t.local") == "407", "no credential"
        assert connect("forge.t.local", "atlas:" + secrets.token_urlsafe(24)) == "407", "wrong password"
        assert connect("forge.t.local", "nobody:" + PW["atlas"]) == "407", "unknown user"
        # DES would accept the first eight characters; the hash must not.
        assert connect("forge.t.local", "atlas:" + PW["atlas"][:8]) == "407", "a prefix of the password"

    with subtest("with an empty list every user is restricted"):
        assert connect("other.t.local", UMA) == "403", "uma, not listed yet"
        assert connect("other.t.local", ATLAS) == "403"
        assert connect("forge.t.local", UMA) == "200", "control: uma still reaches a listed name"

    with subtest("a listed user reaches an unlisted name and the others are refused"):
        write_list(*UNRESTRICTED)
        wait_status("other.t.local", UMA, "200")
        assert connect("other.t.local", ATLAS) == "403", "restricted"
        assert connect("forge.t.local", ATLAS) == "200", "restricted, listed name"
        assert connect("forge.t.local", UMA) == "200", "unrestricted, listed name"
        assert connect("other.t.local") == "407", "no credential"
        assert connect("other.t.local", "uma:" + secrets.token_urlsafe(24)) == "407", "wrong password"
        assert connect("deny-canary.invalid", ATLAS) == "403", "restricted, an unlisted name"

    with subtest("neither kind of user reaches a host-local or private destination, or another port"):
        for cred in (ATLAS, UMA):
            who = cred.split(":")[0]
            assert connect("loop.t.local", cred) == "403", f"{who}: loopback"
            assert connect("bridge.t.local", cred) == "403", f"{who}: bridge address"
            assert connect("meta.t.local", cred) == "403", f"{who}: link-local"
            assert connect("127.0.0.1", cred) == "403", f"{who}: loopback literal"
            assert connect("203.0.113.2", cred) == "403", f"{who}: IP literal"
            for name in PRIVATE:
                assert connect(name, cred) == "403", f"{who}: {name}, a listed name that resolves to a private address"
            assert connect("other.t.local:22", cred) == "403", f"{who}: port 22"
            assert connect("other.t.local:80", cred) == "403", f"{who}: CONNECT to port 80"

    with subtest("an agent cannot read cachemgr"):
        agent.succeed(
            "test $(curl -s -m 15 -o /dev/null -w '%{http_code}' "
            "http://${bridgeIp}:3128/squid-internal-mgr/counters) = 403"
        )

    # The variables an agent's `proxy.env` carries (nix/agent-modules/egress-proxy.nix),
    # as a curl on the agent reads them.
    def agent_env(cred, no_proxy):
        url = f"http://{cred}@${bridgeIp}:3128"
        pairs = {
            "HTTPS_PROXY": url, "HTTP_PROXY": url, "https_proxy": url, "http_proxy": url,
            "NO_PROXY": no_proxy, "no_proxy": no_proxy,
        }
        return " ".join(f"{k}='{v}'" for k, v in pairs.items())

    def agent_get(cred, no_proxy, url):
        # The status of the answer, whoever gave it; 000 when none came.
        return agent.succeed(
            f"env {agent_env(cred, no_proxy)} curl -sS -m 15 -o /dev/null -w '%{{http_code}}' {url} 2>/dev/null || true"
        ).strip()

    def squid_saw(pattern):
        return re.search(pattern, proxy.succeed("journalctl -u squid.service -o cat"))

    with subtest("an agent with the proxy variables reaches an allowed name through squid and a swarm-internal name directly"):
        NO_PROXY = "localhost,127.0.0.1,::1,bridge.t.local,${bridgeIp}"
        # Plain HTTP to the listeners' port 443, so the answer carries a status.
        assert agent_get(ATLAS, NO_PROXY, "http://forge.t.local:443/") not in ("000", "403", "407")
        assert squid_saw(r"192\.168\.1\.3 TCP_\w+/\d+ .* GET http://forge\.t\.local:443/ atlas "), "through squid, as atlas"
        assert agent_get(ATLAS, NO_PROXY, "http://bridge.t.local:443/") not in ("000", "403", "407")
        assert not squid_saw(r"GET http://bridge\.t\.local:443/ "), "a swarm-internal name never reached squid"
        # Control: without the exemption the same request goes to squid, which refuses the bridge.
        assert agent_get(ATLAS, "localhost", "http://bridge.t.local:443/") == "403"
        assert squid_saw(r"192\.168\.1\.3 TCP_DENIED/403 .* GET http://bridge\.t\.local:443/ "), "control: squid refuses it"
        # CONNECT takes the same path: HTTPS_PROXY names the proxy for https URLs.
        assert agent.succeed(
            f"env {agent_env(ATLAS, NO_PROXY)} curl -sS -m 15 -o /dev/null -w '%{{http_connect}}' https://forge.t.local/ 2>/dev/null || true"
        ).strip() == "200"

    with subtest("an agent without the proxy variables goes direct"):
        assert agent.succeed(
            "curl -sS -m 15 -o /dev/null -w '%{http_code}' http://bridge.t.local:443/ 2>/dev/null || true"
        ).strip() not in ("000", "403", "407")

    with subtest("the self-check passes with its credential"):
        proxy.succeed("install -d -m 0700 /run/credstore")
        proxy.succeed(f"printf '%s\\n' {PW['selfcheck.squid']} > /run/credstore/hive-squid.selfcheck")
        proxy.succeed("chmod 400 /run/credstore/hive-squid.selfcheck")
        proxy.succeed("systemctl start squid-selfcheck.service")
        proxy.succeed(
            "journalctl -u squid-selfcheck.service -o cat | "
            "grep -F 'squid-selfcheck ok=true allow_canary=forge.t.local "
            "allow_status=200 deny_canary=deny-canary.invalid deny_status=403 "
            "anon_status=407 credential=present "
            "unrestricted_canary=other.t.local unrestricted_status=200'"
        )

    with subtest("the self-check fails, and says why, when the unrestricted user is not listed"):
        write_list()
        wait_status("other.t.local", UMA, "403")
        proxy.fail("systemctl start squid-selfcheck.service")
        proxy.succeed(
            "journalctl -u squid-selfcheck.service -o cat | "
            "grep -F 'squid-selfcheck ok=false allow_canary=forge.t.local "
            "allow_status=200 deny_canary=deny-canary.invalid deny_status=403 "
            "anon_status=407 credential=present "
            "unrestricted_canary=other.t.local unrestricted_status=403'"
        )
        write_list(*UNRESTRICTED)
        wait_status("other.t.local", UMA, "200")

    with subtest("the self-check fails, and says why, without a credential or with the wrong one"):
        proxy.succeed("mv /run/credstore/hive-squid.selfcheck /run/credstore/kept")
        proxy.fail("systemctl start squid-selfcheck.service")
        proxy.succeed(
            "journalctl -u squid-selfcheck.service -o cat | "
            "grep -F 'squid-selfcheck ok=false allow_canary=forge.t.local "
            "allow_status=407 deny_canary=deny-canary.invalid deny_status=407 "
            "anon_status=407 credential=absent "
            "unrestricted_canary=other.t.local unrestricted_status=407'"
        )
        proxy.succeed(f"printf '%s\\n' {secrets.token_urlsafe(24)} > /run/credstore/hive-squid.selfcheck")
        proxy.fail("systemctl start squid-selfcheck.service")
        proxy.succeed(
            "journalctl -u squid-selfcheck.service -o cat | "
            "grep -F 'allow_status=407 deny_canary=deny-canary.invalid deny_status=407 "
            "anon_status=407 credential=present "
            "unrestricted_canary=other.t.local unrestricted_status=407'"
        )
        proxy.succeed("mv -f /run/credstore/kept /run/credstore/hive-squid.selfcheck")

    with subtest("squid's access log reaches the journal under squid.service, with the user"):
        journal = proxy.succeed("journalctl -u squid.service -o cat")
        assert re.search(r"192\.168\.1\.3 TCP_TUNNEL/200 .* CONNECT forge\.t\.local:443 atlas ", journal), journal
        assert re.search(r"192\.168\.1\.3 TCP_TUNNEL/200 .* CONNECT other\.t\.local:443 uma ", journal), journal
        assert re.search(r"192\.168\.1\.3 TCP_DENIED/407 .* CONNECT forge\.t\.local:443 - ", journal), journal
        assert re.search(r"192\.168\.1\.3 TCP_DENIED/403 .* CONNECT other\.t\.local:443 atlas ", journal), journal
        assert re.search(r"192\.168\.1\.3 TCP_DENIED/403 .* CONNECT loop\.t\.local:443 ", journal), journal
        # Exporter scrapes are cachemgr requests and stay out of the log.
        assert "squid-internal-mgr" not in journal, journal

    with subtest("the exporter serves squid's counters on loopback"):
        metrics = proxy.succeed("curl -sf http://127.0.0.1:9301/metrics")
        print("\n".join(l for l in metrics.splitlines() if l.startswith(("squid_up", "squid_client_http"))))
        assert 'squid_up{host="${bridgeIp}"} 1' in metrics
        m = re.search(r"^squid_client_http_requests_total (\S+)$", metrics, re.M)
        assert m and float(m.group(1)) >= 9, metrics
        m = re.search(r"^squid_client_http_errors_total (\S+)$", metrics, re.M)
        assert m and float(m.group(1)) >= 1, metrics
        proxy.fail("curl -sf -m 5 http://${bridgeIp}:9301/metrics")

    with subtest("squid survives a reload and keeps the list"):
        proxy.succeed("systemctl reload squid.service")
        proxy.wait_for_unit("squid.service")
        assert connect("forge.t.local", ATLAS) == "200"
        assert connect("other.t.local", UMA) == "200"
        assert connect("other.t.local", ATLAS) == "403"

    with subtest("a user added while squid runs can log in, with no reload, and is restricted"):
        write_users(ATLAS, UMA, SELFCHECK, SELFCHECK_U, RUTH)
        assert connect("forge.t.local", RUTH) == "200"
        assert connect("forge.t.local", ATLAS) == "200"
        assert connect("other.t.local", RUTH) == "403"

    with subtest("a user the list gains is unrestricted after the reload and loses it again"):
        write_list(*UNRESTRICTED, "ruth")
        wait_status("other.t.local", RUTH, "200")
        assert connect("loop.t.local", RUTH) == "403", "still no loopback"
        write_list(*UNRESTRICTED)
        wait_status("other.t.local", RUTH, "403")
        assert connect("other.t.local", UMA) == "200", "control: uma is still listed"

    with subtest("squid starts on a missing list, treats every user as restricted, and recovers when it appears"):
        proxy.succeed(f"rm {LIST}")
        proxy.succeed("systemctl restart squid.service")
        proxy.wait_for_unit("squid.service")
        proxy.wait_for_open_port(3128, "${bridgeIp}")
        assert connect("forge.t.local", UMA) == "200", "logins work"
        assert connect("other.t.local", UMA) == "403", "no list, no unrestricted user"
        write_list(*UNRESTRICTED)
        wait_status("other.t.local", UMA, "200")

    with subtest("squid starts on a missing users file, refuses logins, and recovers when it appears"):
        proxy.succeed(f"rm {USERS}")
        proxy.succeed("systemctl restart squid.service")
        proxy.wait_for_unit("squid.service")
        proxy.wait_for_open_port(3128, "${bridgeIp}")
        assert connect("forge.t.local", ATLAS) == "407", "no file, no login"
        assert connect("other.t.local", UMA) == "407", "no file, no login, even when listed"
        write_users(ATLAS, UMA, SELFCHECK, SELFCHECK_U)
        assert connect("forge.t.local", ATLAS) == "200", "the file appeared"
        assert connect("other.t.local", UMA) == "200", "listed, and now logged in"
  '';
}