vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

swarm: let every hive read every agent's credential, and say so

Browse source 
A hive reads its agents' credentials with its own certificate, and nothing
said which paths that certificate may read, so the read half of a delivery
answered 403.

The grant is wide on purpose. An agent's path does not name the hive
hosting it -- agents move -- so a per-hive grant has to be an enumeration
the controller re-emits whenever the roster changes, and an enumeration
that can drift or land out of order advertises a boundary it does not
hold. A wide grant that says what it is beats a narrow one that only looks
narrow. mara's call, on the PR: rather a too-lax scope than one that
pretends to be strict.

What that buys, beyond honesty: the document is identical for every hive
and depends on nothing, so it is written once at startup beside the rest of
a hive's provisioning instead of on every declaration. No derived state, no
re-emission, and the ordering hazard that came with one stops existing.

What still holds is read-only. A hive cannot write an agent's credential,
so it cannot hand itself an agent's identity, and the grant reaches nothing
in the store outside the agent-credential prefix.

The fact is documented where someone meets the boundary rather than only in
this message, and the two ways to narrow it later -- scope per hive, or
give agents their own store identity -- are tracked.
This commit is contained in:
atlas 2026-09-09 17:41:31 +02:00 • committed by mara
commit 3752482524
5 changed files with 119 additions and 306 deletions
Download patch file Download diff file Expand all files Collapse all files

171
swarm-controller/src/wanted.rs
View file

