# `vmTests.<system>.hive-egress-antispoof`: the source check of
# ../host-modules/hive-egress.nix on a real agent bridge, with two nspawn
# containers on it that take their addresses from the hive's dnsmasq
# (../host-modules/hive-gateway/dnsmasq.nix' lease script, the real
# `hive-priv egress-load`, `egress-pin` and `egress-lease`). `hive-squid` cannot host this: its
# stand-in bridge is one NIC, so every agent shares one port.
#
# What it pins:
#   - each port is bound to its container's MAC and each lease is in the
#     table, without the test adding either;
#   - a container sending from its own address reaches the host and the other
#     container (the control for every drop below);
#   - another container's IPv4 address, another container's MAC, an IPv6 address
#     that is not link-local and a DHCP message with another's `chaddr` are
#     dropped, and each leaves a `hive-egress-spoof ` line that names the port;
#   - a DHCP renewal, a link-local IPv6 packet and a restored MAC go through;
#   - a port with no bound MAC sends nothing, and the timer binds it again;
#   - `antiSpoof = "log"`, reached by a reload of the unit, brings the bindings
#     and the leases with the new table (nothing is dropped in between),
#     lets the same spoofs through and logs them.
{ pkgs, hivePriv }:
let
  bridge = "hive0";
  gateway = "10.42.0.1";

  # A DHCP DISCOVER by hand, unicast to dnsmasq, with the `chaddr` given as
  # the argument. The container shares the host's store, so it runs from there.
  bootp = pkgs.writeText "bootp.pl" ''
    use IO::Socket::INET;
    use Socket;
    my $mac = pack("H12", join("", split(/:/, $ARGV[0])));
    my $s = IO::Socket::INET->new(Proto => "udp") or die "socket: $!";
    my $head = pack("C4 N n n a4 a4 a4 a4", 1, 1, 6, 0, 4660, 0, 0, "", "", "", "");
    my $msg = $head . $mac . ("\0" x 10) . ("\0" x 192) . pack("H8", "63825363") . pack("C4", 53, 1, 1, 255);
    $s->send($msg, 0, pack_sockaddr_in(67, inet_aton("${gateway}"))) or die "send: $!";
  '';

  hiveStubs =
    { lib, ... }:
    {
      options.services.hyperhive = {
        deploy.hive-controller.package = lib.mkOption { type = lib.types.package; };
        # Off: hive-priv.nix then adds only the link the pin command runs by.
        deploy.hive-controller.enable = lib.mkOption {
          type = lib.types.bool;
          default = false;
        };
        swarm.forge.domain = lib.mkOption {
          type = lib.types.str;
          default = "forge.t.local";
        };
        network = {
          enable = lib.mkOption {
            type = lib.types.bool;
            default = true;
          };
          bridgeName = lib.mkOption {
            type = lib.types.str;
            default = bridge;
          };
        };
      };
      config.services.hyperhive.deploy.hive-controller.package = hivePriv;
    };

  # A container that takes its address by DHCP, as the agents do.
  container = {
    autoStart = true;
    privateNetwork = true;
    hostBridge = bridge;
    config =
      { lib, pkgs, ... }:
      {
        system.stateVersion = "25.11";
        networking.useDHCP = lib.mkForce true;
        networking.dhcpcd.extraConfig = "noipv4ll";
        networking.dhcpcd.wait = "ipv4";
        networking.firewall.allowedUDPPorts = [ 68 ];
        environment.systemPackages = [
          pkgs.perl
          pkgs.socat
        ];
      };
  };
