fix(#3384): source the queue policy's principals from the modules that mint them

The auth-callout responder decides what an admitted client may publish
from two strings: the prefix marking a hive client, and the client id
allowed to read every hive's key. Both were literals in three places --
swarm-authelia.nix mints "hive-${name}", swarm-controller.nix defines
"swarm-controller", and the responder carried its own copies as clap
defaults because swarm-nats.nix passed neither.

Each producer now publishes its value as a readOnly option and the
responder's ExecStart reads them, so the agreement is one evaluation
rather than three strings that happen to be equal. Same pattern the
module already uses for `--account`, and the same argument
swarm-authelia.nix gives for publishing `machine` and `unit`.

Worth the change because the failure is silent and misattributed:
rename either principal and the responder starts denying the one that
stopped matching, a denial reaches a NATS client as a timeout rather
than an error, and a hive that is refused looks exactly like a hive
that has not reported yet.

Cargo.lock

@@ -4619,6 +4619,7 @@ dependencies = [
  "serde",
  "serde_json",
  "sha2 0.11.0",
+ "swarm-queue-client",
  "tokio",
  "tracing",
  "tracing-subscriber",

nix/host-modules/swarm-authelia.nix

@@ -91,7 +91,7 @@ let
   # declared entries and not on these, which is an eval error reachable
   # only once hive identities are on.
   hiveClients = lib.mapAttrsToList (name: _: {
-    id = "hive-${name}";
+    id = "${cfg.hiveClientPrefix}${name}";
     description = "HyperHive hive ${name}";
     kind = "machine";
     redirectUris = [ ];
@@ -508,6 +508,26 @@ in
     # the call site: the machine and unit names are derived from
     # `instance` here, so a second copy elsewhere is a second thing to
     # keep in step, and the one that drifts is the one nobody tests.
+    hiveClientPrefix = lib.mkOption {
+      type = lib.types.str;
+      readOnly = true;
+      default = "hive-";
+      description = ''
+        Prefix of the OAuth2 client id minted for each hive in
+        `services.hyperhive.swarm.hives` — the client for hive `alpha` is
+        `${config.services.hyperhive.swarm.authelia.hiveClientPrefix}alpha`.
+        Read-only for the same reason as `machine` and `unit`: it is what
+        this module produces, published so a consumer does not carry a
+        second copy.
+
+        The consumer that matters is the queue's auth-callout responder,
+        which decides *which hive* a connection is by stripping this
+        prefix off the introspected client id. Split the two spellings and
+        every hive is denied — as a timeout, indistinguishable from a hive
+        that simply has not reported.
+      '';
+    };
+
     machine = lib.mkOption {
       type = lib.types.str;
       readOnly = true;

nix/host-modules/swarm-controller.nix

@@ -41,7 +41,7 @@ let
   # belongs to the responder. One identity per principal — the rule is that
   # a principal's credentials all derive from the same identity, not that
   # the swarm has one.
-  queueClientId = "swarm-controller";
+  queueClientId = cfg.queueClientId;
 
   # `LoadCredential` and not a copy-oneshot, which is where this deliberately
   # differs from the callout responder: that one delivers INTO a container,
@@ -114,6 +114,25 @@ let
 in
 {
   options.services.hyperhive.swarm.controller = {
+    queueClientId = lib.mkOption {
+      type = lib.types.str;
+      readOnly = true;
+      default = "swarm-controller";
+      description = ''
+        The OAuth2 client id the controller presents to the swarm queue.
+        Read-only: it is what this module registers, published so the
+        auth-callout responder can be told which client may read every
+        hive's key without repeating the string.
+
+        The responder decides that from a client id, and a client id it
+        does not recognise is **denied**. A denial reaches a NATS client
+        as a timeout rather than an error, and a controller that cannot
+        read looks exactly like a swarm where no hive has reported yet —
+        so a drift between these two spellings is invisible at the point
+        it is introduced and misattributed everywhere it shows up.
+      '';
+    };
+
     enable = lib.mkOption {
       type = lib.types.bool;
       default = false;

nix/host-modules/swarm-nats.nix

@@ -9,6 +9,10 @@ let
   autheliaCfg = config.services.hyperhive.swarm.authelia;
   autheliaUrl = autheliaCfg.url;
   networkCfg = config.services.hyperhive.network;
+  # Read even when the controller runs on a different host: what is needed
+  # is the client id that module *declares*, which is the same string
+  # everywhere, not whether the daemon happens to be enabled here.
+  controllerCfg = config.services.hyperhive.swarm.controller;
 
   # The account the callout responder authenticates as, and the account
   # authorized clients are placed in. Two accounts rather than one: an
@@ -531,6 +535,15 @@ in
                 # be the same string — which is why both come from one let.
                 "--account ${lib.escapeShellArg clientAccount}"
                 "--introspection-url ${lib.escapeShellArg introspectionUrl}"
+                # Both of these name a principal some OTHER module mints,
+                # so both are read out of that module rather than spelled
+                # again here — same argument as `--account` above, one
+                # level wider. The responder denies a client id it does
+                # not recognise, and a NATS denial arrives as a timeout,
+                # so a drift here is silent at the point of change and
+                # misattributed at the point of failure.
+                "--hive-client-prefix ${lib.escapeShellArg autheliaCfg.hiveClientPrefix}"
+                "--reader-client ${lib.escapeShellArg controllerCfg.queueClientId}"
               ];
               # Every credential arrives by `LoadCredential` and is named
               # on the command line only as a **path** — `argv` is

swarm-nats-auth/Cargo.toml

@@ -22,6 +22,11 @@ serde.workspace = true
 serde_json.workspace = true
 # The jti digest: base32hex(sha256(claims)) over every JWT this crate signs.
 sha2.workspace = true
+# For `status::BUCKET` alone - the subjects a hive may publish to are derived
+# from the bucket name, and the reader, the writer and this responder must
+# name the same one. Deliberately WITHOUT the `kv` feature: this crate derives
+# subject strings, it never opens the bucket.
+swarm-queue-client.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true

swarm-nats-auth/src/main.rs

@@ -28,6 +28,7 @@ use clap::Parser;
 use futures_util::StreamExt;
 
 mod introspect;
+mod policy;
 mod request;
 mod respond;
 
@@ -73,6 +74,28 @@ struct Args {
     /// Path to this responder's own OIDC client secret.
     #[arg(long)]
     client_secret_file: PathBuf,
+
+    /// Client-id prefix that marks a hive. `swarm-authelia.nix` mints one
+    /// machine client per roster entry as `hive-<name>`, while the KV key is
+    /// the bare `<name>` — this is the contract between the two, declared
+    /// rather than inferred from the shape of an id.
+    #[arg(long, default_value = "hive-")]
+    hive_client_prefix: String,
+
+    /// Client ids allowed to read every hive's status. Repeatable. The
+    /// default is the swarm controller, which is the only reader that exists.
+    #[arg(long = "reader-client", default_values_t = [String::from("swarm-controller")])]
+    reader_clients: Vec<String>,
+
+    /// Additional subjects a hive may publish to, with `{hive}` standing for
+    /// its own name. Repeatable, empty by default.
+    ///
+    /// The extension point for a second stream published by the same
+    /// `hive-<name>` identity — lifecycle notices, say. Without it, adding one
+    /// means changing this responder; with it, a deployment says so and the
+    /// subject still lands inside that hive's own namespace.
+    #[arg(long = "hive-publish-subject")]
+    hive_publish_subjects: Vec<String>,
 }
 
 /// Read a secret file and strip surrounding whitespace.
@@ -102,6 +125,21 @@ async fn main() -> anyhow::Result<()> {
     let args = Args::parse();
 
     let client_secret = read_secret(&args.client_secret_file)?;
+    // The bucket is NOT a flag. `swarm_queue_client::status`'s own docs say
+    // why: reader and writer must name the same bucket, and an option is a
+    // way for two deployments to disagree about which one that is. This
+    // responder is the third end that names it, so it takes the same
+    // constant rather than a copy of the literal.
+    // Fails the process rather than warning: a policy that cannot express a
+    // per-hive namespace is not a policy this responder should run with, and
+    // the queue's fail-closed state (no responder) is a legible outage where a
+    // silently over-broad grant is not.
+    let policy = policy::Policy::new(
+        args.hive_client_prefix.clone(),
+        swarm_queue_client::status::BUCKET.to_owned(),
+        args.reader_clients.clone(),
+        args.hive_publish_subjects.clone(),
+    )?;
     let http = reqwest::Client::new();
     let issuer = nkeys::KeyPair::from_seed(&read_secret(&args.issuer_seed_file)?)
         .context("parse the account signing seed")?;
@@ -157,12 +195,25 @@ async fn main() -> anyhow::Result<()> {
             }),
             None => None,
         };
+        // Admission said who; the policy says what. A caller the `IdP`
+        // vouches for but no rule matches is denied — see `policy`'s module
+        // docs for why that is deny and not "connect with nothing".
+        let permissions = caller.as_deref().and_then(|id| policy.permissions(id));
+        if let (Some(id), None) = (caller.as_deref(), permissions.as_ref()) {
+            // Loud, and the one case an operator has to be able to find: a
+            // valid credential refused by our own policy. The alternative is
+            // a client that authenticates fine and mysteriously cannot work.
+            tracing::warn!(
+                caller = %id,
+                "authenticated client matches no policy rule; denying"
+            );
+        }
         // The client id is an identifier, not a credential, and it is the
         // only thing tying a connection in this log to a hive.
         tracing::info!(
             user_nkey = %req.user_nkey,
             server_id = %req.server_id.id,
-            granted = caller.is_some(),
+            granted = permissions.is_some(),
             caller = caller.as_deref().unwrap_or("-"),
             "auth request"
         );
@@ -175,14 +226,15 @@ async fn main() -> anyhow::Result<()> {
             tracing::warn!("auth request had no reply subject; dropping");
             continue;
         };
-        // The grant is still unscoped: knowing *who* connected is what makes
-        // scoping possible, not what performs it. Narrowing the permissions
-        // to the caller's own subjects is the next slice, and lands in
-        // `respond::grant` where the JWT is minted.
-        let token = if caller.is_some() {
-            respond::grant(&issuer, &args.account, &req.server_id.id, &req.user_nkey)
-        } else {
-            respond::deny(&issuer, &req.server_id.id, &req.user_nkey)
+        let token = match &permissions {
+            Some(permissions) => respond::grant(
+                &issuer,
+                &args.account,
+                &req.server_id.id,
+                &req.user_nkey,
+                permissions,
+            ),
+            None => respond::deny(&issuer, &req.server_id.id, &req.user_nkey),
         };
         if let Err(e) = client.publish(reply_to, token.into()).await {
             tracing::warn!(error = ?e, "failed to publish auth response");

swarm-nats-auth/src/policy.rs

@@ -0,0 +1,401 @@
+//! What an admitted client is allowed to do.
+//!
+//! [`crate::introspect`] answers *whether* to admit and *as whom*. This
+//! answers *what that identity may touch* — the two are deliberately separate
+//! decisions: admission is the `IdP`'s, authorisation is the swarm's.
+//!
+//! # Deny is the default, and that is a choice with a measurement behind it
+//!
+//! A client whose id matches no rule gets **no grant at all**, not an
+//! unrestricted one. Every other shape here fails open: an unmatched client
+//! that kept today's unscoped grant would make the whole policy advisory, and
+//! the failure would be silent. Denying costs a loud refusal the first time a
+//! new consumer appears, which is a config line to fix.
+//!
+//! Each subject set below is *minimal by removal*, and carries its own note
+//! saying what breaks without it — the reasoning lives next to the list it
+//! constrains rather than in one block here, because that is where a reader
+//! about to edit the list will meet it.
+
+/// The subjects an admitted client may publish to.
+///
+/// **Publish only — subscription is left unrestricted.** The queue's
+/// confidentiality boundary is the account, and a hive reading another hive's
+/// *published* status is not the problem this solves; writing it is. Scoping
+/// `sub` is a separate change with its own measurement, and claiming it here
+/// without one would repeat the `$JS.API.>` mistake described on
+/// `Policy::hive_subjects`.
+#[derive(Debug, PartialEq, Eq)]
+pub struct Permissions {
+    /// Subjects allowed for publish. Never empty: an empty allow-list is a
+    /// grant that can do nothing, which is a denial wearing a grant's shape.
+    pub publish: Vec<String>,
+}
+
+/// Which client ids get which permissions.
+///
+/// Constructed from configuration so that adding a subject a hive may publish
+/// — a second event stream, say — is a deployment change rather than a change
+/// to this responder.
+#[derive(Debug, Clone)]
+pub struct Policy {
+    hive_prefix: String,
+    bucket: String,
+    readers: Vec<String>,
+    extra_hive_subjects: Vec<String>,
+}
+
+/// Placeholder replaced with the hive's own name in `extra_hive_subjects`.
+const HIVE_PLACEHOLDER: &str = "{hive}";
+
+impl Policy {
+    /// `hive_prefix` is the client-id prefix that marks a hive, `bucket` the
+    /// KV bucket hives report status in, `readers` the client ids allowed to
+    /// read every hive's key, and `extra_hive_subjects` additional subjects a
+    /// hive may publish to (with `{hive}` standing for its own name).
+    ///
+    /// # Errors
+    ///
+    /// An extra subject with no `{hive}` in it is refused. Such a template
+    /// expands to the **same** subject for every hive, which is not a per-hive
+    /// namespace — it is the absence of one, arrived at through the option
+    /// whose only purpose is to provide one. Returning a `Result` rather than
+    /// checking at the call site is deliberate: it makes an unscoped policy
+    /// unconstructible instead of merely unlikely.
+    pub fn new(
+        hive_prefix: String,
+        bucket: String,
+        readers: Vec<String>,
+        extra_hive_subjects: Vec<String>,
+    ) -> anyhow::Result<Self> {
+        if let Some(bad) = extra_hive_subjects
+            .iter()
+            .find(|s| !s.contains(HIVE_PLACEHOLDER))
+        {
+            anyhow::bail!(
+                "--hive-publish-subject {bad:?} contains no {HIVE_PLACEHOLDER}: every hive \
+                 would be granted that exact subject, so it is a shared channel rather than \
+                 a per-hive namespace"
+            );
+        }
+        Ok(Self {
+            hive_prefix,
+            bucket,
+            readers,
+            extra_hive_subjects,
+        })
+    }
+
+    /// The permissions for `client_id`, or `None` when no rule matches.
+    ///
+    /// `None` is a denial. It is not "grant nothing and let them connect":
+    /// a connected client with no permissions still holds a slot and still
+    /// looks admitted in the logs, which is a worse answer than a refusal.
+    pub fn permissions(&self, client_id: &str) -> Option<Permissions> {
+        if let Some(hive) = self.hive_name(client_id) {
+            return Some(Permissions {
+                publish: self.hive_subjects(hive),
+            });
+        }
+        if self.readers.iter().any(|r| r == client_id) {
+            return Some(Permissions {
+                publish: self.reader_subjects(),
+            });
+        }
+        None
+    }
+
+    /// The hive a client id names, when it names one.
+    ///
+    /// The prefix is configuration, not a pattern this module guesses at: the
+    /// authelia client is `hive-<name>` while the KV key is the bare `<name>`,
+    /// and stripping by eye is how a client id that merely *looks*
+    /// hive-shaped ends up with a hive's permissions. An id that is the prefix
+    /// and nothing else names no hive and is refused — `$KV.<bucket>.` is not
+    /// a narrower subject than `$KV.<bucket>.<name>`, it is a different one.
+    ///
+    /// # This matches the prefix; it does not verify the roster
+    ///
+    /// A client id is a hive here because it *starts with the prefix*, not
+    /// because it appears in `services.hyperhive.swarm.hives`. This responder
+    /// runs inside a container and has no view of the roster, so an
+    /// operator-declared client called `hive-anything` would be granted
+    /// `$KV.<bucket>.anything` — a key no real hive owns, which the controller
+    /// renders as a hive it does not recognise.
+    ///
+    /// Passing the roster in would close that, and would also be a **second
+    /// place deciding who may connect as what**, which `crate::introspect`'s
+    /// docs argue against for the same reason admission lives in one place.
+    /// The prefix is a contract with `swarm-authelia.nix`, and this is the end
+    /// of it that can be checked from in here.
+    fn hive_name<'a>(&self, client_id: &'a str) -> Option<&'a str> {
+        client_id
+            .strip_prefix(&self.hive_prefix)
+            .filter(|name| !name.is_empty())
+    }
+
+    fn stream(&self) -> String {
+        format!("KV_{}", self.bucket)
+    }
+
+    /// What *any* `JetStream` client must be able to ask before it can do
+    /// anything at all, bucket-specific or not.
+    ///
+    /// Both were measured from the server's own refusals, not reasoned about:
+    /// a grant carrying every bucket-specific subject and neither of these
+    /// cannot even create the bucket — the client times out on `$JS.API.INFO`
+    /// long before it reaches a subject that was granted.
+    ///
+    /// - `$JS.API.INFO` — account-level `JetStream` info, requested on connect.
+    /// - `$JS.API.STREAM.NAMES` — how a client finds the stream backing a
+    ///   bucket. It lets a client enumerate stream names in the account, which
+    ///   in an account holding one bucket discloses a name both ends already
+    ///   share.
+    ///
+    /// 🩸 Earlier measurements missed both, because they either granted
+    /// `$JS.API.>` wholesale or ran against a bucket the *setup* had already
+    /// created while unscoped. A minimum established against an existing
+    /// bucket is not the minimum for making one.
+    fn jetstream_minimum() -> [String; 2] {
+        ["$JS.API.INFO".to_owned(), "$JS.API.STREAM.NAMES".to_owned()]
+    }
+
+    /// Creating the bucket, which **both** ends need.
+    ///
+    /// `swarm_queue_client::status::open_or_create` is called by the hive that
+    /// writes and the controller that reads, because either may arrive first
+    /// on a fresh swarm — a bucket that must pre-exist turns "deployed in the
+    /// wrong order" into a permanent, silent absence of data. Whichever
+    /// connects first therefore has to be able to create it.
+    ///
+    /// Narrower than it looks: this is `CREATE` on one named stream, not
+    /// `UPDATE` and not the `$JS.API.>` wildcard. Measured rather than
+    /// assumed — a hive holding this grant and running `stream edit` leaves
+    /// the stream's config untouched, and no `$JS.API.STREAM.UPDATE` is ever
+    /// published. Worth stating because the failure it would hide is quiet: a
+    /// hive able to reshape the shared bucket could set `MaxMsgs: 1` and evict
+    /// every other hive's status without ever touching `STREAM.DELETE`, and
+    /// per-key scoping would still look intact.
+    fn create(&self) -> String {
+        format!("$JS.API.STREAM.CREATE.{}", self.stream())
+    }
+
+    /// What one hive may publish: the account minimum, the bucket lookup,
+    /// creation, and its **own** key.
+    ///
+    /// Two things a reader would reasonably assume, both false and both
+    /// measured (`state/attack-3297-js-api-door.sh` in atlas's notes):
+    ///
+    /// - `$KV.<bucket>.<key>` alone does **not** let a client write that key.
+    ///   The client resolves the bucket first, so the `STREAM.INFO` subject is
+    ///   part of the minimum for a plain write.
+    /// - `$JS.API.>` is not "the `JetStream` permission". It also covers
+    ///   `$JS.API.STREAM.DELETE`, with which a hive correctly refused on a
+    ///   neighbour's individual key can destroy the whole bucket — every
+    ///   hive's data. Granting it would make per-key scoping decorative, which
+    ///   is why these are named one at a time and a test asserts the wildcard
+    ///   never returns as a convenience.
+    fn hive_subjects(&self, hive: &str) -> Vec<String> {
+        let mut subjects = Self::jetstream_minimum().to_vec();
+        subjects.extend([
+            format!("$JS.API.STREAM.INFO.{}", self.stream()),
+            self.create(),
+            format!("$KV.{}.{hive}", self.bucket),
+        ]);
+        subjects.extend(
+            self.extra_hive_subjects
+                .iter()
+                .map(|s| s.replace(HIVE_PLACEHOLDER, hive)),
+        );
+        subjects
+    }
+
+    fn reader_subjects(&self) -> Vec<String> {
+        let stream = self.stream();
+        let mut subjects = Self::jetstream_minimum().to_vec();
+        subjects.extend([
+            format!("$JS.API.STREAM.INFO.{stream}"),
+            self.create(),
+            // The `.>` form specifically: the bare `$JS.API.DIRECT.GET.<stream>`
+            // is not the subject the client uses, and granting it was measured
+            // to make no difference.
+            format!("$JS.API.DIRECT.GET.{stream}.>"),
+            // `store.keys()` — the controller lists before it fetches, so a
+            // reader without this can get a key it already knows and discover
+            // nothing.
+            format!("$JS.API.CONSUMER.CREATE.{stream}.>"),
+        ]);
+        subjects
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn policy() -> Policy {
+        Policy::new(
+            "hive-".to_owned(),
+            "hive-status".to_owned(),
+            vec!["swarm-controller".to_owned()],
+            Vec::new(),
+        )
+        .expect("the default policy is valid")
+    }
+
+    #[test]
+    fn an_extra_subject_without_the_placeholder_is_refused() {
+        // 🩸 The option exists to put a second stream inside ONE hive's
+        // namespace. A template with no `{hive}` expands to the same subject
+        // for all of them, so the flag whose purpose is scoping becomes the
+        // way to remove it - silently, and only in the deployment that set it.
+        let err = Policy::new(
+            "hive-".to_owned(),
+            "hive-status".to_owned(),
+            Vec::new(),
+            vec!["$SWARM.events.all".to_owned()],
+        )
+        .expect_err("a subject shared by every hive must not be accepted");
+        let msg = format!("{err}");
+        assert!(
+            msg.contains("$SWARM.events.all"),
+            "the error must name the offending template, got: {msg}"
+        );
+    }
+
+    #[test]
+    fn an_extra_subject_with_the_placeholder_is_accepted() {
+        // The other half: a check that only ever rejects would be indistinguishable
+        // from the option being broken.
+        Policy::new(
+            "hive-".to_owned(),
+            "hive-status".to_owned(),
+            Vec::new(),
+            vec!["$SWARM.events.{hive}.>".to_owned()],
+        )
+        .expect("a per-hive template is the shape this option is for");
+    }
+
+    #[test]
+    fn a_hive_may_publish_its_own_key_and_nothing_elses() {
+        let p = policy()
+            .permissions("hive-alpha")
+            .expect("a hive is admitted");
+        assert!(p.publish.contains(&"$KV.hive-status.alpha".to_owned()));
+        assert!(!p.publish.iter().any(|s| s.contains("beta")));
+    }
+
+    #[test]
+    fn a_hive_grant_never_includes_the_jetstream_wildcard() {
+        // `$JS.API.>` also covers `$JS.API.STREAM.DELETE.KV_hive-status`, with
+        // which a hive refused on a neighbour's key can delete the entire
+        // bucket. Measured, not theorised - this test is the guard against it
+        // coming back as a convenience.
+        let p = policy()
+            .permissions("hive-alpha")
+            .expect("a hive is admitted");
+        assert!(!p.publish.iter().any(|s| s.contains("$JS.API.>")));
+        assert!(!p.publish.iter().any(|s| s.contains("STREAM.DELETE")));
+        assert!(!p.publish.iter().any(|s| s.contains("STREAM.PURGE")));
+    }
+
+    #[test]
+    fn the_bucket_lookup_is_part_of_a_publishers_minimum() {
+        // Without it the put is refused: the client resolves the bucket before
+        // it writes. Dropping this line looks like tightening and is breaking.
+        let p = policy()
+            .permissions("hive-alpha")
+            .expect("a hive is admitted");
+        assert!(
+            p.publish
+                .contains(&"$JS.API.STREAM.INFO.KV_hive-status".to_owned())
+        );
+    }
+
+    #[test]
+    fn every_grant_carries_the_jetstream_minimum() {
+        // 🩸 Found by the shipping gate, not by any unit test: a grant with
+        // every bucket-specific subject and neither of these cannot create the
+        // bucket at all. The client times out on `$JS.API.INFO` before it
+        // reaches anything that was granted, and a NATS denial looks like a
+        // hang from the client side — the server log is what named them.
+        for client in ["hive-alpha", "swarm-controller"] {
+            let p = policy().permissions(client).expect("admitted");
+            for required in ["$JS.API.INFO", "$JS.API.STREAM.NAMES"] {
+                assert!(
+                    p.publish.iter().any(|s| s == required),
+                    "{client} is missing {required}, so it cannot use JetStream at all"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn both_ends_may_create_the_bucket_but_not_reshape_it() {
+        // 🩸 The bug the subject measurements could not see: they ran against
+        // a bucket the *setup* had already created while unscoped, so
+        // "minimal" meant minimal-given-a-bucket-that-exists. `open_or_create`
+        // is called by both ends by design — whichever arrives first on a
+        // fresh swarm makes the bucket — so without CREATE a new swarm never
+        // gets one, and every other test still passes.
+        for client in ["hive-alpha", "swarm-controller"] {
+            let p = policy().permissions(client).expect("admitted");
+            assert!(
+                p.publish
+                    .contains(&"$JS.API.STREAM.CREATE.KV_hive-status".to_owned()),
+                "{client} cannot create the bucket on a fresh swarm"
+            );
+            // CREATE is not UPDATE: a second arrival must not be able to
+            // reshape the bucket the first one made.
+            assert!(!p.publish.iter().any(|s| s.contains("STREAM.UPDATE")));
+        }
+    }
+
+    #[test]
+    fn a_reader_may_list_and_fetch_but_not_write() {
+        let p = policy()
+            .permissions("swarm-controller")
+            .expect("the reader is admitted");
+        assert!(
+            p.publish
+                .contains(&"$JS.API.CONSUMER.CREATE.KV_hive-status.>".to_owned())
+        );
+        assert!(
+            p.publish
+                .contains(&"$JS.API.DIRECT.GET.KV_hive-status.>".to_owned())
+        );
+        assert!(!p.publish.iter().any(|s| s.starts_with("$KV.")));
+    }
+
+    #[test]
+    fn an_unmatched_client_is_denied_not_granted_everything() {
+        // The whole policy is advisory if this returns a grant.
+        assert_eq!(policy().permissions("some-other-service"), None);
+        assert_eq!(policy().permissions(""), None);
+    }
+
+    #[test]
+    fn the_prefix_alone_names_no_hive() {
+        // `hive-` would otherwise produce `$KV.hive-status.`, a subject nobody
+        // reviewed and that no hive owns.
+        assert_eq!(policy().permissions("hive-"), None);
+    }
+
+    #[test]
+    fn extra_subjects_are_scoped_to_the_hive_that_publishes_them() {
+        // The extension point: a second stream (lifecycle notices, say) is
+        // published by the same `hive-<name>` identity on a different subject.
+        // It has to land inside that hive's namespace, or one hive can write
+        // another's events even though its status key is scoped.
+        let p = Policy::new(
+            "hive-".to_owned(),
+            "hive-status".to_owned(),
+            Vec::new(),
+            vec!["$SWARM.events.{hive}.>".to_owned()],
+        )
+        .expect("a per-hive template is valid");
+        let g = p.permissions("hive-alpha").expect("a hive is admitted");
+        assert!(g.publish.contains(&"$SWARM.events.alpha.>".to_owned()));
+        assert!(!g.publish.iter().any(|s| s.contains("{hive}")));
+    }
+}

swarm-nats-auth/src/respond.rs

@@ -149,6 +149,14 @@ fn now_secs() -> i64 {
 struct UserNats {
     #[serde(rename = "type")]
     kind: &'static str,
+    /// Subjects this client may publish to.
+    ///
+    /// The server enforces what is minted here — measured, because a
+    /// non-operator server validates minted claims against its own config
+    /// rather than trusting them, and this responder's own history includes a
+    /// field the server rejected while the responder said `granted=true`.
+    #[serde(rename = "pub")]
+    publish: Permission,
     subs: i64,
     data: i64,
     payload: i64,
@@ -156,12 +164,29 @@ struct UserNats {
     version: i64,
 }
 
+/// A NATS permission block.
+///
+/// Only `allow` is modelled. NATS also takes `deny`, and a struct that has it
+/// is a struct someone will use: an allow-list plus a deny-list has two places
+/// deciding the same question, and the interaction between them is a thing to
+/// remember rather than to read.
+#[derive(Serialize)]
+struct Permission {
+    allow: Vec<String>,
+}
+
 /// Mint the user JWT an admitted client presents.
 ///
 /// `account` is the account **name** from the server's `accounts` block (the
 /// module's `clientAccount`), not a public key — in config mode the server
 /// resolves `aud` against its own config rather than against a key.
-fn user_jwt(now: i64, issuer: &KeyPair, account: &str, user_nkey: &str) -> String {
+fn user_jwt(
+    now: i64,
+    issuer: &KeyPair,
+    account: &str,
+    user_nkey: &str,
+    publish: Vec<String>,
+) -> String {
     sign(
         Claims {
             iat: now,
@@ -172,6 +197,7 @@ fn user_jwt(now: i64, issuer: &KeyPair, account: &str, user_nkey: &str) -> Strin
             aud: Some(account.to_owned()),
             nats: UserNats {
                 kind: "user",
+                publish: Permission { allow: publish },
                 subs: -1,
                 data: -1,
                 payload: -1,
@@ -183,10 +209,21 @@ fn user_jwt(now: i64, issuer: &KeyPair, account: &str, user_nkey: &str) -> Strin
     )
 }
 
-/// Grant: mint a user JWT placing the client in `account` and wrap it.
-pub fn grant(issuer: &KeyPair, account: &str, server_id: &str, user_nkey: &str) -> String {
+/// Grant: mint a user JWT placing the client in `account`, scoped to
+/// `permissions`, and wrap it.
+///
+/// Taking the permissions by value rather than defaulting them is the point:
+/// there is no way to call this and get an unscoped grant by omission, so a
+/// future caller cannot forget the argument that makes the scoping real.
+pub fn grant(
+    issuer: &KeyPair,
+    account: &str,
+    server_id: &str,
+    user_nkey: &str,
+    permissions: &crate::policy::Permissions,
+) -> String {
     let now = now_secs();
-    let jwt = user_jwt(now, issuer, account, user_nkey);
+    let jwt = user_jwt(now, issuer, account, user_nkey, permissions.publish.clone());
     response(now, issuer, server_id, user_nkey, Ok(jwt))
 }
 
@@ -304,10 +341,38 @@ mod tests {
         assert_eq!(claims["aud"], "NSERVER");
     }
 
+    /// The permissions a test grant carries. Any non-empty set will do for
+    /// the wrapper-shape assertions; `policy.rs` owns what the real ones are.
+    fn perms() -> crate::policy::Permissions {
+        crate::policy::Permissions {
+            publish: vec!["$KV.hive-status.alpha".to_owned()],
+        }
+    }
+
+    #[test]
+    fn the_issued_user_jwt_carries_the_permissions_it_was_given() {
+        // The scoping is only real if it survives into the minted token. A
+        // policy that computes the right subjects and a grant that drops them
+        // look identical from every test that stops at the policy.
+        let account = KeyPair::new_account();
+        let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &perms()));
+        let user = decode_claims(wrapper["nats"]["jwt"].as_str().expect("a user jwt"));
+        assert_eq!(user["nats"]["pub"]["allow"][0], "$KV.hive-status.alpha");
+        // An empty or absent allow-list is NATS' "everything": the one shape
+        // that turns this whole change into a no-op while every other
+        // assertion still passes.
+        assert!(
+            user["nats"]["pub"]["allow"]
+                .as_array()
+                .is_some_and(|a| !a.is_empty()),
+            "an empty pub.allow is an unscoped grant"
+        );
+    }
+
     #[test]
     fn a_grant_carries_a_user_jwt_and_no_error() {
         let account = KeyPair::new_account();
-        let claims = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT"));
+        let claims = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &perms()));
         assert!(claims["nats"]["error"].is_null());
         assert!(
             claims["nats"]["jwt"]
@@ -331,7 +396,7 @@ mod tests {
     #[test]
     fn the_issued_user_jwt_names_its_account_by_aud_and_sets_no_issuer_account() {
         let account = KeyPair::new_account();
-        let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT"));
+        let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &perms()));
         let user = decode_claims(wrapper["nats"]["jwt"].as_str().expect("a user jwt"));
 
         // In server-config mode the account is named by `aud`...

swarm-queue-client/src/lib.rs

@@ -144,9 +144,16 @@ pub fn chain(error: &dyn std::error::Error) -> String {
 }
 
 /// The hive-status KV bucket, shared by the hive that writes it and the
-/// controller that reads it. Behind the `kv` feature — see the module doc
-/// for why a bucket name and its config belong to neither end alone.
-#[cfg(feature = "kv")]
+/// controller that reads it. See the module doc for why a bucket name and
+/// its config belong to neither end alone.
+///
+/// The module itself is unconditional; only the parts that *open* the bucket
+/// need the `kv` feature. The name is a `&str` with no dependencies, and a
+/// third end names it too — the auth-callout responder, which derives the
+/// subjects a hive may publish to from it without ever speaking `jetstream`.
+/// Gating the name behind `kv` would have forced that consumer to choose
+/// between pulling a JetStream stack it does not use and copying the literal,
+/// which is the disagreement this module exists to prevent.
 pub mod status;
 
 /// Only the fields this needs; authelia returns several.

swarm-queue-client/src/status.rs

@@ -7,7 +7,7 @@
 //! repeated across crates is an agreement nothing checks.**
 //!
 //! The name is the obvious half. The sharper half is the *config*: both
-//! ends open the bucket with [`crate::status::open_or_create`], because either may
+//! ends open the bucket with `open_or_create`, because either may
 //! arrive first on a fresh swarm and neither can assume the other has
 //! run. If the two ends passed different `Config`s, whichever created it
 //! would win and the other's `get_key_value` would succeed against a
@@ -15,10 +15,17 @@
 //! nobody chose. Sharing the constructor makes the race have one outcome
 //! instead of two.
 //!
-//! Feature-gated (`kv`) so the crate's other consumer, the auth-callout
-//! responder, still pulls neither `jetstream` nor `kv`: it speaks the
-//! connect and nothing else.
+//! The bucket *name* is unconditional; only `open_or_create` is behind the
+//! `kv` feature. (Named in backticks rather than linked: with `kv` off the item
+//! does not exist, and an intra-doc link to it fails the rustdoc gate in
+//! exactly the configuration this split exists to support.) A third end names the bucket without ever opening it — the
+//! auth-callout responder, which derives the subjects a hive may publish to
+//! from it — and it speaks neither `jetstream` nor `kv`. Gating the name too
+//! would have made that consumer choose between a JetStream stack it does not
+//! use and a copied literal, and a copied literal is precisely the agreement
+//! nothing checks.
 
+#[cfg(feature = "kv")]
 use crate::Error;
 
 /// The KV bucket hives publish their status snapshots into, one key per
@@ -39,6 +46,7 @@ pub const BUCKET: &str = "hive-status";
 /// controller and the hives come up in no particular order, and a bucket
 /// that must pre-exist turns "the swarm was deployed in the wrong order"
 /// into a permanent, silent absence of data.
+#[cfg(feature = "kv")]
 pub async fn open_or_create(
     client: &async_nats::Client,
 ) -> Result<async_nats::jetstream::kv::Store, Error> {