@ -13,14 +13,12 @@
//! declaration to reconcile against, because the current value can be read
//! back from the queue whenever it is needed.
use std::{fmt, sync::Arc};
use std::fmt;
use anyhow::{Context, Result};
use async_nats::jetstream::kv::{CreateErrorKind, UpdateErrorKind};
use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted};
use crate::read_policy::ReadPolicySink;
/// The one failure out of [`apply`] that is the caller's mistake, not a
/// server fault — worth a distinct type so `set_agent_state` can tell it
/// apart from every other error this module raises (an undecodable
@ -103,60 +101,15 @@ fn apply(current: Option<&[u8]>, agent: &str, state: AgentState) -> Result<(Hive
Ok((declaration, encoded))
}
/// Which of a declaration's agents a hive may read credentials for.
///
/// Not all of them: the declared set is the hive's whole history, destroyed
/// entries included — they stay so that redeclaring one back to `Up` can be
/// refused as the terminal transition it is, rather than silently
/// un-destroying an agent. A torn-down agent keeping read access on its
/// credentials is what this filter exists to stop.
fn readable_agents(declaration: &HiveWanted) -> Vec<&str> {
declaration
.agents
.iter()
.filter(|(_, wanted)| wanted.state != AgentState::Destroyed)
.map(|(agent, _)| agent.as_str())
.collect()
}
/// Re-render `hive`'s read grant from the declaration about to be published.
///
/// Separate from [`WantedWriter::set`] so the trigger is assertable: `set`
/// needs a live queue, this needs nothing, and "was the grant emitted, over
/// exactly which agents" is the half that fails as a 403 nobody can trace.
async fn publish_read_grant(
sink: Option<&Arc<dyn ReadPolicySink>>,
hive: &str,
declaration: &HiveWanted,
) -> Result<()> {
let Some(sink) = sink else {
return Ok(());
};
let agents = readable_agents(declaration);
sink.publish(hive, &agents)
.await
.with_context(|| format!("publishing the read grant for {hive}"))
}
/// Writes the wanted-state bucket, and reads it back.
pub struct WantedWriter {
client: async_nats::Client,
/// Where a hive's read grant is published, or `None` on a deployment
/// with no store identity — see [`WantedWriter::new`].
read_policy: Option<Arc<dyn ReadPolicySink>>,
}
impl WantedWriter {
/// `read_policy` is `None` when this deployment has no secret store to
/// authenticate to. A declaration still publishes; there is simply no
/// grant to keep in step with it, and the credential delivery it would
/// have served does not exist either.
#[must_use]
pub fn new(client: async_nats::Client, read_policy: Option<Arc<dyn ReadPolicySink>>) -> Self {
Self {
client,
read_policy,
}
pub fn new(client: async_nats::Client) -> Self {
Self { client }
}
/// One hive's bucket handle, created on first use if nothing has made it
@ -217,14 +170,6 @@ impl WantedWriter {
let (declaration, encoded) =
apply(entry.as_ref().map(|e| e.value.as_ref()), agent, state)?;
// Before the write, not after: a grant that lands late is a 403
// on the agent's first credential fetch, while
// a grant that shrinks a moment early only affects an agent
// already being torn down. If the write below then fails, the
// grant is a harmless superset and the next declaration
// re-renders it.
publish_read_grant(self.read_policy.as_ref(), hive, &declaration).await?;
// `update` and `create` have separate error types, and only one
// variant of each means "someone else got there first". Every
// other failure returns immediately: retrying a disconnect or a
@ -264,114 +209,8 @@ impl WantedWriter {
#[cfg(test)]
mod tests {
use super::{apply, publish_read_grant, readable_agents};
use std::sync::{Arc, Mutex};
use swarm_queue_client::wanted::{AgentState, HiveWanted};
/// Records what a caller published instead of talking to a store.
#[derive(Default)]
struct Recorder {
published: Mutex<Vec<(String, Vec<String>)>>,
}
#[async_trait::async_trait]
impl super::ReadPolicySink for Recorder {
async fn publish(&self, hive: &str, agents: &[&str]) -> anyhow::Result<()> {
self.published.lock().unwrap().push((
hive.to_owned(),
agents.iter().map(|a| (*a).to_owned()).collect(),
));
Ok(())
}
}
/// A sink that always refuses, to pin that a failed grant is not swallowed.
struct Refuses;
#[async_trait::async_trait]
impl super::ReadPolicySink for Refuses {
async fn publish(&self, _hive: &str, _agents: &[&str]) -> anyhow::Result<()> {
anyhow::bail!("the store said no")
}
}
fn declaration(pairs: &[(&str, AgentState)]) -> HiveWanted {
let mut declaration = HiveWanted::default();
for (agent, state) in pairs {
let (d, _) = apply(
Some(&serde_json::to_vec(&declaration).unwrap()),
agent,
*state,
)
.unwrap();
declaration = d;
}
declaration
}
#[test]
fn a_destroyed_agent_is_not_granted_read_while_the_live_ones_are() {
let d = declaration(&[
("iris", AgentState::Up),
("argus", AgentState::Offline),
("atlas", AgentState::Destroyed),
]);
let mut granted = readable_agents(&d);
granted.sort_unstable();
assert_eq!(
granted,
["argus", "iris"],
"destroyed keeps its declaration"
);
}
/// The control on the filter above: without it every assertion there
/// could pass because `readable_agents` returns nothing at all.
#[test]
fn a_declaration_with_no_destroyed_agent_grants_every_one_of_them() {
let d = declaration(&[("iris", AgentState::Up), ("argus", AgentState::Offline)]);
assert_eq!(readable_agents(&d).len(), 2);
}
#[tokio::test]
async fn declaring_an_agent_publishes_the_grant_for_exactly_the_live_set() {
let recorder = Arc::new(Recorder::default());
let sink: Arc<dyn super::ReadPolicySink> = recorder.clone();
let d = declaration(&[("iris", AgentState::Up), ("atlas", AgentState::Destroyed)]);
publish_read_grant(Some(&sink), "pr1ma", &d).await.unwrap();
let published = recorder.published.lock().unwrap().clone();
assert_eq!(
published,
vec![("pr1ma".to_owned(), vec!["iris".to_owned()])]
);
}
/// A deployment with no store still declares agent state — there is just
/// no grant to keep in step with it.
#[tokio::test]
async fn no_sink_is_a_no_op_rather_than_a_failure() {
let d = declaration(&[("iris", AgentState::Up)]);
publish_read_grant(None, "pr1ma", &d)
.await
.expect("a deployment with no store is not an error");
}
/// The failure this ordering exists for: a refused grant must stop the
/// declaration, not publish one whose reader will 403.
#[tokio::test]
async fn a_refused_grant_is_an_error_the_caller_sees() {
let sink: Arc<dyn super::ReadPolicySink> = Arc::new(Refuses);
let d = declaration(&[("iris", AgentState::Up)]);
let err = publish_read_grant(Some(&sink), "pr1ma", &d)
.await
.expect_err("the sink refused");
assert!(
format!("{err:#}").contains("the store said no"),
"the underlying refusal should survive the context: {err:#}"
);
}
use super::apply;
use swarm_queue_client::wanted::AgentState;
#[test]
fn declaring_one_agent_preserves_every_other() {