diff --git a/docs/swarm/README.md b/docs/swarm/README.md index f891b997..1123f741 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -256,8 +256,17 @@ and nothing else. The harness and the subagent daemon connect with `_INBOX.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 -the account. +outright. + +Hives and the swarm controller get a subscribe allow-list too, each with its +own reply inbox `_INBOX.client..>`, 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.`, +so it can't read another hive's deploys or any agent's terminal. The controller +may also subscribe to `$SWARM.term.`, `$SWARM.term..sub.` +and `$SWARM.agent-state.`, the streams it relays to swarm-ui. A hive or +controller that connects with the library's default `_INBOX.` prefix gets +no replies, so every request it makes times out. diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index cd6c226c..b72b9dfd 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/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 - /// inbox and their own input subject. - /// Hives and readers get `None`. Never `Some(vec![])`: NATS reads an empty - /// allow-list as no restriction at all. + /// Every grant [`Policy`] makes is `Some`: the subjects that client + /// subscribes to plus its own reply inbox. Never `Some(vec![])`: NATS + /// reads an empty allow-list as no restriction at all. pub subscribe: Option>, } @@ -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 { + 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 { + 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() { - let p = policy(); - assert_eq!(p.permissions("hive-alpha").expect("a hive").subscribe, None); - assert_eq!( - p.permissions("swarm-controller") - .expect("the reader") - .subscribe, - None - ); + fn a_hive_may_subscribe_to_its_own_events_and_inbox_only() { + use swarm_queue_client::{ + KNOWLEDGE_SUBJECT, agent_inbox_prefix, agent_input_subject, client_inbox_prefix, + deploy_subject, + }; + + let alpha = policy().permissions("hive-alpha").expect("a hive"); + 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] diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 7ff3c597..ae1efa3c 100644 --- a/swarm-queue-client/src/lib.rs +++ b/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. -/// -/// ⚠️ **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. +/// for someone else. The queue grants each hive subscribe on its own subject +/// only, so it cannot read another hive's deploys either. #[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.` 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 { 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 { // 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