vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

swarm-controller: check every first declaration, and fail closed on the roster

Browse source 
Exempting offline and paused from the name rules let a reserved name be
placed anyway: a first `offline` declared it, and the `up` after it
passed as an agent already declared on the hive. `paused` alone sufficed,
since the hive deploys an absent agent declared paused. A first
declaration in any placing state now runs the name rules and the
placed-elsewhere check; an agent already declared on the hive skips both.

A rule-breaking name whose roster read fails, including when no identity
bridge is configured, was accepted with a warning nobody sees, as in
create_agent. No later step on this route refuses the name, so it now
refuses with 503.

The gate is taken only for a placing declaration, so a destroy and its
credential revocation no longer wait on creations.

Refs #4804
This commit is contained in:
atlas 2026-09-29 18:19:42 +02:00 • committed by mara
commit 544a8dd228
1 changed files with 179 additions and 132 deletions
Download patch file Download diff file Expand all files Collapse all files

311
swarm-controller/src/main.rs
View file

@ -1037,7 +1037,7 @@ fn declaration_target(
(status = 200, description = "the declaration as now published", body = Vec<AgentDeclaration>), (status = 200, description = "the declaration as now published", body = Vec<AgentDeclaration>),
(status = 400, description = "a name is not an identifier, the hive is not in this swarm, the state is unknown, or `agent` is new and reserved or refused by the forge (problem+json)", body = String), (status = 400, description = "a name is not an identifier, the hive is not in this swarm, the state is unknown, or `agent` is new and reserved or refused by the forge (problem+json)", body = String),
(status = 409, description = "the swarm has already placed `agent` on a different hive (problem+json)", body = String), (status = 409, description = "the swarm has already placed `agent` on a different hive (problem+json)", body = String),
(status = 503, description = "no swarm queue is wired up, or it is not connected (problem+json)", body = String), (status = 503, description = "no swarm queue is wired up, or it is not connected, or `agent` is new here and breaks a naming rule but whether it already exists could not be checked (problem+json)", body = String),
(status = 500, description = "the declaration could not be published, or another hive's wanted state could not be read (problem+json)", body = String), (status = 500, description = "the declaration could not be published, or another hive's wanted state could not be read (problem+json)", body = String),
), ),
tag = "agents" tag = "agents"
@ -1053,14 +1053,17 @@ async fn set_agent_state(
.map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))? .map_err(|reason| error_problem(axum::http::StatusCode::BAD_REQUEST, reason))?
.into_string(); .into_string();
// Only a first declaration of the name on this hive is checked, and only // Only a first declaration of the name on this hive is checked, against
// against what `create_agent` refuses; `placement_checks` says which // what `create_agent` refuses: an agent already declared here changes
// checks apply. An undeclared name that passes them is accepted: that is // state whatever its name. An undeclared name that passes is accepted:
// how the swarm adopts an agent that predates swarm-level creation. // that is how the swarm adopts an agent that predates swarm-level
// Deliberate, pending the operator's decision on whether a name the swarm // creation. Deliberate, pending the operator's decision on whether a name
// never placed should be refused here instead. // the swarm never placed should be refused here instead.
let _gate = state.create_gate.lock().await; //
if places_agent(req.state) { // The gate is held from the checks through the write. A destroy places
// nothing, so it is not checked and does not wait on creations.
let _gate = if places_agent(req.state) {
let gate = state.create_gate.lock().await;
let here = writer.view(&hive).await.map_err(|e| { let here = writer.view(&hive).await.map_err(|e| {
let detail = format!( let detail = format!(
"cannot tell whether {agent:?} is already declared on hive {hive:?}, so it was \ "cannot tell whether {agent:?} is already declared on hive {hive:?}, so it was \
@ -1071,14 +1074,17 @@ async fn set_agent_state(
refuse_placement( refuse_placement(
&agent, &agent,
&hive, &hive,
placement_checks(&agent, req.state, here.as_ref()), here.as_ref(),
hive_types::reserved_names_raw().as_deref(), hive_types::reserved_names_raw().as_deref(),
|| in_roster(state.auth.as_deref(), &agent), || in_roster(state.auth.as_deref(), &agent),
|| queued_placements_now(&state), || queued_placements_now(&state),
|| declarations_elsewhere(&state, &agent, &hive), || declarations_elsewhere(&state, &agent, &hive),
) )
.await?; .await?;
} Some(gate)
} else {
None
};
let declaration = writer.set(&hive, &agent, req.state).await.map_err(|e| { let declaration = writer.set(&hive, &agent, req.state).await.map_err(|e| {
tracing::warn!(hive = %hive, agent = %agent, error = %format!("{e:#}"), "declaring agent state failed"); tracing::warn!(hive = %hive, agent = %agent, error = %format!("{e:#}"), "declaring agent state failed");
@ -1108,49 +1114,19 @@ fn places_agent(state: swarm_queue_client::wanted::AgentState) -> bool {
} }
} }
/// Which of [`refuse_placement`]'s checks a declaration must pass. /// Refuse a first placement of `agent` on `hive` that `create_agent`
#[derive(Debug, Clone, Copy, PartialEq, Eq)] /// refuses: 400 for a new name that breaks a naming rule, 409 for a name the
enum PlacementChecks { /// swarm has placed on another hive. An agent `here`, the hive's current
Neither, /// declaration, already declares in a placing state is placed on this hive,
Elsewhere, /// and passes.
NameAndElsewhere,
}
/// The checks declaring `agent` into `state` must pass, given `here`, the
/// hive's current declaration.
/// ///
/// An agent `here` already declares in a placing state is placed on this /// The roster is read only for a name that breaks a rule. Unlike in
/// hive, so it passes both. Otherwise the name rules apply to `Up` only: /// `create_agent`, no later step refuses a bad name, so a roster that cannot
/// `Offline` and `Paused` are how an operator stops an agent, and that is /// be read refuses (503) rather than warns.
/// never refused over its name.
fn placement_checks(
agent: &str,
state: swarm_queue_client::wanted::AgentState,
here: Option<&swarm_queue_client::wanted::HiveWanted>,
) -> PlacementChecks {
use swarm_queue_client::wanted::AgentState;
let declared_here = here
.and_then(|declaration| declaration.agents.get(agent))
.is_some_and(|wanted| places_agent(wanted.state));
match state {
_ if declared_here => PlacementChecks::Neither,
AgentState::Up => PlacementChecks::NameAndElsewhere,
AgentState::Offline | AgentState::Paused => PlacementChecks::Elsewhere,
AgentState::Destroyed => PlacementChecks::Neither,
}
}
/// Refuse a placement of `agent` on `hive` that `create_agent` refuses: 400
/// for a new name that breaks a naming rule, 409 for a name the swarm has
/// placed on another hive. `checks` says which of the two to run.
///
/// The roster is read only for a name that breaks a rule, as in
/// `create_agent`, and an unreadable one warns rather than refuses (see
/// [`name_verdict`]).
async fn refuse_placement<R, D>( async fn refuse_placement<R, D>(
agent: &str, agent: &str,
hive: &str, hive: &str,
checks: PlacementChecks, here: Option<&swarm_queue_client::wanted::HiveWanted>,
reserved: Option<&str>, reserved: Option<&str>,
read_roster: impl FnOnce() -> R, read_roster: impl FnOnce() -> R,
snapshot_queued: impl FnOnce() -> Vec<(String, String)>, snapshot_queued: impl FnOnce() -> Vec<(String, String)>,
@ -1165,18 +1141,27 @@ where
>, >,
>, >,
{ {
if checks == PlacementChecks::Neither { let declared_here = here
.and_then(|declaration| declaration.agents.get(agent))
.is_some_and(|wanted| places_agent(wanted.state));
if declared_here {
return Ok(()); return Ok(());
} }
if checks == PlacementChecks::NameAndElsewhere { // Both calls log what they push here; this response has nowhere to
// Both calls log what they push here; this response has nowhere to // carry warnings.
// carry warnings. let mut warnings = Vec::new();
let mut warnings = Vec::new(); let broken = broken_name_rules(agent, reserved, &mut warnings);
let broken = broken_name_rules(agent, reserved, &mut warnings); if !broken.is_empty() {
if !broken.is_empty() { let existing = read_roster().await.map_err(|why| {
name_verdict(agent, &broken, read_roster().await, &mut warnings) let detail = format!(
.map_err(|detail| error_problem(axum::http::StatusCode::BAD_REQUEST, &detail))?; "{}; whether {agent:?} is an existing agent, which would keep its name, could \
} not be checked, so it was not placed: {why}",
broken.join("; ")
);
error_problem(axum::http::StatusCode::SERVICE_UNAVAILABLE, &detail)
})?;
name_verdict(agent, &broken, Ok(existing), &mut warnings)
.map_err(|detail| error_problem(axum::http::StatusCode::BAD_REQUEST, &detail))?;
} }
let elsewhere = placements_elsewhere(agent, hive, snapshot_queued, read_declared).await?; let elsewhere = placements_elsewhere(agent, hive, snapshot_queued, read_declared).await?;
if !elsewhere.is_empty() { if !elsewhere.is_empty() {
@ -3298,13 +3283,13 @@ mod tests {
assert!(super::placed_elsewhere("atlas", "pr1ma", &declared, &queued).is_empty()); assert!(super::placed_elsewhere("atlas", "pr1ma", &declared, &queued).is_empty());
} }
/// The handler's two calls for declaring `agent` into `state` on `pr1ma`, /// `refuse_placement` of `agent` on `pr1ma`, with `admin` reserved,
/// with `admin` reserved, `agent` not in the roster, and `declared` as /// `roster` as the roster read, and `declared` as every hive's wanted
/// every hive's wanted state, `pr1ma`'s included. /// state, `pr1ma`'s included.
async fn refuse_on_pr1ma( async fn refuse_on_pr1ma_with(
agent: &str, agent: &str,
state: swarm_queue_client::wanted::AgentState,
declared: Vec<(String, swarm_queue_client::wanted::HiveWanted)>, declared: Vec<(String, swarm_queue_client::wanted::HiveWanted)>,
roster: Result<bool, String>,
) -> Result<(), problem_details::ProblemDetails> { ) -> Result<(), problem_details::ProblemDetails> {
let here = declared let here = declared
.iter() .iter()
@ -3313,30 +3298,46 @@ mod tests {
super::refuse_placement( super::refuse_placement(
agent, agent,
"pr1ma", "pr1ma",
super::placement_checks(agent, state, here.as_ref()), here.as_ref(),
Some("admin"), Some("admin"),
|| std::future::ready(Ok(false)), || std::future::ready(roster),
Vec::new, Vec::new,
|| std::future::ready(Ok(declared)), || std::future::ready(Ok(declared)),
) )
.await .await
} }
/// [`refuse_on_pr1ma_with`] for an `agent` the roster does not hold.
async fn refuse_on_pr1ma(
agent: &str,
declared: Vec<(String, swarm_queue_client::wanted::HiveWanted)>,
) -> Result<(), problem_details::ProblemDetails> {
refuse_on_pr1ma_with(agent, declared, Ok(false)).await
}
const PLACING: [swarm_queue_client::wanted::AgentState; 3] = [ const PLACING: [swarm_queue_client::wanted::AgentState; 3] = [
swarm_queue_client::wanted::AgentState::Up, swarm_queue_client::wanted::AgentState::Up,
swarm_queue_client::wanted::AgentState::Offline, swarm_queue_client::wanted::AgentState::Offline,
swarm_queue_client::wanted::AgentState::Paused, swarm_queue_client::wanted::AgentState::Paused,
]; ];
/// The handler checks exactly the placing states; a destroy is never
/// checked.
#[test]
fn only_a_destroy_places_nothing() {
for state in PLACING {
assert!(super::places_agent(state), "{state:?}");
}
assert!(!super::places_agent(
swarm_queue_client::wanted::AgentState::Destroyed
));
}
#[tokio::test] #[tokio::test]
async fn placing_a_new_reserved_name_is_refused() { async fn placing_a_new_reserved_name_is_refused() {
let err = refuse_on_pr1ma( let err = refuse_on_pr1ma("admin", Vec::new())
"admin", .await
swarm_queue_client::wanted::AgentState::Up, .expect_err("a new reserved name must be refused");
Vec::new(),
)
.await
.expect_err("a new reserved name must be refused");
assert_eq!( assert_eq!(
err.status, err.status,
Some(axum::http::StatusCode::BAD_REQUEST), Some(axum::http::StatusCode::BAD_REQUEST),
@ -3344,6 +3345,49 @@ mod tests {
); );
} }
/// `offline` first, then `up`: the first step is a first declaration, so
/// it is refused and leaves nothing on the hive for `up` to pass as.
/// `refuse_placement` takes no state: every placing state reaches it the
/// same way, which `only_a_destroy_places_nothing` pins.
#[tokio::test]
async fn offline_first_cannot_carry_a_reserved_name_into_up() {
let err = refuse_on_pr1ma("admin", Vec::new())
.await
.expect_err("a first `offline` of a reserved name must be refused");
assert_eq!(
err.status,
Some(axum::http::StatusCode::BAD_REQUEST),
"{err:?}"
);
// Nothing was declared, so `up` is a first declaration as well.
let err = refuse_on_pr1ma("admin", Vec::new())
.await
.expect_err("`up` after a refused `offline` must be refused");
assert_eq!(
err.status,
Some(axum::http::StatusCode::BAD_REQUEST),
"{err:?}"
);
}
/// No later step refuses a bad name, so an unreadable roster refuses
/// rather than lets it through. An ordinary name never reads it.
#[tokio::test]
async fn an_unreadable_roster_refuses_a_rule_breaking_name() {
let unreadable = || Err("no identity bridge is configured".to_owned());
let err = refuse_on_pr1ma_with("admin", Vec::new(), unreadable())
.await
.expect_err("an unreadable roster must not let a reserved name through");
assert_eq!(
err.status,
Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
"{err:?}"
);
refuse_on_pr1ma_with("atlas", Vec::new(), unreadable())
.await
.expect("an ordinary name does not depend on the roster");
}
/// A destroyed agent is gone from its hive, so declaring it again is a /// A destroyed agent is gone from its hive, so declaring it again is a
/// first declaration and its name is checked. /// first declaration and its name is checked.
#[tokio::test] #[tokio::test]
@ -3351,7 +3395,6 @@ mod tests {
use swarm_queue_client::wanted::AgentState; use swarm_queue_client::wanted::AgentState;
let err = refuse_on_pr1ma( let err = refuse_on_pr1ma(
"admin", "admin",
AgentState::Up,
vec![declaring("pr1ma", "admin", AgentState::Destroyed)], vec![declaring("pr1ma", "admin", AgentState::Destroyed)],
) )
.await .await
@ -3363,82 +3406,60 @@ mod tests {
); );
} }
/// Stopping is never refused over a name, even on a first declaration. /// An agent already declared here keeps a rule-breaking name through any
#[tokio::test] /// state change, whether or not the roster holds it or can be read.
async fn stopping_a_new_rule_breaking_name_is_accepted() {
use swarm_queue_client::wanted::AgentState;
for state in [AgentState::Offline, AgentState::Paused] {
refuse_on_pr1ma("admin", state, Vec::new())
.await
.unwrap_or_else(|err| panic!("{state:?}: {err:?}"));
}
}
/// An agent already declared here keeps a rule-breaking name through
/// every state change, though the roster does not know it.
#[tokio::test] #[tokio::test]
async fn an_agent_declared_here_is_not_refused_over_its_name() { async fn an_agent_declared_here_is_not_refused_over_its_name() {
use swarm_queue_client::wanted::AgentState; for current in PLACING {
for state in PLACING { for roster in [Ok(false), Err("bridge down".to_owned())] {
refuse_on_pr1ma( refuse_on_pr1ma_with(
"admin", "admin",
state, vec![declaring("pr1ma", "admin", current)],
vec![declaring("pr1ma", "admin", AgentState::Up)], roster.clone(),
) )
.await .await
.unwrap_or_else(|err| panic!("{state:?}: {err:?}")); .unwrap_or_else(|err| panic!("{current:?}, {roster:?}: {err:?}"));
}
} }
} }
#[tokio::test] #[tokio::test]
async fn a_first_declaration_of_a_name_placed_on_another_hive_is_refused() { async fn a_first_declaration_of_a_name_placed_on_another_hive_is_refused() {
use swarm_queue_client::wanted::AgentState; use swarm_queue_client::wanted::AgentState;
for state in PLACING { let err = refuse_on_pr1ma("atlas", vec![declaring("sec0nd", "atlas", AgentState::Up)])
let err = refuse_on_pr1ma(
"atlas",
state,
vec![declaring("sec0nd", "atlas", AgentState::Up)],
)
.await .await
.expect_err("a name placed on another hive must be refused"); .expect_err("a name placed on another hive must be refused");
assert_eq!( assert_eq!(
err.status, err.status,
Some(axum::http::StatusCode::CONFLICT), Some(axum::http::StatusCode::CONFLICT),
"{state:?}: {err:?}" "{err:?}"
); );
assert!(format!("{err:?}").contains("sec0nd"), "{err:?}"); assert!(format!("{err:?}").contains("sec0nd"), "{err:?}");
}
} }
/// An agent that runs on `pr1ma` but was never declared anywhere, and is /// An agent that runs on `pr1ma` but was never declared anywhere, and is
/// not in the roster, is one the swarm adopts by declaring it. /// not in the roster, is one the swarm adopts by declaring it.
#[tokio::test] #[tokio::test]
async fn adopting_an_undeclared_agent_on_its_own_hive_is_accepted() { async fn adopting_an_undeclared_agent_on_its_own_hive_is_accepted() {
for state in PLACING { refuse_on_pr1ma("legacy", Vec::new())
refuse_on_pr1ma("legacy", state, Vec::new()) .await
.await .expect("an undeclared agent must still be adoptable");
.unwrap_or_else(|err| panic!("{state:?}: {err:?}"));
}
} }
#[tokio::test] #[tokio::test]
async fn changing_the_state_of_an_agent_on_its_own_hive_is_accepted() { async fn changing_the_state_of_an_agent_on_its_own_hive_is_accepted() {
use swarm_queue_client::wanted::AgentState; use swarm_queue_client::wanted::AgentState;
for state in PLACING { refuse_on_pr1ma("atlas", vec![declaring("pr1ma", "atlas", AgentState::Up)])
refuse_on_pr1ma(
"atlas",
state,
vec![declaring("pr1ma", "atlas", AgentState::Up)],
)
.await .await
.unwrap_or_else(|err| panic!("{state:?}: {err:?}")); .expect("an agent's own hive must accept a new state for it");
}
} }
/// Through the handler: a wanted state that cannot be read refuses the /// `set_agent_state` on `pr1ma`, in a two-hive swarm whose queue never
/// declaration before it is written, as it refuses creation. /// connected and with no identity bridge.
#[tokio::test] async fn set_on_disconnected_pr1ma(
async fn set_agent_state_refuses_when_the_wanted_state_is_unreadable() { agent: &str,
wanted_state: swarm_queue_client::wanted::AgentState,
) -> problem_details::ProblemDetails {
let (state, _sched) = state_with_two_hives(); let (state, _sched) = state_with_two_hives();
let state = super::AppState { let state = super::AppState {
wanted: Some(std::sync::Arc::new(wanted::WantedWriter::new( wanted: Some(std::sync::Arc::new(wanted::WantedWriter::new(
@ -3446,16 +3467,23 @@ mod tests {
))), ))),
..state ..state
}; };
super::set_agent_state(
let err = super::set_agent_state(
axum::extract::State(state), axum::extract::State(state),
axum::extract::Path(("pr1ma".to_owned(), "atlas".to_owned())), axum::extract::Path(("pr1ma".to_owned(), agent.to_owned())),
axum::Json(SetAgentStateRequest { axum::Json(SetAgentStateRequest {
state: swarm_queue_client::wanted::AgentState::Up, state: wanted_state,
}), }),
) )
.await .await
.expect_err("an unreadable placement must refuse"); .expect_err("a queue that never connected cannot publish")
}
/// Through the handler: a wanted state that cannot be read refuses the
/// declaration before it is written, as it refuses creation.
#[tokio::test]
async fn set_agent_state_refuses_when_the_wanted_state_is_unreadable() {
let err =
set_on_disconnected_pr1ma("atlas", swarm_queue_client::wanted::AgentState::Up).await;
assert_eq!( assert_eq!(
err.status, err.status,
Some(axum::http::StatusCode::SERVICE_UNAVAILABLE), Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
@ -3467,6 +3495,25 @@ mod tests {
); );
} }
/// Through the handler: a destroy of a reserved name reaches the write
/// without being checked. It fails there only because the queue never
/// connected; the placing control above fails in the check instead.
#[tokio::test]
async fn set_agent_state_does_not_check_a_destroy() {
let err =
set_on_disconnected_pr1ma("admin", swarm_queue_client::wanted::AgentState::Destroyed)
.await;
assert_eq!(
err.status,
Some(axum::http::StatusCode::SERVICE_UNAVAILABLE),
"{err:?}"
);
assert!(
!format!("{err:?}").contains("not placed"),
"a destroy must not be refused by the placement check, got: {err:?}"
);
}
/// Forgejo v16.0.5's reserved usernames our charset can spell, plus the /// Forgejo v16.0.5's reserved usernames our charset can spell, plus the
/// bare `-` — the forge section of `nix/reserved-names.nix`. Read from /// bare `-` — the forge section of `nix/reserved-names.nix`. Read from
/// the list nix exports, so dropping one from the file reds this. /// the list nix exports, so dropping one from the file reds this.