feat(#3297): scope a hive's queue grant to its own subjects

Every admitted client got the same unrestricted grant, so any hive could
write any other hive's status key. The responder now derives a
permission set from the caller's identity and mints it into the user
JWT.

A hive may publish to its own KV key and the two JetStream subjects
needed to reach it; the controller may list and fetch every key and
write none; anything else is denied outright. Deny is the default
because every other shape fails open, and silently: a client that
matched no rule and kept the old grant would make the policy advisory.

The subject sets are measured rather than reasoned about, and two of
them are counter-intuitive. `$KV.<bucket>.<key>` alone does not let a
client write that key, because the client resolves the bucket first. And
`$JS.API.>` is not "the JetStream permission": it also covers
`$JS.API.STREAM.DELETE`, with which a hive correctly refused on a
neighbour's key can delete the whole bucket and every hive's data with
it. Granting it would have made per-key scoping decorative, so the
subjects are named individually and a test asserts the wildcard does not
come back as a convenience.

Minimality is by removal: each subject was dropped in turn to confirm
the client breaks without it. That is not pedantry — an additive search
had called a set minimal while two of its five subjects were never
needed, which ships an unnecessary grant with a measurement attached
making it look earned.

Both grants include `STREAM.CREATE` on the one named stream, because
`status::open_or_create` is called by both ends: either may arrive first
on a fresh swarm, and without it a new swarm never gets a bucket at all.
`CREATE` is not `UPDATE`, so a second arrival cannot reshape the bucket
the first one made.

`status::BUCKET` moves out from behind the `kv` feature so this
responder can share it. The name is a `&str` with no dependencies and
only `open_or_create` needs JetStream; gating the name forced a third
consumer to choose between a stack it does not use and a copied literal,
and the copied literal is exactly the disagreement that module exists to
prevent.

Only publish is scoped. Subscription permissions are unrestricted and
unmeasured, and the module docs say so rather than implying a property
nothing established.

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,17 @@ 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.
+    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 +191,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 +222,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");