diff --git a/docs/swarm/README.md b/docs/swarm/README.md index bc43d1e6..2ea8e5bb 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -233,6 +233,16 @@ key. Swarm-side, `GET /api/agents//icon` serves the stored bytes, and 404 means the agent has no icon. swarm-ui's agent cards load it as an `` and show the dimmed hyperhive mark for an agent without one, as the hive dashboard does. +The queue also limits what an agent may **subscribe** to, because agents are the +least trusted clients on the queue. An agent's own queue credential, which its +subagent daemon also presents, may subscribe to its reply inbox +`_INBOX.agent..>` and its operator-input subject `$SWARM.input.`, +and nothing else. The hive's shared agent client may subscribe to its reply +inbox `_INBOX.client..>` only. Each connects with its prefix as its +inbox prefix, so the replies to the icon write and the schedule read arrive. +Neither can read another agent's terminal, state, input or replies. Hives and +the swarm controller may still subscribe to any subject in the account. + ### Swarm-wide forge objects diff --git a/docs/swarm/ui.md b/docs/swarm/ui.md index 1e462ddb..9e6fb0db 100644 --- a/docs/swarm/ui.md +++ b/docs/swarm/ui.md @@ -78,7 +78,9 @@ request, replacing any the browser sent. `POST /api/agents//input` takes that header as the sender and sends `{from, body}` as a NATS request on `$SWARM.input.`. swarm-controller is the only queue client allowed to publish there. The agent's harness subscribes to its own subject and queues -each message for its turn loop with its reminders. The harness doesn't reply, +each message for its turn loop with its reminders. Only an agent connected with +its own queue credential may subscribe there; an agent on its hive's shared +client takes no input this way, and the request gets the 503 below. The harness doesn't reply, so swarm-controller answers 202 when the request times out after two seconds, and 503 when the queue reports no subscriber. diff --git a/hive-agent/src/swarm_input.rs b/hive-agent/src/swarm_input.rs index c77b1702..238dc5b1 100644 --- a/hive-agent/src/swarm_input.rs +++ b/hive-agent/src/swarm_input.rs @@ -12,6 +12,10 @@ //! publish on the subject (`swarm-nats-auth`'s reader grant), and it names //! the sender from the swarm gateway's authelia answer. //! +//! **Own credential only.** The queue grants subscribe on this subject to the +//! agent's own credential alone; the hive's shared client names no agent, so +//! it cannot be granted one agent's input without every agent's. +//! //! **Core NATS, no reply.** A message sent while this task is not subscribed //! is gone; the controller sends it as a request so the operator gets a 503 //! for that instead. No agent grant covers a reply subject, so nothing here @@ -21,7 +25,7 @@ use futures_util::{Stream, StreamExt as _}; use hive_sh4re::inbox::DeliveredMessage; use tokio::sync::mpsc::UnboundedSender; -use crate::swarm_queue::Connection; +use crate::swarm_queue::{Connection, Presented}; /// Start the subscriber, if this agent has a queue credential and a label to /// name its subject with. @@ -38,9 +42,16 @@ pub fn spawn(tx: UnboundedSender) { } async fn run(tx: UnboundedSender, agent: String) { - let Some(Connection { client, .. }) = crate::swarm_queue::client().await else { + let Some(Connection { client, presented }) = crate::swarm_queue::client().await else { return; }; + if presented != Presented::Agent { + tracing::warn!( + "connected with the hive's shared client, which may not subscribe to an \ + agent's input subject; not taking input from the swarm queue" + ); + return; + } let subject = swarm_queue_client::agent_input_subject(&agent); let subscriber = match client.subscribe(subject.clone()).await { Ok(subscriber) => subscriber, diff --git a/hive-agent/src/swarm_queue.rs b/hive-agent/src/swarm_queue.rs index 23cb2017..607eb1cb 100644 --- a/hive-agent/src/swarm_queue.rs +++ b/hive-agent/src/swarm_queue.rs @@ -492,7 +492,8 @@ async fn connect_once() -> Option { } } let cfg = config()?; - match swarm_queue_client::connect(cfg.clone()).await { + let inbox = swarm_queue_client::client_inbox_prefix(&cfg.client_id); + match swarm_queue_client::connect_with_inbox_prefix(cfg.clone(), inbox).await { Ok(client) => { tracing::info!( url = %cfg.url, @@ -534,7 +535,8 @@ async fn connect_as_agent( ) -> anyhow::Result { secret.prime().await?; let secret = Arc::new(secret); - swarm_queue_client::connect_with_token(url, ca_file, move || { + let agent = name.clone(); + swarm_queue_client::connect_with_token(url, ca_file, &agent, move || { let secret = Arc::clone(&secret); let name = name.clone(); // Spawned so the future handed back is only a `JoinHandle`, which is diff --git a/hive-subagent-mcp/src/swarm_term.rs b/hive-subagent-mcp/src/swarm_term.rs index 46f69284..21a21462 100644 --- a/hive-subagent-mcp/src/swarm_term.rs +++ b/hive-subagent-mcp/src/swarm_term.rs @@ -228,7 +228,8 @@ impl StoreSource { async fn connect(target: &Arc) -> anyhow::Result { let url = target.url.clone(); let target = Arc::clone(target); - swarm_queue_client::connect_with_token(&url, None, move || { + let agent = target.agent.clone(); + swarm_queue_client::connect_with_token(&url, None, &agent, move || { let target = Arc::clone(&target); // Spawned so the future handed back is only a `JoinHandle`, which is // `Sync` as async-nats's auth callback requires; the store read is not. diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 15a32251..85be9c9a 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -17,19 +17,20 @@ //! 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`. +/// The subjects an admitted client may publish and subscribe to. #[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, + /// Subjects allowed for subscribe, or `None` for every subject in the + /// account. + /// + /// `Some` only for agents, the least trusted clients: their own reply + /// inbox, plus their own input subject for an agent's own credential. + /// Hives and readers get `None`. Never `Some(vec![])`: NATS reads an empty + /// allow-list as no restriction at all. + pub subscribe: Option>, } /// Which client ids get which permissions. @@ -168,7 +169,13 @@ impl Policy { .iter() .map(|s| s.replace(AGENT_PLACEHOLDER, agent)) .collect(); - (!publish.is_empty()).then_some(Permissions { publish }) + (!publish.is_empty()).then(|| Permissions { + publish, + subscribe: Some(vec![ + format!("{}.>", swarm_queue_client::agent_inbox_prefix(agent)), + swarm_queue_client::agent_input_subject(agent), + ]), + }) } /// The permissions for `client_id`, or `None` when no rule matches. @@ -191,18 +198,30 @@ impl Policy { // for agents, expanded for that hive, and **nothing when it configured // none**: an empty publish list would be a denial wearing a grant's // shape, so this returns `None` and the client is refused outright. + // + // Subscribe is its own reply inbox alone. The identity names no agent, + // so it cannot be granted one agent's input subject without all of + // them. if let Some(hive) = self.agent_hive(client_id) { let publish = self.agent_subjects(hive); - return (!publish.is_empty()).then_some(Permissions { publish }); + return (!publish.is_empty()).then(|| Permissions { + publish, + subscribe: Some(vec![format!( + "{}.>", + swarm_queue_client::client_inbox_prefix(client_id) + )]), + }); } if let Some(hive) = self.hive_name(client_id) { return Some(Permissions { publish: self.hive_subjects(hive), + subscribe: None, }); } if self.readers.iter().any(|r| r == client_id) { return Some(Permissions { publish: self.reader_subjects(), + subscribe: None, }); } None @@ -921,6 +940,103 @@ mod tests { assert_eq!(policy().permissions(""), None); } + /// Whether the grant `grant` covers the concrete subject `subject`, by + /// NATS' token rules: `*` is one token, `>` one or more. + fn covers(grant: &str, subject: &str) -> bool { + let g: Vec<&str> = grant.split('.').collect(); + let s: Vec<&str> = subject.split('.').collect(); + for (i, t) in g.iter().enumerate() { + match *t { + ">" => return s.len() > i, + "*" if i < s.len() => {} + t if s.get(i) == Some(&t) => {} + _ => return false, + } + } + g.len() == s.len() + } + + fn may_subscribe(p: &Permissions, subject: &str) -> bool { + let grants = p + .subscribe + .as_ref() + .expect("an agent'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 + /// request under it would receive on. + fn reply(prefix: &str) -> String { + format!("{prefix}.nuid123.token456") + } + + #[test] + fn an_agent_token_may_subscribe_to_its_own_inbox_and_input_only() { + use swarm_queue_client::{agent_inbox_prefix, agent_input_subject, client_inbox_prefix}; + + let p = policy_with_agent_subject() + .with_agent_token_subjects(vec!["$SWARM.term.{agent}".to_owned()]) + .expect("a per-agent template is valid"); + let atlas = p.agent_token_permissions("atlas").expect("configured"); + assert!(may_subscribe(&atlas, &reply(&agent_inbox_prefix("atlas")))); + assert!(may_subscribe(&atlas, &agent_input_subject("atlas"))); + for refused in [ + ">".to_owned(), + "$SWARM.term.>".to_owned(), + "$SWARM.term.argus".to_owned(), + "$SWARM.term.atlas".to_owned(), + "$SWARM.input.*".to_owned(), + agent_input_subject("argus"), + reply(&agent_inbox_prefix("argus")), + reply(&client_inbox_prefix("hive-alpha-agent")), + "_INBOX.nuid123.token456".to_owned(), + ] { + assert!( + !may_subscribe(&atlas, &refused), + "atlas may subscribe to {refused}" + ); + } + } + + #[test] + fn the_hives_shared_agent_client_may_subscribe_to_its_own_inbox_only() { + use swarm_queue_client::{agent_inbox_prefix, agent_input_subject, client_inbox_prefix}; + + let p = policy_with_agent_subject(); + let alpha = p.permissions("hive-alpha-agent").expect("alpha's agents"); + assert!(may_subscribe( + &alpha, + &reply(&client_inbox_prefix("hive-alpha-agent")) + )); + for refused in [ + ">".to_owned(), + "$SWARM.term.>".to_owned(), + "$SWARM.term.alpha.atlas".to_owned(), + agent_input_subject("atlas"), + reply(&client_inbox_prefix("hive-beta-agent")), + reply(&agent_inbox_prefix("atlas")), + "_INBOX.nuid123.token456".to_owned(), + ] { + assert!( + !may_subscribe(&alpha, &refused), + "alpha's agents may subscribe to {refused}" + ); + } + } + + #[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 + ); + } + #[test] fn the_prefix_alone_names_no_hive() { // `hive-` would otherwise produce `$KV.hive-status.`, a subject nobody @@ -1313,19 +1429,6 @@ mod tests { assert_eq!(argus.publish, vec![direct_get_subject("argus")]); assert!(!atlas.publish.contains(&direct_get_subject("argus"))); - let covers = |grant: &str, subject: &str| { - let g: Vec<&str> = grant.split('.').collect(); - let s: Vec<&str> = subject.split('.').collect(); - for (i, t) in g.iter().enumerate() { - match *t { - ">" => return s.len() > i, - "*" if i < s.len() => {} - t if s.get(i) == Some(&t) => {} - _ => return false, - } - } - g.len() == s.len() - }; for refused in [ direct_get_subject("argus"), format!("$JS.API.DIRECT.GET.{STREAM}"), diff --git a/swarm-nats-auth/src/respond.rs b/swarm-nats-auth/src/respond.rs index 49402ed9..8ae52c61 100644 --- a/swarm-nats-auth/src/respond.rs +++ b/swarm-nats-auth/src/respond.rs @@ -157,6 +157,9 @@ struct UserNats { /// field the server rejected while the responder said `granted=true`. #[serde(rename = "pub")] publish: Permission, + /// Subjects this client may subscribe to; absent means every subject. + #[serde(rename = "sub", skip_serializing_if = "Option::is_none")] + subscribe: Option, subs: i64, data: i64, payload: i64, @@ -186,6 +189,7 @@ fn user_jwt( account: &str, user_nkey: &str, publish: Vec, + subscribe: Option>, ) -> String { sign( Claims { @@ -198,6 +202,7 @@ fn user_jwt( nats: UserNats { kind: "user", publish: Permission { allow: publish }, + subscribe: subscribe.map(|allow| Permission { allow }), subs: -1, data: -1, payload: -1, @@ -223,7 +228,14 @@ pub fn grant( permissions: &crate::policy::Permissions, ) -> String { let now = now_secs(); - let jwt = user_jwt(now, issuer, account, user_nkey, permissions.publish.clone()); + let jwt = user_jwt( + now, + issuer, + account, + user_nkey, + permissions.publish.clone(), + permissions.subscribe.clone(), + ); response(now, issuer, server_id, user_nkey, Ok(jwt)) } @@ -346,6 +358,7 @@ mod tests { fn perms() -> crate::policy::Permissions { crate::policy::Permissions { publish: vec!["$KV.hive-status.alpha".to_owned()], + subscribe: None, } } @@ -369,6 +382,25 @@ mod tests { ); } + #[test] + fn the_issued_user_jwt_carries_a_subscribe_allow_list_only_when_given_one() { + let account = KeyPair::new_account(); + let scoped = crate::policy::Permissions { + subscribe: Some(vec!["_INBOX.agent.atlas.>".to_owned()]), + ..perms() + }; + let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &scoped)); + let user = decode_claims(wrapper["nats"]["jwt"].as_str().expect("a user jwt")); + assert_eq!( + user["nats"]["sub"]["allow"], + serde_json::json!(["_INBOX.agent.atlas.>"]) + ); + + let wrapper = decode_claims(&grant(&account, "APP", "NSERVER", "UCLIENT", &perms())); + let user = decode_claims(wrapper["nats"]["jwt"].as_str().expect("a user jwt")); + assert!(user["nats"]["sub"].is_null(), "{}", user["nats"]); + } + #[test] fn a_grant_carries_a_user_jwt_and_no_error() { let account = KeyPair::new_account(); diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index c66c8393..e6f6d918 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -277,6 +277,27 @@ pub const AGENT_INPUT_SUBJECT_WILDCARD: &str = "$SWARM.input.*"; /// Shared by [`agent_input_subject`] and [`AGENT_INPUT_SUBJECT_WILDCARD`]. const AGENT_INPUT_SUBJECT_PREFIX: &str = "$SWARM.input"; +/// The reply-inbox prefix a client presenting `agent`'s own credential +/// connects with. +/// +/// The queue grants that credential subscribe on this prefix's `.>` and its +/// [`agent_input_subject`], nothing else, so a request made under the +/// library's default `_INBOX.` is refused its reply and times out. The +/// `agent` token keeps it disjoint from [`client_inbox_prefix`] and from the +/// default, whose second token is a 22-character nuid. +#[must_use] +pub fn agent_inbox_prefix(agent: &str) -> String { + format!("_INBOX.agent.{agent}") +} + +/// The reply-inbox prefix for a client authenticating as the OIDC client +/// `client_id`, for a client whose grant limits subscribe to this prefix's +/// `.>` — see [`agent_inbox_prefix`]. +#[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 { @@ -750,6 +771,22 @@ pub fn install_crypto_provider() { /// token expires, the controller keeps serving, its status data quietly stops /// updating, and nothing says so until someone reads a dashboard. pub async fn connect(cfg: QueueConfig) -> Result { + connect_inner(cfg, None).await +} + +/// [`connect`], with reply inboxes under `inbox_prefix`. For a client whose +/// grant limits subscribe to that prefix, such as [`client_inbox_prefix`]. +pub async fn connect_with_inbox_prefix( + cfg: QueueConfig, + inbox_prefix: String, +) -> Result { + connect_inner(cfg, Some(inbox_prefix)).await +} + +async fn connect_inner( + cfg: QueueConfig, + inbox_prefix: Option, +) -> Result { install_crypto_provider(); // Trusts `cfg.ca_file` when set (the swarm's own CA, when the token // endpoint is signed by it) — see `build_http_client`. A timeout too, @@ -834,6 +871,9 @@ pub async fn connect(cfg: QueueConfig) -> Result { if let Some(path) = ca_file { options = options.add_root_certificates(path); } + if let Some(prefix) = inbox_prefix { + options = options.custom_inbox_prefix(prefix); + } let client = options .connect(&url) @@ -846,8 +886,8 @@ pub async fn connect(cfg: QueueConfig) -> Result { Ok(client) } -/// Connect presenting the token `token` resolves to, as an agent presents its -/// own queue credential. +/// Connect presenting the token `token` resolves to, as `agent` presents its +/// own queue credential, with reply inboxes under [`agent_inbox_prefix`]. /// /// `token` runs once per connection attempt, reconnects included, so a /// credential replaced at its source is presented on the next attempt. An @@ -861,6 +901,7 @@ pub async fn connect(cfg: QueueConfig) -> Result { pub async fn connect_with_token( url: &str, ca_file: Option<&std::path::Path>, + agent: &str, token: F, ) -> Result where @@ -875,7 +916,8 @@ where Ok(auth) } }) - .reconnect_delay_callback(reconnect_delay); + .reconnect_delay_callback(reconnect_delay) + .custom_inbox_prefix(agent_inbox_prefix(agent)); if let Some(path) = ca_file { options = options.add_root_certificates(path.to_path_buf()); }