swarm-nats-auth: limit hive and swarm-controller subscribe to the subjects they use

A hive's queue credential and the swarm controller's could subscribe to
every subject in the swarm account, so a hive could read every other
hive's deploy requests and every agent's terminal, state, input and
request replies.

Their grants now carry a subscribe allow-list:

- a hive `hive-<h>`: `$SWARM.knowledge`, `$SWARM.deploy.<h>` and its
  reply inbox `_INBOX.client.hive-<h>.>`
- the reader (swarm-controller): `$SWARM.term.*`, `$SWARM.term.*.sub.*`,
  `$SWARM.agent-state.*` and its reply inbox
  `_INBOX.client.<client-id>.>`

`swarm_queue_client::connect`, which hive-c0re and swarm-controller both
use, now connects with `_INBOX.client.<client-id>` as its inbox prefix,
the same shape agents got for their own credential. In async-nats 0.50
every request reply and every ordered consumer (the hive's wanted-state
watch, the controller's KV key listings) delivers under that prefix, so
the one inbox grant covers them all.

A hive or controller still on a build without the prefix has every
JetStream reply refused once the responder runs this policy, so the
clients need to be rebuilt before, or with, swarm-nats-auth.

Closes #4945

docs/swarm/README.md

@@ -256,8 +256,17 @@ and nothing else. The harness and the subagent daemon connect with
 `_INBOX.agent.<agent>` as their inbox prefix, so the replies to the icon write
 and the schedule read arrive. An agent can't read another agent's terminal,
 state, input or replies. The queue refuses the hive's shared agent client
-outright. Hives and the swarm controller may still subscribe to any subject in
+outright.
-the account.
+
+Hives and the swarm controller get a subscribe allow-list too, each with its
+own reply inbox `_INBOX.client.<client-id>.>`, which `swarm_queue_client::connect`
+sets as their inbox prefix. A KV watch or key listing delivers there as well. A
+hive may also subscribe to `$SWARM.knowledge` and its own `$SWARM.deploy.<hive>`,
+so it can't read another hive's deploys or any agent's terminal. The controller
+may also subscribe to `$SWARM.term.<agent>`, `$SWARM.term.<agent>.sub.<subagent>`
+and `$SWARM.agent-state.<agent>`, the streams it relays to swarm-ui. A hive or
+controller that connects with the library's default `_INBOX.<nuid>` prefix gets
+no replies, so every request it makes times out.
 
 </details>
 

swarm-nats-auth/src/policy.rs

@@ -26,10 +26,9 @@ pub struct Permissions {
     /// Subjects allowed for subscribe, or `None` for every subject in the
     /// account.
     ///
-    /// `Some` only for agents, the least trusted clients: their own reply
+    /// Every grant [`Policy`] makes is `Some`: the subjects that client
-    /// inbox and their own input subject.
+    /// subscribes to plus its own reply inbox. Never `Some(vec![])`: NATS
-    /// Hives and readers get `None`. Never `Some(vec![])`: NATS reads an empty
+    /// reads an empty allow-list as no restriction at all.
-    /// allow-list as no restriction at all.
     pub subscribe: Option<Vec<String>>,
 }
 
@@ -182,18 +181,55 @@ impl Policy {
         if let Some(hive) = self.hive_name(client_id) {
             return Some(Permissions {
                 publish: self.hive_subjects(hive),
-                subscribe: None,
+                subscribe: Some(Self::hive_subscribe(client_id, hive)),
             });
         }
         if self.readers.iter().any(|r| r == client_id) {
             return Some(Permissions {
                 publish: self.reader_subjects(),
-                subscribe: None,
+                subscribe: Some(Self::reader_subscribe(client_id)),
             });
         }
         None
     }
 
+    /// What a hive may subscribe to: the two events the controller sends it,
+    /// and its own reply inbox.
+    ///
+    /// The inbox carries every `JetStream` reply and the wanted-bucket watch,
+    /// whose consumer delivers to a fresh inbox subject. The client sets the
+    /// matching prefix in `swarm_queue_client::connect`. A refused subscribe
+    /// is reported on the connection, not to the caller, so a subject missing
+    /// here shows up as a hive that silently stops hearing it.
+    fn hive_subscribe(client_id: &str, hive: &str) -> Vec<String> {
+        vec![
+            swarm_queue_client::KNOWLEDGE_SUBJECT.to_owned(),
+            swarm_queue_client::deploy_subject(hive),
+            Self::inbox(client_id),
+        ]
+    }
+
+    /// What a reader may subscribe to: every agent's terminal, its
+    /// subagents' terminals and its turn state, which the controller relays
+    /// to swarm-ui, and its own reply inbox.
+    ///
+    /// The three families are the publishers' (`hive-agent`'s `swarm_term`
+    /// and `swarm_agent_state`, `swarm_queue_client::subagent_term`), which no
+    /// shared constant names; a rename there must change these too. Operator
+    /// input, deploys and knowledge are absent: the reader publishes those.
+    fn reader_subscribe(client_id: &str) -> Vec<String> {
+        vec![
+            "$SWARM.term.*".to_owned(),
+            "$SWARM.term.*.sub.*".to_owned(),
+            "$SWARM.agent-state.*".to_owned(),
+            Self::inbox(client_id),
+        ]
+    }
+
+    fn inbox(client_id: &str) -> String {
+        format!("{}.>", swarm_queue_client::client_inbox_prefix(client_id))
+    }
+
     /// The hive a client id names, when it names one.
     ///
     /// The prefix is configuration, not a pattern this module guesses at: the
@@ -901,12 +937,12 @@ mod tests {
         let grants = p
             .subscribe
             .as_ref()
-            .expect("an agent's subscribe is an allow-list, never unrestricted");
+            .expect("a grant's subscribe is an allow-list, never unrestricted");
         assert!(!grants.is_empty(), "an empty allow-list is unrestricted");
         grants.iter().any(|g| covers(g, subject))
     }
 
-    /// The reply inbox each agent client connects with, as a subject a
+    /// The reply inbox each client connects with, as a subject a
     /// request under it would receive on.
     fn reply(prefix: &str) -> String {
         format!("{prefix}.nuid123.token456")
@@ -939,16 +975,91 @@ mod tests {
         }
     }
 
+    /// The subscribe the reply multiplexer makes under `prefix`: a wildcard,
+    /// so the grant has to cover it as a subset, not just one reply.
+    fn multiplexer(prefix: &str) -> String {
+        format!("{prefix}.nuid123.*")
+    }
+
     #[test]
-    fn hives_and_readers_keep_an_unrestricted_subscribe() {
+    fn a_hive_may_subscribe_to_its_own_events_and_inbox_only() {
-        let p = policy();
+        use swarm_queue_client::{
-        assert_eq!(p.permissions("hive-alpha").expect("a hive").subscribe, None);
+            KNOWLEDGE_SUBJECT, agent_inbox_prefix, agent_input_subject, client_inbox_prefix,
-        assert_eq!(
+            deploy_subject,
-            p.permissions("swarm-controller")
+        };
-                .expect("the reader")
+
-                .subscribe,
+        let alpha = policy().permissions("hive-alpha").expect("a hive");
-            None
+        let inbox = client_inbox_prefix("hive-alpha");
-        );
+        for allowed in [
+            KNOWLEDGE_SUBJECT.to_owned(),
+            deploy_subject("alpha"),
+            reply(&inbox),
+            multiplexer(&inbox),
+        ] {
+            assert!(
+                may_subscribe(&alpha, &allowed),
+                "alpha may not subscribe to {allowed}"
+            );
+        }
+        for refused in [
+            ">".to_owned(),
+            "$SWARM.>".to_owned(),
+            deploy_subject("beta"),
+            swarm_queue_client::DEPLOY_SUBJECT_WILDCARD.to_owned(),
+            "$SWARM.term.>".to_owned(),
+            "$SWARM.term.atlas".to_owned(),
+            "$SWARM.agent-state.atlas".to_owned(),
+            agent_input_subject("atlas"),
+            reply(&agent_inbox_prefix("atlas")),
+            reply(&client_inbox_prefix("hive-beta")),
+            reply(&client_inbox_prefix("swarm-controller")),
+            "_INBOX.nuid123.token456".to_owned(),
+        ] {
+            assert!(
+                !may_subscribe(&alpha, &refused),
+                "alpha may subscribe to {refused}"
+            );
+        }
+    }
+
+    #[test]
+    fn the_reader_may_subscribe_to_agent_streams_and_its_inbox_only() {
+        use swarm_queue_client::{
+            KNOWLEDGE_SUBJECT, agent_inbox_prefix, agent_input_subject, client_inbox_prefix,
+            deploy_subject,
+        };
+
+        let reader = policy()
+            .permissions("swarm-controller")
+            .expect("the reader");
+        let inbox = client_inbox_prefix("swarm-controller");
+        for allowed in [
+            "$SWARM.term.atlas".to_owned(),
+            swarm_queue_client::subagent_term::subject("atlas", "scout"),
+            "$SWARM.agent-state.atlas".to_owned(),
+            reply(&inbox),
+            multiplexer(&inbox),
+        ] {
+            assert!(
+                may_subscribe(&reader, &allowed),
+                "the reader may not subscribe to {allowed}"
+            );
+        }
+        for refused in [
+            ">".to_owned(),
+            "$SWARM.>".to_owned(),
+            agent_input_subject("atlas"),
+            deploy_subject("alpha"),
+            KNOWLEDGE_SUBJECT.to_owned(),
+            reply(&agent_inbox_prefix("atlas")),
+            reply(&client_inbox_prefix("hive-alpha")),
+            "_INBOX.nuid123.token456".to_owned(),
+        ] {
+            assert!(
+                !may_subscribe(&reader, &refused),
+                "the reader may subscribe to {refused}"
+            );
+        }
     }
 
     #[test]

swarm-queue-client/src/lib.rs

@@ -214,12 +214,8 @@ pub const KNOWLEDGE_SUBJECT: &str = "$SWARM.knowledge";
 /// of its agents. Lives here for the same three-crate reason as
 /// [`KNOWLEDGE_SUBJECT`], but unlike it this is a **family, one subject per
 /// hive** — a hive subscribes to its own and is never woken by a deploy meant
-/// for someone else.
+/// for someone else. The queue grants each hive subscribe on its own subject
-///
+/// only, so it cannot read another hive's deploys either.
-/// ⚠️ **That split is about noise, not confidentiality.** The auth-callout
-/// responder scopes *publish* only and leaves `sub` unrestricted, so a hive
-/// that wanted another's messages could still subscribe to them. What the
-/// family buys is that it does not receive them by default.
 #[must_use]
 pub fn deploy_subject(hive: &str) -> String {
     format!("{DEPLOY_SUBJECT_PREFIX}.{hive}")
@@ -289,6 +285,19 @@ pub fn agent_inbox_prefix(agent: &str) -> String {
     format!("_INBOX.agent.{agent}")
 }
 
+/// The reply-inbox prefix [`connect`] uses for the client `client_id`: a hive
+/// or the swarm controller.
+///
+/// The queue grants that client subscribe on this prefix's `.>`, so a request
+/// made under the library's default `_INBOX.<nuid>` is refused its reply and
+/// times out, as is a KV watch or key listing, whose consumer delivers there
+/// too. The `client` token keeps it disjoint from [`agent_inbox_prefix`] and
+/// from the default.
+#[must_use]
+pub fn client_inbox_prefix(client_id: &str) -> String {
+    format!("_INBOX.client.{client_id}")
+}
+
 /// What an [`agent_input_subject`] message carries.
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct AgentInput {
@@ -755,7 +764,8 @@ pub fn install_crypto_provider() {
 }
 
 /// Connect to the swarm queue, presenting a token on each connection attempt
-/// and minting a fresh one only when the cached one is near expiry.
+/// and minting a fresh one only when the cached one is near expiry. Reply
+/// inboxes are under [`client_inbox_prefix`] of `cfg.client_id`.
 ///
 /// The rejected alternative — mint once, pass a static `auth_token`, own the
 /// reconnect loop — fails in the way this subsystem exists to prevent: the
@@ -773,6 +783,7 @@ pub async fn connect(cfg: QueueConfig) -> Result<async_nats::Client, Error> {
     let http = build_http_client(&cfg)?;
     let url = cfg.url.clone();
     let ca_file = cfg.ca_file.clone();
+    let inbox_prefix = client_inbox_prefix(&cfg.client_id);
 
     // Shared across every invocation of the callback below, which is the
     // whole point: the callback fires per connection ATTEMPT, so without
@@ -837,7 +848,8 @@ pub async fn connect(cfg: QueueConfig) -> Result<async_nats::Client, Error> {
     // near expiry. (Before the cache, "runs the callback" meant "mints",
     // and this retry loop was a token-request loop. See
     // `MAX_RECONNECT_DELAY`.)
-    .retry_on_initial_connect();
+    .retry_on_initial_connect()
+    .custom_inbox_prefix(inbox_prefix);
 
     // The queue's leaf chains to the swarm's own PKI root, which a host
     // process's platform store does not hold: the same anchor the token