in
pkgs.testers.runNixOSTest {
  name = "hive-egress-antispoof";

  nodes.host =
    { config, lib, ... }:
    {
      imports = [
        hiveStubs
        ../host-modules/hive-egress.nix
        ../host-modules/hive-priv.nix
      ];
      virtualisation.memorySize = 2048;
      networking.bridges.${bridge}.interfaces = [ ];
      networking.interfaces.${bridge}.ipv4.addresses = [
        {
          address = gateway;
          prefixLength = 24;
        }
      ];
      # The test's own datagram listener, and dnsmasq's DHCP port.
      networking.firewall.interfaces.${bridge} = {
        allowedUDPPorts = [
          67
          9999
        ];
      };
      services.dnsmasq = {
        enable = true;
        resolveLocalQueries = false;
        settings = {
          interface = bridge;
          bind-interfaces = true;
          port = 0;
          dhcp-range = "10.42.0.2,10.42.0.254,1h";
          dhcp-leasefile = "/var/lib/dnsmasq/dnsmasq.leases";
          dhcp-script = toString config.services.hyperhive.network.egress.leaseScript;
          log-dhcp = true;
        };
      };
      systemd.services.dnsmasq = {
        wants = [ "network-addresses-${bridge}.service" ];
        after = [ "network-addresses-${bridge}.service" ];
      };
      environment.systemPackages = [ pkgs.socat ];

      containers.a = container;
      containers.b = container;

      specialisation.logmode.configuration = {
        services.hyperhive.network.egress.antiSpoof = lib.mkForce "log";
      };
    };

  testScript = ''
    import re
    import shlex

    TABLE = "bridge hive_egress"

    def ct(name, cmd):
        return host.succeed(f"nixos-container run {name} -- sh -c {shlex.quote(cmd)}").strip()

    def ct_ok(name, cmd):
        return host.execute(f"nixos-container run {name} -- sh -c {shlex.quote(cmd)}")[0] == 0

    def mac(name):
        return ct(name, "cat /sys/class/net/eth0/address")

    def ipv4(name):
        out = ct(name, "ip -4 -o addr show dev eth0 scope global")
        found = re.search(r"inet (10\.42\.0\.\d+)/", out)
        assert found, out
        return found.group(1)

    def spoof_lines(pattern=""):
        out = host.succeed("journalctl -k -o cat --no-pager | grep 'hive-egress-spoof ' || true")
        return [line for line in out.splitlines() if re.search(pattern, line)]

    def udp_reaches_host(name, src, v6=False):
        """One datagram from the container, sent from `src`. True when the host got it."""
        host.succeed("rm -f /tmp/rx")
        host.succeed(
            "systemd-run --unit=rx --collect "
            f"timeout 6 sh -c 'socat -u UDP{6 if v6 else 4}-RECV:9999 - > /tmp/rx'"
        )
        host.sleep(1)
        if v6:
            ll = host.succeed("ip -6 -o addr show dev ${bridge} scope link").split()[3].split("/")[0]
            target = f"UDP6-SENDTO:[{ll}%eth0]:9999,bind=[{src}]"
        else:
            target = f"UDP4-SENDTO:${gateway}:9999,bind={src}"
        ct_ok(name, f"echo hi | socat -u - {target}")
        host.sleep(1)
        got = host.succeed("cat /tmp/rx").strip() == "hi"
        host.execute("systemctl stop rx.service")
        return got

    host.start()
    host.wait_for_unit("multi-user.target")
    host.wait_for_unit("dnsmasq.service")
    host.wait_for_unit("hive-egress.service")
    for name in ("a", "b"):
        host.wait_for_unit(f"container@{name}.service")
        host.wait_until_succeeds(
            f"nixos-container run {name} -- sh -c 'ip -4 -o addr show dev eth0 | grep -q \"inet 10.42.0.\"'"
        )
    MAC_A, MAC_B = mac("a"), mac("b")
    IP_A, IP_B = ipv4("a"), ipv4("b")
    assert MAC_A != MAC_B and IP_A != IP_B

    with subtest("each port is bound to its container's MAC and each lease is loaded, with no help from the test"):
        host.wait_until_succeeds(f"nft list set {TABLE} port_mac | grep -q '\"vb-a\" . {MAC_A}'")
        host.wait_until_succeeds(f"nft list set {TABLE} port_mac | grep -q '\"vb-b\" . {MAC_B}'")
        host.wait_until_succeeds(f"nft list set {TABLE} lease_v4 | grep -q '{MAC_A} . {IP_A}'")
        host.wait_until_succeeds(f"nft list set {TABLE} lease_v4 | grep -q '{MAC_B} . {IP_B}'")
        assert not spoof_lines(), spoof_lines()

    with subtest("a container sending from its own address reaches the host and the other container"):
        assert ct_ok("a", "ping -c1 -W2 ${gateway}")
        assert ct_ok("a", f"ping -c1 -W2 {IP_B}")
        assert ct_ok("b", f"ping -c1 -W2 {IP_A}")
        assert udp_reaches_host("a", IP_A), "control: a datagram from a's own address arrives"
        assert not spoof_lines(), spoof_lines()

    with subtest("another container's IPv4 address is dropped, and the line names the port"):
        ct("a", f"ip addr add {IP_B}/24 dev eth0")
        assert not udp_reaches_host("a", IP_B), "a datagram from b's address must not arrive"
        host.wait_until_succeeds("journalctl -k -o cat --no-pager | grep -E 'hive-egress-spoof .*IN=vb-a .*SRC=" + IP_B.replace(".", "\\.") + " '")
        assert not spoof_lines(r"IN=vb-b "), "b sent nothing wrong"
        ct("a", f"ip addr del {IP_B}/24 dev eth0")
        assert udp_reaches_host("a", IP_A), "a's own address still works"

    with subtest("an IPv6 address that is not link-local is dropped, a link-local one is not"):
        ct("a", "ip -6 addr add fd00::bad/64 dev eth0 nodad")
        assert not udp_reaches_host("a", "fd00::bad", v6=True)
        host.wait_until_succeeds("journalctl -k -o cat --no-pager | grep -E 'hive-egress-spoof .*IN=vb-a .*SRC=fd00:0*:0*:0*:0*:0*:0*:0*bad '")
        ll_a = ct("a", "ip -6 -o addr show dev eth0 scope link").split()[3].split("/")[0]
        assert udp_reaches_host("a", ll_a, v6=True), "control: a link-local source goes through"
        ct("a", "ip -6 addr del fd00::bad/64 dev eth0")

    with subtest("another container's MAC is dropped, and the restored MAC works again"):
        ct("a", f"ip link set dev eth0 address {MAC_B}")
        assert not ct_ok("a", "ping -c1 -W2 ${gateway}")
        host.wait_until_succeeds("journalctl -k -o cat --no-pager | grep -E 'hive-egress-spoof .*IN=vb-a .*MAC=[0-9a-f:]*" + MAC_B + ":'")
        ct("a", f"ip link set dev eth0 address {MAC_A}")
        assert ct_ok("a", "ping -c1 -W2 ${gateway}"), "the port stays bound to the MAC it had"
        assert ct_ok("b", "ping -c1 -W2 ${gateway}"), "b was never affected"

    with subtest("another container's MAC and IPv4 address together are dropped, so the leased pair alone is no credential"):
        # A static neighbour entry for the host, so the datagram needs no ARP:
        # only the check of the port's MAC is left between it and the host.
        host_mac = host.succeed("cat /sys/class/net/${bridge}/address").strip()
        ct("a", f"ip neigh replace ${gateway} lladdr {host_mac} dev eth0 nud permanent")
        ct("a", f"ip link set dev eth0 address {MAC_B}")
        ct("a", f"ip addr add {IP_B}/24 dev eth0")
        seen = len(spoof_lines(r"IN=vb-a "))
        assert not udp_reaches_host("a", IP_B), "b's MAC and address from a's port must not arrive"
        assert len(spoof_lines(r"IN=vb-a ")) > seen, "and it is logged against a's port"
        ct("a", f"ip addr del {IP_B}/24 dev eth0")
        ct("a", f"ip link set dev eth0 address {MAC_A}")
        ct("a", "ip neigh del ${gateway} dev eth0")
        assert ct_ok("a", "ping -c1 -W2 ${gateway}"), "a works again with its own pair"

    with subtest("a DHCP message with another container's chaddr is dropped, one with its own is answered, and a renewal works"):
        def discovers(mac_addr):
            out = host.succeed("journalctl -u dnsmasq -o cat --no-pager")
            return len(re.findall(r"DHCPDISCOVER\(${bridge}\) " + mac_addr, out))
        before_b, before_a = discovers(MAC_B), discovers(MAC_A)
        ct("a", f"perl ${bootp} {MAC_B}")
        ct("a", f"perl ${bootp} {MAC_A}")
        host.wait_until_succeeds(f"journalctl -u dnsmasq -o cat --no-pager | grep -c 'DHCPDISCOVER(${bridge}) {MAC_A}' | grep -qv '^{before_a}$'")
        assert discovers(MAC_A) == before_a + 1, "the message with a's own chaddr reached dnsmasq"
        assert discovers(MAC_B) == before_b, "the message with b's chaddr did not"
        host.wait_until_succeeds("journalctl -k -o cat --no-pager | grep -E 'hive-egress-spoof .*IN=vb-a .*DPT=67 '")
        acks = host.succeed("journalctl -u dnsmasq -o cat --no-pager").count("DHCPACK(${bridge}) " + IP_A)
        ct("a", "dhcpcd --rebind eth0")
        host.wait_until_succeeds(
            "test $(journalctl -u dnsmasq -o cat --no-pager | grep -c 'DHCPACK(${bridge}) " + IP_A + "') -gt " + str(acks)
        )
        assert ipv4("a") == IP_A, "the renewal kept the address"
        host.wait_until_succeeds(f"nft list set {TABLE} lease_v4 | grep -q '{MAC_A} . {IP_A}'")
        assert ct_ok("a", "ping -c1 -W2 ${gateway}")

    with subtest("a port with no bound MAC sends nothing, and the timer binds it again"):
        host.succeed(f"nft flush set {TABLE} port_mac")
        assert not ct_ok("a", "ping -c1 -W2 ${gateway}")
        assert not ct_ok("b", "ping -c1 -W2 ${gateway}")
        # Nothing starts the pin unit by hand: the timer finds the unbound ports.
        host.wait_until_succeeds("nixos-container run a -- ping -c1 -W2 ${gateway}")
        host.wait_until_succeeds("nixos-container run b -- ping -c1 -W2 ${gateway}")

    with subtest("antiSpoof = log keeps the bindings and the leases, lets a spoofed datagram through and logs it"):
        host.succeed("/run/current-system/specialisation/logmode/bin/switch-to-configuration test")
        assert "counter accept" in host.succeed(f"nft list chain {TABLE} spoof"), "the table was reloaded in log mode"
        # No waiting: the reload put the sets in with the table.
        host.succeed(f"nft list set {TABLE} port_mac | grep -q '\"vb-a\" . {MAC_A}'")
        host.succeed(f"nft list set {TABLE} lease_v4 | grep -q '{MAC_B} . {IP_B}'")
        assert ct_ok("a", "ping -c1 -W2 ${gateway}"), "control: own address, right after the reload"
        seen = len(spoof_lines())
        ct("a", f"ip addr add {IP_B}/24 dev eth0")
        assert udp_reaches_host("a", IP_B), "log mode lets the spoofed datagram through"
        host.wait_until_succeeds("journalctl -k -o cat --no-pager | grep 'hive-egress-spoof ' | wc -l | grep -qv '^" + str(seen) + "$'")
        assert spoof_lines(r"IN=vb-a .*SRC=" + IP_B.replace(".", r"\.") + " ")
  '';
}