# `checks.module-eval-nats-authelia` — see ./lib.nix for the shared
# rationale (why this suite exists, naming convention, "evaluates
# not executes").
{
  pkgs,
  lib,
  self,
  nixosSystem,
  shared,
}:
let
  inherit
    (import ./lib.nix {
      inherit
        pkgs
        lib
        self
        nixosSystem
        ;
    })
    hive
    runGroup
    ;

  # No store identity, which the first case below relies on.
  natsQueueOnly = shared.natsQueueOnly;

  # The IdP and the queue on one host.
  autheliaAndNats = hive {
    deploy.authelia.enable = true;
    deploy.nats.enable = true;
    deploy.authelia.usersFile = "/var/lib/test-authelia/users.yml";
    deploy.authelia.package = pkgs.emptyDirectory;
    deploy.authelia.bridgePackage = pkgs.emptyDirectory;
  };

  # The queue with a store identity for its responder, placed by hand: the
  # queue host that is not the store's.
  natsWithStoreIdentity = hive {
    deploy.nats.enable = true;
    deploy.nats.authPackage = pkgs.emptyDirectory;
    deploy.nats.authBaoClientCertFile = "/etc/pki/nats-auth.pem";
    deploy.nats.authBaoClientKeyFile = "/etc/pki/nats-auth-key.pem";
  };

  # The queue on the store's own host, where the PKI glue mints every leaf.
  natsOnStoreHost = hive {
    deploy.bao.enable = true;
    deploy.nats.enable = true;
  };

  inherit (shared) natsOnStoreHostReaderMoved;

  # A queue host holding only the hive's own store reader, placed by hand: no
  # store here, and no responder leaf named.
  natsWithHiveReaderOnly = hive {
    deploy.nats.enable = true;
    deploy.nats.authPackage = pkgs.emptyDirectory;
    deploy.bao.clientCertFile = "/etc/pki/h1.pem";
    deploy.bao.clientKeyFile = "/etc/pki/h1-key.pem";
  };

  responderOf = h: h.containers.swarm-nats.config.systemd.services.swarm-nats-auth;

  refusedWithoutIdentity =
    m:
    lib.any (
      a: !a.assertion && lib.hasInfix "deploy.nats.authBaoClientCertFile and" a.message
    ) m.assertions;

  # A hive running NOTHING of the swarm's own services — no IdP here, no
  # `swarm.authelia.url` set by hand. The whole point of the fixture is what it
  # does *not* say: no IdP address is given and none is derived from
  # co-location.
  autheliaNotColocated = shared.bare;
  # The same swarm's IdP host, for the arm that the two agree.
  autheliaColocated = shared.autheliaOnly;

  # The other three queue clients with a container of their own, each forwarding
  # its journal as the queue's does. One shared host: the switch that turns on
  # every swarm service, which other suites already evaluate.
  servicesHost = shared.swarmServicesWithController;
  forwarderServices = [
    {
      machine = "swarm-builder";
      name = "builder";
      credential = "builder-bao";
      port = 8892;
    }
    {
      machine = "swarm-claude-login";
      name = "claude-login";
      credential = "claude-login-bao";
      port = 8893;
    }
    {
      machine = "swarm-review-classifier";
      name = "review-classifier";
      credential = "review-classifier-bao";
      port = 8894;
    }
  ];
  forwarderFetchDir = "/run/credentials/swarm-journal-forwarder-oidc.service";

  forwarderCases = lib.concatMap (
    svc:
    let
      h = servicesHost;
      container = h.containers.${svc.machine};
      c = container.config;
      s = c.services.opentelemetry-collector.settings;
      otel = h.services.hyperhive.swarm.otel;
      fetch = c.systemd.services.swarm-journal-forwarder-oidc;
      clientId = h.services.hyperhive.swarm.${svc.name}.clientId;
    in
    [
      {
        # The container is on the bridge, so its identity is a system credential
        # rather than a file: the fetch unit loads it, and reads the same client
        # secret the daemon's own identity is granted.
        name = "the ${svc.name} container forwards its journal to its own route on the swarm collector";
        ok =
          c.services.opentelemetry-collector.enable
          && s.exporters.otlphttp.endpoint == "https://${otel.domain}/${svc.machine}"
          && s.exporters.otlphttp.auth.authenticator == "oauth2client/swarm"
          && s.extensions."oauth2client/swarm".client_id == clientId
          && s.extensions."oauth2client/swarm".endpoint_params.audience == clientId
          && lib.elem "oauth2client/swarm" s.service.extensions
          && s.service.pipelines.logs.receivers == [ "journald" ]
          &&
            (lib.findFirst (a: a.key == "service.name") null s.processors.resource.attributes).value == svc.name
          && lib.elem "oidc-client-secret:/run/swarm-journal-forwarder-oidc/client-secret" c.systemd.services.opentelemetry-collector.serviceConfig.LoadCredential
          && fetch.environment.BAO_CLIENT_CERT == "${forwarderFetchDir}/${svc.credential}-cert"
          && fetch.environment.BAO_CLIENT_KEY == "${forwarderFetchDir}/${svc.credential}-key"
          && lib.elem "${svc.credential}-cert" fetch.serviceConfig.LoadCredential
          && lib.elem "${svc.credential}-key" fetch.serviceConfig.LoadCredential
          && lib.hasInfix "bao kv get -field=value secret/swarm/services/${clientId}/oidc/client" fetch.script
          # The guest keeps its journal; a host-linked one is empty inside. The
          # leaf's credential flag is what remains.
          && !(lib.elem "--link-journal=host" container.extraFlags)
          && lib.any (lib.hasPrefix "--load-credential=${svc.credential}-cert:") container.extraFlags;
      }
      {
        name = "the ${svc.name} forwarder's metrics port is its own and loopback";
        ok =
          let
            # Every other collector port on the host, and the queue's forwarder.
            taken = [
              8888
              8889
              8890
              h.services.hyperhive.swarm.nats.otel.telemetryPort
            ]
            ++ map (o: o.port) (lib.filter (o: o.machine != svc.machine) forwarderServices);
            reader = lib.head s.service.telemetry.metrics.readers;
          in
          c.services.hyperhive.swarmJournalForwarder.telemetryPort == svc.port
          && !(lib.elem svc.port taken)
          && reader.pull.exporter.prometheus.host == "127.0.0.1"
          && reader.pull.exporter.prometheus.port == svc.port;
      }
    ]
  ) forwarderServices;

  # The forwarders that authenticate as a client of their own rather than the
  # service's, each with the store identity the host already holds. The CI
  # runner's container cases are in ./forge-placement.nix, beside its fixture;
  # the homeserver's, which reads a bound-in identity, are below.
  ownClientForwarders = [
    {
      machine = "swarm-ci-runner";
      option = "ci-runner";
    }
    {
      machine = "swarm-grafana";
      option = "grafana";
    }
    {
      machine = "swarm-victorialogs";
      option = "victorialogs";
    }
    {
      machine = "hive-forge";
      option = "forge";
    }
    {
      machine = "hive-matrix";
      option = "matrix";
    }
    {
      machine = "swarm-victoriametrics";
      option = "victoriametrics";
    }
  ];
  ownClientIdOf = h: svc: h.services.hyperhive.swarm.${svc.option}.otel.clientId;

  # Those that share the host netns, on the shared services host.
  hostNetnsForwarders = [
    {
      machine = "swarm-grafana";
      option = "grafana";
      name = "grafana";
      cert = "grafana-bao-cert";
      key = "grafana-bao-key";
      port = 8896;
    }
    {
      machine = "swarm-victorialogs";
      option = "victorialogs";
      name = "victorialogs";
      cert = "bao-client.pem";
      key = "bao-client-key.pem";
      port = 8897;
    }
    {
      machine = "hive-forge";
      option = "forge";
      name = "forgejo";
      cert = "forge-bao-cert";
      key = "forge-bao-key";
      port = 8898;
    }
  ];

  hostNetnsForwarderCases = lib.concatMap (
    svc:
    let
      h = servicesHost;
      container = h.containers.${svc.machine};
      c = container.config;
      s = c.services.opentelemetry-collector.settings;
      otel = h.services.hyperhive.swarm.otel;
      fetch = c.systemd.services.swarm-journal-forwarder-oidc;
      clientId = ownClientIdOf h svc;
    in
    [
      {
        name = "the ${svc.name} container forwards its journal to its own route as its own client";
        ok =
          c.services.opentelemetry-collector.enable
          && s.exporters.otlphttp.endpoint == "https://${otel.domain}/${svc.machine}"
          && s.extensions."oauth2client/swarm".client_id == clientId
          && s.extensions."oauth2client/swarm".endpoint_params.audience == clientId
          && otel.serviceProducers.${svc.machine}.clientId == clientId
          &&
            (lib.findFirst (a: a.key == "service.name") null s.processors.resource.attributes).value == svc.name
          && fetch.environment.BAO_CLIENT_CERT == "${forwarderFetchDir}/${svc.cert}"
          && fetch.environment.BAO_CLIENT_KEY == "${forwarderFetchDir}/${svc.key}"
          && lib.elem svc.cert fetch.serviceConfig.LoadCredential
          && lib.elem svc.key fetch.serviceConfig.LoadCredential
          && lib.hasInfix "bao kv get -field=value secret/swarm/services/${clientId}/oidc/client" fetch.script
          && !(lib.elem "--link-journal=host" container.extraFlags)
          && lib.any (lib.hasPrefix "--load-credential=${svc.cert}:") container.extraFlags
          && lib.any (lib.hasPrefix "--load-credential=${svc.key}:") container.extraFlags;
      }
      {
        # Shared host netns: one port space with every other collector there.
        name = "the ${svc.name} forwarder's metrics port is free in the host netns and loopback";
        ok =
          let
            taken = [
              8888
              8889
              8890
              h.services.hyperhive.swarm.nats.otel.telemetryPort
            ]
            ++ map (o: o.port) (lib.filter (o: o.machine != svc.machine) hostNetnsForwarders);
            reader = lib.head s.service.telemetry.metrics.readers;
          in
          c.services.hyperhive.swarmJournalForwarder.telemetryPort == svc.port
          && !(lib.elem svc.port taken)
          && reader.pull.exporter.prometheus.host == "127.0.0.1"
          && reader.pull.exporter.prometheus.port == svc.port;
      }
    ]
  ) hostNetnsForwarders;

  cases = [
    {
      name = "authelia and nats declare no scrape target on a host with no collector";
      ok =
        !(autheliaAndNats.services.hyperhive.deploy.swarm-otel.enable)
        && !(autheliaAndNats.services.hyperhive.swarm.otel.scrapeTargets ? authelia)
        && !(autheliaAndNats.services.hyperhive.swarm.otel.scrapeTargets ? nats);
    }
    {
      # The xkey and the responder's OIDC client secret are read only from
      # the store, so a queue with no responder leaf cannot start. Refused at
      # eval; the controls are a leaf named by hand and the store host's
      # minted one.
      name = "a queue with no responder store identity is refused at eval, one with a leaf is not";
      ok =
        refusedWithoutIdentity natsQueueOnly
        && !(refusedWithoutIdentity natsWithStoreIdentity)
        && !(refusedWithoutIdentity natsOnStoreHost);
    }
    {
      # The hive's own reader is not the responder's identity, and no path is
      # derived from it: a queue host without the store that holds only the
      # hive's reader is refused like one holding nothing.
      name = "a queue host without the store is refused when it holds only the hive's reader";
      ok =
        refusedWithoutIdentity natsWithHiveReaderOnly
        && natsWithHiveReaderOnly.services.hyperhive.deploy.nats.authBaoClientCertFile == null;
    }
    {
      # Reads the RENDERED settings, not the option: `calloutBlocks {…} // {
      # … }` is a shallow merge, and a future edit that dropped or shadowed
      # this key would still evaluate cleanly — the only reader that would
      # notice is a publisher whose row exceeds upstream's much smaller
      # default, and by then it is a dropped row, not an eval failure.
      # Piggybacks on the queue-only fixture above, which already
      # renders this container's full config.
      name = "the queue's payload ceiling is set, not inherited from the server's default";
      ok = natsQueueOnly.containers.swarm-nats.config.services.nats.settings.max_payload == 8388608;
    }
    {
      # What keeps a callout request inside one principal: the responder is
      # the callout account's only member, and no system account exists to
      # observe it from. The runtime fragment is rendered by the same
      # `calloutBlocks`, so the count is the same there. The account name is
      # the control that the assertions read the callout block at all.
      name = "the callout account has exactly one member and no system account is configured";
      ok =
        lib.all
          (
            h:
            let
              settings = h.containers.swarm-nats.config.services.nats.settings;
            in
            settings.authorization.auth_callout.account == "AUTH"
            && lib.length settings.accounts.AUTH.users == 1
            && !(settings ? system_account)
          )
          [
            natsQueueOnly
            natsOnStoreHost
          ];
    }
    {
      # The xkey exists only at runtime, so the server's config is rendered as
      # it starts: always the wrapper, and never the store-rendered settings
      # alone, which carry no xkey. `+` so the render may write `runtimeDir`.
      name = "the queue's server starts only through the config rendered with its xkey";
      ok =
        lib.all
          (
            h:
            let
              c = h.containers.swarm-nats.config;
              nats = c.systemd.services.nats.serviceConfig;
              pre = nats.ExecStartPre or "";
            in
            lib.hasSuffix "/bin/nats-server -c /var/lib/nats-callout/nats.conf" nats.ExecStart
            && lib.hasPrefix "+/nix/store/" pre
            && lib.hasSuffix "-swarm-nats-render-config" pre
            && !(c.services.nats.settings.authorization.auth_callout ? xkey)
          )
          [
            natsQueueOnly
            natsOnStoreHost
          ];
    }
    {
      # A server that starts late, once its xkey arrives, must bring the
      # responder with it: the responder requires the server, so a server that
      # failed to start left the responder inactive rather than failed.
      name = "starting the queue's server starts its responder";
      ok = lib.elem "nats.service" (responderOf natsQueueOnly).wantedBy;
    }
    {
      # The keys are created before anything needs them, by the responder's
      # own verb under its own leaf, then each public half is read back from
      # the store and left where the server's render reads it. Before the
      # container and wanted by it, and a fetch that finds nothing fails so
      # `storeRetry` retries.
      name = "the callout keys are created, then their public halves fetched from the store before the queue's container starts";
      ok =
        let
          u = natsWithStoreIdentity.systemd.services.swarm-bao-nats-callout-xkey;
          s = builtins.unsafeDiscardStringContext u.script;
          at = needle: lib.stringLength (lib.head (lib.splitString needle s));
          ensure = "swarm-nats-auth ensure-callout-keys --store-cert-role swarm-nats-auth";
          fetched =
            leaf: file:
            let
              fetch = "bao kv get -field=value secret/swarm/public/swarm-nats/${leaf}";
              path = "/var/lib/nixos-containers/swarm-nats/var/lib/nats-callout/${file}";
            in
            lib.hasInfix fetch s
            && at ensure < at fetch
            && lib.hasInfix "atomic_write_secret 0444 \"\" ${path}" s
            && lib.hasInfix "refresh_consumer swarm-nats nats.service ${path}" s;
        in
        lib.elem "container@swarm-nats.service" u.before
        && lib.elem "container@swarm-nats.service" u.wantedBy
        && u.environment.BAO_CLIENT_CERT == "/etc/pki/nats-auth.pem"
        && lib.hasInfix ensure s
        && fetched "callout-xkey" "callout-xkey.pub"
        && fetched "callout-issuer" "issuer.pub"
        && fetched "callout-user" "callout-user.pub"
        && !(lib.hasInfix "exit 0" s)
        && u.serviceConfig.Restart == "on-failure";
    }
    {
      # The seeds a host kept on disk are what the verb stores when the store
      # holds none, so the keys keep their public halves; the verb removes
      # them itself. Nothing mints a seed on the host, and no
      # tmpfiles rule touches the host seeds, which could run before the
      # import does.
      name = "the callout keys import the host's seed files, which only the importing unit removes";
      ok =
        let
          s = natsWithStoreIdentity.systemd.services.swarm-bao-nats-callout-xkey.script;
          rules = natsWithStoreIdentity.systemd.tmpfiles.rules;
        in
        lib.hasInfix "--import-issuer-seed-file /var/lib/swarm-nats-callout/issuer.seed" s
        && lib.hasInfix "--import-user-seed-file /var/lib/swarm-nats-callout/callout-user.seed" s
        && !(natsWithStoreIdentity.systemd.services ? swarm-nats-callout-keys)
        && !(lib.any (lib.hasInfix "/var/lib/swarm-nats-callout") rules);
    }
    {
      # Its creates are what the responder's role was granted, and the
      # publisher's policy is what denies the publisher the seeds' paths: a
      # seed found there is kept. On the store's host both units run first.
      name = "on the store's host the callout key fetch waits for the responder's role and the publisher's denial";
      ok =
        let
          u = natsOnStoreHost.systemd.services.swarm-bao-nats-callout-xkey;
          waitsFor = p: lib.elem "${p}.service" u.after && lib.elem "${p}.service" u.wants;
        in
        u.environment.BAO_CLIENT_CERT == "/var/lib/swarm-bao-pki/nats-auth.pem"
        && waitsFor "swarm-bao-nats-auth-policy"
        && waitsFor "swarm-bao-secret-publisher-policy"
        && natsOnStoreHost.systemd.services ? swarm-bao-secret-publisher-policy;
    }
    {
      # The queue's container pulls the copy in, outside the switch's
      # `multi-user.target` transaction; see the same case in ./builder.nix.
      name = "the responder's leaf copy runs the leaf-minting unit first";
      ok =
        let
          u = natsOnStoreHost.systemd.services.swarm-nats-auth-secrets;
        in
        lib.elem "swarm-bao-pki.service" u.wants && lib.elem "swarm-bao-pki.service" u.after;
    }
    {
      # The control is the fixture above the store: off the store's host
      # there is no local policy unit to wait for.
      name = "off the store's host the callout key fetch waits for no policy unit";
      ok =
        let
          u = natsWithStoreIdentity.systemd.services.swarm-bao-nats-callout-xkey;
        in
        !(lib.any (lib.hasSuffix "-policy.service") (u.after ++ u.wants));
    }
    {
      # The responder refuses the hive's shared agent client and takes no flag
      # granting it subjects: an `--agent-publish-subject` fails the unit at
      # start.
      name = "the responder grants the hive's shared agent client nothing";
      ok =
        let
          exec =
            natsQueueOnly.containers.swarm-nats.config.systemd.services.swarm-nats-auth.serviceConfig.ExecStart;
        in
        !(lib.hasInfix "--agent-publish-subject " exec);
    }
    {
      # The per-agent grant: the terminal and agent-state streams keyed on the
      # agent alone, with the doubled dollar, and the role the store writes for
      # it.
      name = "the responder grants a verified agent its own hive-free subjects";
      ok =
        let
          exec = (responderOf natsQueueOnly).serviceConfig.ExecStart;
        in
        lib.hasInfix "--agent-token-publish-subject '$$SWARM.term.{agent}'" exec
        && lib.hasInfix "--agent-token-publish-subject '$$SWARM.agent-state.{agent}'" exec
        && lib.hasInfix "--store-cert-role swarm-nats-auth" exec;
    }
    {
      # The whole per-agent grant, compared as a list rather than by infix: a
      # wider subject added beside these (`$$JS.API.>`, another stream's name)
      # would pass every presence check above.
      name = "a verified agent's grant is exactly its own subjects, and its JetStream API subjects read its own schedule, inference and permissions keys";
      ok =
        let
          args = lib.splitString " " (responderOf natsQueueOnly).serviceConfig.ExecStart;
          granted = lib.concatLists (
            lib.imap0 (
              i: a:
              lib.optional (a == "--agent-token-publish-subject" && i + 1 < lib.length args) (
                lib.elemAt args (i + 1)
              )
            ) args
          );
        in
        granted == [
          "'$$SWARM.term.{agent}'"
          "'$$SWARM.agent-state.{agent}'"
          "'$$KV.agent-icons.{agent}'"
          "'$$SWARM.term.{agent}.sub.>'"
          "'$$JS.API.DIRECT.GET.KV_agent-schedules.$$KV.agent-schedules.{agent}'"
          "'$$JS.API.DIRECT.GET.KV_agent-inference.$$KV.agent-inference.{agent}'"
          "'$$JS.API.DIRECT.GET.KV_agent-permissions.$$KV.agent-permissions.{agent}'"
        ];
    }
    {
      # Each credential by `LoadCredential`, and the environment naming where
      # the unit sees it: a DynamicUser cannot read the copies directly.
      name = "a responder with a store identity loads it and is pointed at the store";
      ok =
        let
          u = responderOf natsWithStoreIdentity;
          creds = u.serviceConfig.LoadCredential;
        in
        lib.elem "bao-client.pem:/var/lib/swarm-nats-auth/bao-client.pem" creds
        && lib.elem "bao-client-key.pem:/var/lib/swarm-nats-auth/bao-client-key.pem" creds
        && u.environment.BAO_CLIENT_CERT == "/run/credentials/swarm-nats-auth.service/bao-client.pem"
        && u.environment.BAO_CLIENT_KEY == "/run/credentials/swarm-nats-auth.service/bao-client-key.pem"
        && lib.hasPrefix "https://" u.environment.BAO_ADDR;
    }
    {
      # Copied so an absent leaf fails the unit rather than reaching the
      # responder as a file it cannot log in with. No seed is copied: the
      # responder reads every callout seed from the store.
      name = "the copy unit delivers the store identity and no seed, and nothing in its place when absent";
      ok =
        let
          s = natsWithStoreIdentity.systemd.services.swarm-nats-auth-secrets.script;
        in
        lib.hasInfix "install -m 0400 /etc/pki/nats-auth.pem " s
        && lib.hasInfix "install -m 0400 /etc/pki/nats-auth-key.pem " s
        && !(lib.hasInfix "/dev/null" s)
        && !(lib.hasInfix ".seed" s);
    }
    {
      # The responder names no seed file, by flag or by credential, and the
      # copies a container may still hold are removed. The store identity is
      # the control that the credential list is read at all.
      name = "the responder takes no seed from a file, and the container's seed copies are removed";
      ok =
        let
          u = responderOf natsWithStoreIdentity;
          creds = u.serviceConfig.LoadCredential;
          dir = "/var/lib/nixos-containers/swarm-nats/var/lib/swarm-nats-auth";
          rules = natsWithStoreIdentity.systemd.tmpfiles.rules;
        in
        lib.elem "bao-client.pem:/var/lib/swarm-nats-auth/bao-client.pem" creds
        && !(lib.any (lib.hasInfix ".seed") creds)
        && !(lib.hasInfix "seed-file" u.serviceConfig.ExecStart)
        && lib.elem "r ${dir}/issuer.seed" rules
        && lib.elem "r ${dir}/callout-user.seed" rules;
    }
    {
      # Under either namespace, each removed option fails evaluation naming
      # itself and the store. The control is a queue host that sets none.
      name = "a definition of a removed callout option fails evaluation, naming the option and the secret store";
      ok =
        let
          refused =
            m: path:
            lib.any (
              a:
              !a.assertion
              && lib.hasInfix "services.hyperhive.${path}" a.message
              && lib.hasInfix "live in the swarm secret store" a.message
            ) m.assertions;
          removedOptions = {
            autoGenerateCallout = true;
            calloutUserPublicKey = "U";
            calloutIssuerPublicKey = "A";
            calloutUserSeedFile = "/run/u.seed";
            calloutIssuerSeedFile = "/run/i.seed";
          };
          setting = lib.listToAttrs (
            lib.concatMap
              (
                ns:
                lib.mapAttrsToList (name: value: {
                  name = "${ns}.nats.${name}";
                  value = hive { ${ns}.nats.${name} = value; };
                }) removedOptions
              )
              [
                "deploy"
                "swarm"
              ]
          );
        in
        lib.length (lib.attrNames setting) == 10
        && lib.all (path: refused setting.${path} path) (lib.attrNames setting)
        && !(lib.any (path: refused natsWithStoreIdentity path) (lib.attrNames setting));
    }
    {
      # The responder reads its OIDC secret from the store itself, so no host
      # unit fetches it, no credential or flag carries it, and a copy a
      # container still holds is removed, also on a host with the queue off.
      # On or off the store's host alike; the callout key fetch beside it is
      # the control that the unit set is read at all, and the store cert role
      # the control on the flags.
      name = "the responder's OIDC secret is fetched by no unit and carried by no credential or flag, and the container's copy is removed";
      ok =
        let
          fetchedNowhere =
            h:
            !(h.systemd.services ? swarm-bao-nats-auth-oidc)
            && h.systemd.services ? swarm-bao-nats-callout-xkey;
          u = responderOf natsWithStoreIdentity;
          copy = natsWithStoreIdentity.systemd.services.swarm-nats-auth-secrets.script;
          removesCopy = h: lib.elem "r ${dir}/oidc-client.secret" h.systemd.tmpfiles.rules;
          dir = "/var/lib/nixos-containers/swarm-nats/var/lib/swarm-nats-auth";
        in
        fetchedNowhere natsWithStoreIdentity
        && fetchedNowhere natsOnStoreHost
        && !(lib.any (lib.hasInfix "oidc-client") u.serviceConfig.LoadCredential)
        && !(lib.hasInfix "client-secret" u.serviceConfig.ExecStart)
        && lib.hasInfix "--store-cert-role swarm-nats-auth" u.serviceConfig.ExecStart
        && removesCopy natsWithStoreIdentity
        && removesCopy autheliaNotColocated
        && !(lib.hasInfix "oidc-client" copy)
        && !(lib.hasInfix "authelia" copy);
    }
    {
      # On the store's host the glue pairs the responder with a leaf of its
      # own, never the queue's TLS-issuing one or the hive's.
      name = "on the store's host the responder presents its own leaf";
      ok =
        let
          n = natsOnStoreHost.services.hyperhive.deploy.nats;
          b = natsOnStoreHost.services.hyperhive.deploy.bao;
        in
        n.authBaoClientCertFile == "/var/lib/swarm-bao-pki/nats-auth.pem"
        && n.authBaoClientKeyFile == "/var/lib/swarm-bao-pki/nats-auth-key.pem"
        && n.authBaoClientCertFile != n.baoClientCertFile
        && n.authBaoClientCertFile != b.clientCertFile;
    }
    {
      # The default is the file the minting unit writes, wherever the hive's
      # own reader is pointed.
      name = "on the store's host the responder's leaf is the one minted there, not one beside the hive's reader";
      ok =
        let
          n = natsOnStoreHostReaderMoved.services.hyperhive.deploy.nats;
          mint = natsOnStoreHostReaderMoved.systemd.services.swarm-bao-pki.script;
        in
        n.authBaoClientCertFile == "/var/lib/swarm-bao-pki/nats-auth.pem"
        && n.authBaoClientKeyFile == "/var/lib/swarm-bao-pki/nats-auth-key.pem"
        && lib.hasInfix "[ -s /var/lib/swarm-bao-pki/nats-auth.pem ] ||" mint
        && lib.hasInfix "/var/lib/swarm-bao-pki nats-auth " mint;
    }
    {
      # authelia exits at startup on `users: {}`, so an empty database, and a
      # file holding exactly that document, must be replaced by the disabled
      # placeholder rather than left for authelia to read.
      name = "an empty authelia users database is seeded with the disabled placeholder";
      ok =
        let
          s =
            autheliaColocated.containers.swarm-authelia.config.systemd.services.authelia-swarm-secrets.script;
        in
        lib.hasInfix ''if [ ! -s "$users" ] || [ "$(cat "$users")" = "users: {}" ]; then'' s
        && lib.hasInfix "-authelia-users-placeholder.yml \"$users\"" s
        && !(lib.hasInfix ''echo "users: {}"'' s);
    }
    {
      # The bridge refuses every client but the one named here, so this has
      # to be the client the controller authenticates as. A drift is a
      # controller whose every bridge call is refused.
      name = "the authelia bridge answers the controller's client only";
      ok =
        let
          env =
            autheliaColocated.containers.swarm-authelia.config.systemd.services.swarm-authelia-bridge.environment;
        in
        env.SWARM_AUTHELIA_BRIDGE_CONTROLLER_CLIENT_ID
        == autheliaColocated.services.hyperhive.swarm.controller.queueClientId
        && env.SWARM_AUTHELIA_BRIDGE_OPERATOR_GROUP == "admins";
    }
    {
      # One address from every host, and a location the IdP host serves: a
      # controller off the authelia host has no other way to the bridge.
      name = "the auth bridge address is the authelia vhost's /bridge/ on every host";
      ok =
        let
          url = h: h.services.hyperhive.deploy.swarm-controller.authBridgeUrl;
          authelia = autheliaColocated.services.hyperhive.swarm.authelia;
          location = autheliaColocated.services.nginx.virtualHosts.${authelia.domain}.locations."/bridge/";
        in
        url autheliaNotColocated == "${autheliaNotColocated.services.hyperhive.swarm.authelia.url}/bridge"
        && url autheliaColocated == url autheliaNotColocated
        && location.proxyPass == "http://127.0.0.1:${toString authelia.bridgePort}/";
    }
    {
      # Config NAMES the IdP; it never computes where the IdP is. Both arms
      # matter together: the address is the swarm's name on a hive that runs
      # nothing, and it is the SAME string on the hive that serves the vhost —
      # so a re-introduced co-location branch shows up as the two disagreeing
      # rather than as a value that merely looks plausible on one of them.
      name = "the swarm IdP address is its domain on every hive, co-located or not";
      ok =
        autheliaNotColocated.services.hyperhive.swarm.authelia.url
        == "https://${autheliaNotColocated.services.hyperhive.swarm.authelia.domain}"
        &&
          autheliaColocated.services.hyperhive.swarm.authelia.url
          == autheliaNotColocated.services.hyperhive.swarm.authelia.url;
    }
    {
      # The queue container's own journal forwarder, on the store's host and
      # on a queue host with no store and no swarm collector: it exports to
      # the swarm collector's `swarm-nats` route by name, authenticated as
      # the queue's client, and its secret is read with the responder's
      # store identity inside the container.
      name = "the queue container forwards its journal to its own route on the swarm collector";
      ok =
        lib.all
          (
            h:
            let
              c = h.containers.swarm-nats.config;
              s = c.services.opentelemetry-collector.settings;
              otel = h.services.hyperhive.swarm.otel;
              fetch = c.systemd.services.swarm-journal-forwarder-oidc;
              clientId = h.services.hyperhive.swarm.nats.clientId;
            in
            c.services.opentelemetry-collector.enable
            && s.exporters.otlphttp.endpoint == "https://${otel.domain}/swarm-nats"
            && s.exporters.otlphttp.auth.authenticator == "oauth2client/swarm"
            && s.extensions."oauth2client/swarm".client_id == clientId
            && s.extensions."oauth2client/swarm".endpoint_params.audience == clientId
            && lib.elem "oauth2client/swarm" s.service.extensions
            && s.service.pipelines.logs.receivers == [ "journald" ]
            && lib.elem "oidc-client-secret:/run/swarm-journal-forwarder-oidc/client-secret" c.systemd.services.opentelemetry-collector.serviceConfig.LoadCredential
            && fetch.environment.BAO_CLIENT_CERT == "/var/lib/swarm-nats-auth/bao-client.pem"
            && lib.hasInfix "bao kv get -field=value secret/swarm/services/${clientId}/oidc/client" fetch.script
            # The guest keeps its journal; a host-linked one is empty inside.
            && !(lib.elem "--link-journal=host" h.containers.swarm-nats.extraFlags)
          )
          [
            natsOnStoreHost
            natsWithStoreIdentity
          ];
    }
    {
      # The far end: a receiver, authenticator, pipeline entry and gateway
      # route of its own, checking the queue client as audience. Rendered on
      # a collector host that runs no queue.
      name = "the swarm collector admits the queue's forwarder on a receiver of its own";
      ok =
        let
          h = shared.baoWithCollector;
          otel = h.services.hyperhive.swarm.otel;
          s = h.containers.swarm-otel.config.services.opentelemetry-collector.settings;
        in
        !h.services.hyperhive.deploy.nats.enable
        && s.receivers."otlp/swarm-nats".protocols.http.auth.authenticator == "oidc/swarm-nats"
        && s.extensions."oidc/swarm-nats".audience == h.services.hyperhive.swarm.nats.clientId
        && lib.elem "oidc/swarm-nats" s.service.extensions
        && lib.elem "otlp/swarm-nats" s.service.pipelines."logs/${otel.producerName}".receivers
        && h.services.nginx.virtualHosts.${otel.domain}.locations ? "/swarm-nats/"
        # The store's forwarder keeps its own receiver beside it.
        && s.extensions."oidc/swarm-bao".audience == h.services.hyperhive.swarm.bao.otel.clientId;
    }
    {
      # The token the forwarder mints must name its own client as audience
      # and be a JWT, or the receiver refuses it.
      name = "the queue's client is registered with its own audience and a signed access token";
      ok =
        let
          clients = shared.autheliaOnly.services.hyperhive.swarm.authelia.oidc.clients;
          c = lib.findFirst (c: c.id == "swarm-nats") null clients;
        in
        c != null && c.audience == [ "swarm-nats" ] && c.accessTokenSignedResponseAlg == "RS256";
    }
    {
      # The far end for the three: a receiver, authenticator, pipeline entry and
      # gateway route each, checking that service's client as audience, beside
      # the queue's, on a collector host that runs none of them.
      name = "the swarm collector admits the builder, claude login and classifier forwarders on receivers of their own";
      ok =
        let
          h = shared.baoWithCollector;
          otel = h.services.hyperhive.swarm.otel;
          s = h.containers.swarm-otel.config.services.opentelemetry-collector.settings;
          clientOf = svc: h.services.hyperhive.swarm.${svc.name}.clientId;
          ports = map (svc: otel.serviceProducers.${svc.machine}.port) forwarderServices;
        in
        !h.services.hyperhive.deploy.builder.enable
        && lib.all (
          svc:
          s.receivers."otlp/${svc.machine}".protocols.http.auth.authenticator == "oidc/${svc.machine}"
          && s.extensions."oidc/${svc.machine}".audience == clientOf svc
          && lib.elem "oidc/${svc.machine}" s.service.extensions
          && lib.elem "otlp/${svc.machine}" s.service.pipelines."logs/${otel.producerName}".receivers
          && h.services.nginx.virtualHosts.${otel.domain}.locations ? "/${svc.machine}/"
        ) forwarderServices
        # Four producers' receivers on the collector's host, none sharing a port.
        && lib.length (lib.unique (ports ++ [ otel.serviceProducers.swarm-nats.port ])) == 4;
    }
    {
      # The token each forwarder mints must name its own client as audience and
      # be a JWT, or the receiver refuses it.
      name = "the builder, claude login and classifier clients are registered with their own audience and a signed access token";
      ok =
        let
          clients = shared.autheliaOnly.services.hyperhive.swarm.authelia.oidc.clients;
          ok1 =
            id:
            let
              c = lib.findFirst (c: c.id == id) null clients;
            in
            c != null
            && c.audience == [ id ]
            && c.accessTokenSignedResponseAlg == "RS256"
            && c.kind == "machine";
        in
        lib.all ok1 (map (svc: svc.machine) forwarderServices);
    }
    {
      name = "the swarm collector admits the CI runner, Grafana, log store, forge and homeserver forwarders on receivers of their own";
      ok =
        let
          h = shared.baoWithCollector;
          otel = h.services.hyperhive.swarm.otel;
          s = h.containers.swarm-otel.config.services.opentelemetry-collector.settings;
          ports = map (p: p.port) (lib.attrValues otel.serviceProducers);
        in
        lib.all (
          svc:
          s.receivers."otlp/${svc.machine}".protocols.http.auth.authenticator == "oidc/${svc.machine}"
          && s.extensions."oidc/${svc.machine}".audience == ownClientIdOf h svc
          && lib.elem "oidc/${svc.machine}" s.service.extensions
          && lib.elem "otlp/${svc.machine}" s.service.pipelines."logs/${otel.producerName}".receivers
          && h.services.nginx.virtualHosts.${otel.domain}.locations ? "/${svc.machine}/"
        ) ownClientForwarders
        && lib.length (lib.unique ports) == lib.length ports;
    }
    {
      # Grafana's forwarder is a second client beside its browser login, which
      # stays an interactive client.
      name = "the CI runner, Grafana, log store, forge and homeserver forwarder clients are registered as machine clients with their own audience and a signed access token";
      ok =
        let
          h = shared.autheliaOnly;
          clients = h.services.hyperhive.swarm.authelia.oidc.clients;
          find = id: lib.findFirst (c: c.id == id) null clients;
          ok1 =
            svc:
            let
              id = ownClientIdOf h svc;
              c = find id;
            in
            c != null
            && c.audience == [ id ]
            && c.accessTokenSignedResponseAlg == "RS256"
            && c.kind == "machine";
          browser = find h.services.hyperhive.swarm.grafana.oidc.clientId;
        in
        lib.all ok1 ownClientForwarders
        &&
          h.services.hyperhive.swarm.grafana.otel.clientId != h.services.hyperhive.swarm.grafana.oidc.clientId
        && browser != null
        && browser.kind != "machine";
    }
    {
      # The homeserver's container already holds matrix-ctl's identity, bound
      # in read-only, and its forwarder fetches with those paths rather than a
      # system credential.
      name = "the homeserver container forwards its journal to its own route as its own client";
      ok =
        let
          h = servicesHost;
          container = h.containers.hive-matrix;
          c = container.config;
          s = c.services.opentelemetry-collector.settings;
          otel = h.services.hyperhive.swarm.otel;
          fetch = c.systemd.services.swarm-journal-forwarder-oidc;
          clientId = h.services.hyperhive.swarm.matrix.otel.clientId;
          ctlCert = h.services.hyperhive.deploy.matrix.ctlBaoClientCertFile;
        in
        c.services.opentelemetry-collector.enable
        && s.exporters.otlphttp.endpoint == "https://${otel.domain}/hive-matrix"
        && s.extensions."oauth2client/swarm".client_id == clientId
        && s.extensions."oauth2client/swarm".endpoint_params.audience == clientId
        && otel.serviceProducers.hive-matrix.clientId == clientId
        &&
          (lib.findFirst (a: a.key == "service.name") null s.processors.resource.attributes).value
          == "tuwunel"
        && fetch.environment.BAO_CLIENT_CERT == ctlCert
        && fetch.environment.BAO_CLIENT_KEY == h.services.hyperhive.deploy.matrix.ctlBaoClientKeyFile
        && container.bindMounts ? ${builtins.dirOf ctlCert}
        && lib.hasInfix "bao kv get -field=value secret/swarm/services/${clientId}/oidc/client" fetch.script
        && !(lib.elem "--link-journal=host" container.extraFlags);
    }
    {
      # With no matrix-ctl identity there is nothing to fetch the secret with.
      name = "a homeserver container with no store identity runs no forwarder";
      ok =
        !shared.matrixNoBaoIdentity.containers.hive-matrix.config.services.opentelemetry-collector.enable;
    }
    {
      # The metrics store's container has no network: the forwarder dials the
      # relay on loopback and reads a secret the host fetched and bound in.
      name = "the metrics store container forwards its journal through the relay as its own client";
      ok =
        let
          h = servicesHost;
          container = h.containers.swarm-victoriametrics;
          c = container.config;
          s = c.services.opentelemetry-collector.settings;
          otel = h.services.hyperhive.swarm.otel;
          clientId = h.services.hyperhive.swarm.victoriametrics.otel.clientId;
          secret = "/run/swarm-victoriametrics-forwarder-oidc/${clientId}.secret";
          fetch = h.systemd.services.swarm-victoriametrics-forwarder-oidc;
        in
        c.services.opentelemetry-collector.enable
        && s.exporters.otlphttp.endpoint == "http://127.0.0.1:4380/swarm-victoriametrics"
        && s.extensions."oauth2client/swarm".token_url == "http://127.0.0.1:4380/api/oidc/token"
        && s.extensions."oauth2client/swarm".client_id == clientId
        && s.extensions."oauth2client/swarm".endpoint_params.audience == clientId
        && otel.serviceProducers.swarm-victoriametrics.clientId == clientId
        &&
          (lib.findFirst (a: a.key == "service.name") null s.processors.resource.attributes).value
          == "victoriametrics"
        && lib.elem "oidc-client-secret:${secret}" c.systemd.services.opentelemetry-collector.serviceConfig.LoadCredential
        && !(c.systemd.services ? swarm-journal-forwarder-oidc)
        && container.bindMounts."/run/swarm-victoriametrics-forwarder-oidc".isReadOnly
        && !(lib.elem "--link-journal=host" (container.extraFlags or [ ]))
        &&
          fetch.environment.BAO_CLIENT_CERT
          == h.services.hyperhive.deploy.victoriametrics.authBaoClientCertFile
        &&
          fetch.environment.BAO_CLIENT_KEY == h.services.hyperhive.deploy.victoriametrics.authBaoClientKeyFile
        && lib.hasInfix "bao kv get -field=value secret/swarm/services/${clientId}/oidc/client" fetch.script
        && lib.hasInfix "refresh_consumer swarm-victoriametrics opentelemetry-collector.service" fetch.script;
    }
    {
      # The relay: a loopback listener in the container, a proxy to the bound
      # socket, and the host nginx server on that socket, which opens the
      # forwarder's two routes and nothing else.
      name = "the metrics store's egress relay leads loopback to the collector's route and authelia's token endpoint only";
      ok =
        let
          h = servicesHost;
          container = h.containers.swarm-victoriametrics;
          c = container.config;
          sock = "/run/swarm-victoriametrics-egress/egress.sock";
          relay = "swarm-victoriametrics-egress-relay";
          nginx = h.services.nginx.appendHttpConfig;
          otelDomain = h.services.hyperhive.swarm.otel.domain;
          autheliaDomain = h.services.hyperhive.swarm.authelia.domain;
        in
        c.systemd.sockets.${relay}.listenStreams == [ "127.0.0.1:4380" ]
        &&
          lib.hasSuffix "/lib/systemd/systemd-socket-proxyd ${sock}"
            c.systemd.services.${relay}.serviceConfig.ExecStart
        && !container.bindMounts."/run/swarm-victoriametrics-egress".isReadOnly
        && lib.hasInfix "listen unix:${sock};" nginx
        && lib.hasInfix "location / {\n    return 403;" nginx
        && lib.hasInfix "location ^~ /swarm-victoriametrics/ {\n    proxy_pass https://${otelDomain}" nginx
        && lib.hasInfix "proxy_ssl_name ${otelDomain};" nginx
        && lib.hasInfix "location = /api/oidc/token {\n    proxy_pass https://${autheliaDomain};" nginx
        && lib.hasInfix "proxy_ssl_name ${autheliaDomain};" nginx
        && lib.elem "-/run/swarm-victoriametrics-egress" h.systemd.services.nginx.serviceConfig.ReadWritePaths
        && lib.hasInfix "rm -f ${sock}" h.systemd.services.nginx.preStart;
    }
    {
      # Authelia's forwarder reads its secret from the store under the
      # publisher's identity, handed in for both units. The IdP host with no
      # identity is the control: the forwarder is off there, and on the
      # services host it is on, so the lookup is the right one.
      name = "the authelia container forwards its journal to its own route under the publisher's identity";
      ok =
        let
          h = servicesHost;
          container = h.containers.swarm-authelia;
          c = container.config;
          s = c.services.opentelemetry-collector.settings;
          otel = h.services.hyperhive.swarm.otel;
          clientId = h.services.hyperhive.swarm.authelia.otel.clientId;
          fetch = c.systemd.services.swarm-journal-forwarder-oidc;
          dir = forwarderFetchDir;
          registered = lib.findFirst (
            x: x.id == clientId
          ) null h.services.hyperhive.swarm.authelia.oidc.clients;
          reader = lib.head s.service.telemetry.metrics.readers;
          taken = [
            8888
            8889
            8890
            h.services.hyperhive.swarm.nats.otel.telemetryPort
          ]
          ++ map (o: o.port) hostNetnsForwarders;
        in
        c.services.opentelemetry-collector.enable
        && s.exporters.otlphttp.endpoint == "https://${otel.domain}/swarm-authelia"
        && s.extensions."oauth2client/swarm".client_id == clientId
        && s.extensions."oauth2client/swarm".endpoint_params.audience == clientId
        && otel.serviceProducers.swarm-authelia.clientId == clientId
        &&
          (lib.findFirst (a: a.key == "service.name") null s.processors.resource.attributes).value
          == "authelia"
        && fetch.environment.BAO_CLIENT_CERT == "${dir}/publisher-bao-cert"
        && fetch.environment.BAO_CLIENT_KEY == "${dir}/publisher-bao-key"
        && lib.elem "publisher-bao-cert" fetch.serviceConfig.LoadCredential
        && lib.elem "publisher-bao-key" fetch.serviceConfig.LoadCredential
        && lib.hasInfix "bao kv get -field=value secret/swarm/services/${clientId}/oidc/client" fetch.script
        && registered.kind == "machine"
        && registered.audience == [ clientId ]
        && !(lib.elem "--link-journal=host" container.extraFlags)
        && lib.any (lib.hasPrefix "--load-credential=publisher-bao-cert:") container.extraFlags
        && reader.pull.exporter.prometheus.host == "127.0.0.1"
        && !(lib.elem reader.pull.exporter.prometheus.port taken)
        && !autheliaColocated.containers.swarm-authelia.config.services.opentelemetry-collector.enable
        && !(
          autheliaColocated.containers.swarm-authelia.config.systemd.services ? swarm-journal-forwarder-oidc
        );
    }
  ]
  ++ forwarderCases
  ++ hostNetnsForwarderCases;
in
runGroup "nats-authelia" cases