diff --git a/hive-c0re/src/workers/wanted.rs b/hive-c0re/src/workers/wanted.rs index 6fd39393..68ec3e17 100644 --- a/hive-c0re/src/workers/wanted.rs +++ b/hive-c0re/src/workers/wanted.rs @@ -63,7 +63,7 @@ pub async fn pull(coord: &Arc) -> Result<()> { // A hung node is worse than a failed one: the dashboard shows it running. swarm_queue_client::ensure_connected(&client)?; - let Some(store) = swarm_queue_client::wanted::open_read_only(&client).await else { + let Some(store) = swarm_queue_client::wanted::open_read_only(&client, &hive).await else { tracing::info!(%hive, "wanted state: no bucket yet; nothing has been declared"); return Ok(()); }; diff --git a/swarm-controller/src/wanted.rs b/swarm-controller/src/wanted.rs index b5d2c7ba..7c6314f5 100644 --- a/swarm-controller/src/wanted.rs +++ b/swarm-controller/src/wanted.rs @@ -57,29 +57,31 @@ fn apply(current: Option<&[u8]>, agent: &str, state: AgentState) -> Result<(Hive /// Writes the wanted-state bucket, and reads it back. pub struct WantedWriter { client: async_nats::Client, - store: tokio::sync::OnceCell, } impl WantedWriter { #[must_use] pub fn new(client: async_nats::Client) -> Self { - Self { - client, - store: tokio::sync::OnceCell::new(), - } + Self { client } } - /// The bucket handle, created on first use if nothing has made it yet. + /// One hive's bucket handle, created on first use if nothing has made it + /// yet. /// /// Creation lives in [`swarm_queue_client::wanted`] because a bucket is /// described identically by everyone who may create it. Only the - /// controller creates this one; a hive opens it read-only. + /// controller creates these; a hive opens its own read-only. + /// + /// Resolved per call rather than cached: there is one bucket per hive, so + /// a single cached handle cannot serve them, and the declarations this + /// writes change on operator action rather than on a loop — the extra + /// lookup is per *declaration*, not per tick. A cache here would be a map + /// whose invalidation nobody needs yet. async fn store( &self, - ) -> std::result::Result<&async_nats::jetstream::kv::Store, swarm_queue_client::Error> { - self.store - .get_or_try_init(|| swarm_queue_client::wanted::open_or_create(&self.client)) - .await + hive: &str, + ) -> std::result::Result { + swarm_queue_client::wanted::open_or_create(&self.client, hive).await } /// The declaration currently published for `hive`, or `None`. @@ -87,7 +89,7 @@ impl WantedWriter { // An unconnected client does not fail a JetStream request, it hangs // on it — see `swarm_queue_client::ensure_connected`. swarm_queue_client::ensure_connected(&self.client)?; - let store = self.store().await?; + let store = self.store(hive).await?; let Some(entry) = store .entry(hive) .await @@ -110,7 +112,7 @@ impl WantedWriter { /// fact — the conflict has to be caught here. pub async fn set(&self, hive: &str, agent: &str, state: AgentState) -> Result { swarm_queue_client::ensure_connected(&self.client)?; - let store = self.store().await?; + let store = self.store(hive).await?; for _ in 0..MAX_ATTEMPTS { let entry = store diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index f58fc1ab..4a0b5520 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -138,14 +138,14 @@ impl Policy { format!("KV_{}", self.bucket) } - /// The stream backing the wanted-state bucket. + /// The stream backing one hive's wanted-state bucket. /// - /// Derived from the crate constant rather than from a second configurable - /// bucket name: `wanted::BUCKET` is a `const` precisely so writer and - /// reader cannot disagree about it, and a flag here would reintroduce the - /// disagreement one layer out. - fn wanted_stream() -> String { - format!("KV_{}", swarm_queue_client::wanted::BUCKET) + /// Derived from the crate's own name function rather than from a + /// configurable bucket name: `wanted::bucket` is what the writer and the + /// reader both call, so a flag here would let two deployments disagree + /// about a name they must share. + fn wanted_stream(hive: &str) -> String { + format!("KV_{}", swarm_queue_client::wanted::bucket(hive)) } /// The stream backing the per-agent status bucket. @@ -256,19 +256,22 @@ impl Policy { // pointing at a subject nobody publishes to. Not `extra_hive_subjects`: // that is for streams this crate does not know about. subjects.push(swarm_queue_client::notices::subject(hive)); - // Reading the wanted-state bucket, and **only this hive's key**. + // Reading the wanted-state bucket, which is **this hive's own**. // - // A KV read is a publish: `store.get` is a request, and the direct-get - // subject carries the key, so the grant scopes to one hive rather than - // to the bucket. `$KV..` is deliberately absent — the - // controller declares this and a hive converges to it, so a hive that - // could write its own key could declare its own desired state. + // One bucket per hive rather than one keyed by hive, so that a *watch* + // can be granted without widening the read: a consumer's filter travels + // in the request payload, so `CONSUMER.CREATE` grants a whole stream and + // cannot be scoped to a key the way `DIRECT.GET` can. + // + // `$KV..` stays deliberately absent — the controller + // declares this and a hive converges to it, so a hive that could write + // its own key could declare its own desired state. + let wanted = Self::wanted_stream(hive); subjects.extend([ - format!("$JS.API.STREAM.INFO.{}", Self::wanted_stream()), + format!("$JS.API.STREAM.INFO.{wanted}"), format!( - "$JS.API.DIRECT.GET.{}.$KV.{}.{hive}", - Self::wanted_stream(), - swarm_queue_client::wanted::BUCKET + "$JS.API.DIRECT.GET.{wanted}.$KV.{}.{hive}", + swarm_queue_client::wanted::bucket(hive) ), ]); // Publishing this hive's agents into the per-agent status bucket, and @@ -349,13 +352,25 @@ impl Policy { // admission). Same failure mode as the knowledge event above: a // refused publish reaches the client as a timeout. swarm_queue_client::DEPLOY_SUBJECT_WILDCARD.to_owned(), - // The wanted-state bucket, which the controller creates and writes - // for every hive. `.>` here against one key per hive above: this is - // the single writer, so scoping it per hive would be a list of - // hives to keep, which `hive_name`'s doc argues against. - format!("$JS.API.STREAM.INFO.{}", Self::wanted_stream()), - format!("$JS.API.STREAM.CREATE.{}", Self::wanted_stream()), - format!("$KV.{}.>", swarm_queue_client::wanted::BUCKET), + // The wanted-state buckets — one per hive, all created and written + // by this single client. + // + // Wildcards rather than a name per hive: a bucket name is one + // subject token and subjects have no prefix matching, so + // `KV_hive-wanted-*` is not expressible. The choice is an exact + // name per hive — which needs a hive roster in this responder, + // which `hive_name`'s doc argues against — or `*`. + // + // ⚠️ `*` therefore reaches every stream in the account, including + // the buckets hives publish themselves, so this client *can* + // overwrite a hive's self-reported status. That is an accepted + // operator decision, not an oversight: the controller is the + // swarm's writer of record, and a swarm that cannot trust it has a + // larger problem than this grant. + "$JS.API.STREAM.INFO.*".to_owned(), + "$JS.API.STREAM.CREATE.*".to_owned(), + "$JS.API.DIRECT.GET.*.>".to_owned(), + "$KV.*.>".to_owned(), ]); // The per-agent status bucket, read-side only. Same five subjects as // the hive-status block above and for the same measured reasons — @@ -414,23 +429,38 @@ mod tests { } #[test] - fn the_reader_gets_no_write_subject_for_the_agent_status_bucket() { - // The controller reads this bucket and never writes it. A `$KV.` - // subject here would let a reader forge any agent's status on any - // hive — the containment `hive_subjects` establishes is only worth - // anything if the other role cannot bypass it. + fn no_hive_may_write_another_role_s_agent_status() { + // A hive publishes its *own* agents' status under + // `$KV.agent-status..*` and must reach no further. This is the + // arm of the old reader-side guard that still holds. let bucket = swarm_queue_client::agent_status::BUCKET; - let prefix = format!("$KV.{bucket}."); + let subjects = policy().hive_subjects("alpha"); + for forbidden in [ + format!("$KV.{bucket}.>"), + format!("$KV.{bucket}.*"), + format!("$KV.{bucket}.beta.*"), + ] { + assert!(!subjects.contains(&forbidden), "hive alpha got {forbidden}"); + } + // Control: the containment being asserted is not vacuous — the hive + // does hold its own slice. + assert!(subjects.contains(&format!("$KV.{bucket}.alpha.*"))); + } + + #[test] + fn the_readers_grant_is_deliberately_account_wide() { + // ⚠️ This pins a DECISION, not a safety property, and it replaced a + // guard that asserted the opposite — the operator's call was that the + // controller being able to override a hive is acceptable. + // + // A bucket name is one subject token and subjects have no prefix + // matching, so per-hive wanted-state buckets cannot be covered by any + // wildcard narrower than `*`. Scoping the controller would mean a hive + // roster inside this responder. If that trade is ever revisited, this + // test is the thing to change — it exists so the width reads as chosen. let subjects = policy().reader_subjects(); - assert!( - !subjects.iter().any(|s| s.starts_with(&prefix)), - "the reader was granted a {prefix}* subject" - ); - // Control: the prefix test does fire on a bucket the reader - // legitimately writes, so the pass above is about agent-status rather - // than about a matcher that never matches anything. - let wanted = format!("$KV.{}.", swarm_queue_client::wanted::BUCKET); - assert!(subjects.iter().any(|s| s.starts_with(&wanted))); + assert!(subjects.contains(&"$KV.*.>".to_owned())); + assert!(subjects.contains(&"$JS.API.STREAM.INFO.*".to_owned())); } #[test] @@ -736,8 +766,9 @@ mod tests { .permissions("hive-alpha") .expect("a hive is admitted"); assert!( - p.publish - .contains(&"$JS.API.DIRECT.GET.KV_hive-wanted.$KV.hive-wanted.alpha".to_owned()), + p.publish.contains( + &"$JS.API.DIRECT.GET.KV_hive-wanted-alpha.$KV.hive-wanted-alpha.alpha".to_owned() + ), "got: {:?}", p.publish ); @@ -749,7 +780,12 @@ mod tests { let p = policy() .permissions("hive-alpha") .expect("a hive is admitted"); - assert!(!p.publish.iter().any(|s| s.contains("hive-wanted.beta"))); + // ⚠️ Matched against the CURRENT bucket name. The per-hive split moved + // this from `hive-wanted.beta` to `hive-wanted-beta`, and the old + // spelling kept passing against a subject set that no longer contains + // it — a pass that had stopped meaning anything. + assert!(!p.publish.iter().any(|s| s.contains("hive-wanted-beta"))); + assert!(!p.publish.iter().any(|s| s.contains("beta"))); assert!( !p.publish .iter() @@ -764,8 +800,12 @@ mod tests { let p = policy() .permissions("hive-alpha") .expect("a hive is admitted"); - assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted.alpha")); - assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted.>")); + // Current names, for the reason the sibling test above records: the + // pre-split strings `$KV.hive-wanted.{alpha,>}` still "passed" after + // the rename because nothing produces them any more. + assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted-alpha.alpha")); + assert!(!p.publish.iter().any(|s| s == "$KV.hive-wanted-alpha.>")); + assert!(!p.publish.iter().any(|s| s.starts_with("$KV.hive-wanted"))); } #[test] @@ -775,11 +815,11 @@ mod tests { let p = policy() .permissions("swarm-controller") .expect("a reader is admitted"); - assert!(p.publish.contains(&"$KV.hive-wanted.>".to_owned())); - assert!( - p.publish - .contains(&"$JS.API.STREAM.CREATE.KV_hive-wanted".to_owned()) - ); + // Account-wide since the per-hive split: a bucket name is one subject + // token, so no wildcard narrower than `*` covers N per-hive buckets. + // See `the_readers_grant_is_deliberately_account_wide`. + assert!(p.publish.contains(&"$KV.*.>".to_owned())); + assert!(p.publish.contains(&"$JS.API.STREAM.CREATE.*".to_owned())); } #[test] diff --git a/swarm-queue-client/src/agent_status.rs b/swarm-queue-client/src/agent_status.rs index 9311fc6b..65104bc5 100644 --- a/swarm-queue-client/src/agent_status.rs +++ b/swarm-queue-client/src/agent_status.rs @@ -125,7 +125,7 @@ pub async fn open_or_create( }) .await .map_err(|source| Error::CreateBucket { - bucket: BUCKET, + bucket: BUCKET.to_owned(), source, }) } diff --git a/swarm-queue-client/src/status.rs b/swarm-queue-client/src/status.rs index 1f0e69c2..3e019e86 100644 --- a/swarm-queue-client/src/status.rs +++ b/swarm-queue-client/src/status.rs @@ -67,7 +67,7 @@ pub async fn open_or_create( }) .await .map_err(|source| Error::CreateBucket { - bucket: BUCKET, + bucket: BUCKET.to_owned(), source, }) } diff --git a/swarm-queue-client/src/wanted.rs b/swarm-queue-client/src/wanted.rs index 13df750e..be5e0ac0 100644 --- a/swarm-queue-client/src/wanted.rs +++ b/swarm-queue-client/src/wanted.rs @@ -170,7 +170,10 @@ mod tests { // the grant scoping in `swarm-nats-auth` names this string, so a bucket // the client refuses to open would surface as a permissions problem // rather than as the naming problem it is. - let legal = |s: &str| s.chars().all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-'); + let legal = |s: &str| { + s.chars() + .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-') + }; assert!(legal(&bucket("alpha"))); assert!(legal(&bucket("a-hive-with-hyphens"))); assert!(legal(&bucket("h9")));