swarm-archive: back up every agent's state to an append-only restic repository per agent
swarm-archive serves restic repositories with rclone, append-only, one login and one repository per agent. swarm-controller writes each placed agent's login and restic password to the store, replaces them on a move and removes them when the agent is destroyed. Every hive backs its agents up hourly under its own store identity. The archive host prunes by retention.
This commit is contained in:
atlas
parent
409e7ee5a0
commit
306aae0378
4 changed files with 2214 additions and 5 deletions
|
|
@ -68,6 +68,8 @@ opentelemetry-otlp.workspace = true
|
|||
# is departing, not arriving — see that module's docs.
|
||||
hmac.workspace = true
|
||||
sha2.workspace = true
|
||||
# The `{SHA}` htpasswd line `archive_access.rs` writes for the archive server.
|
||||
sha1.workspace = true
|
||||
# `forge/pull_diff.rs` diffs a PR's changed files itself, at a context width
|
||||
# the forge's own `.diff` does not offer.
|
||||
similar.workspace = true
|
||||
|
|
|
|||
572
swarm-controller/src/archive_access.rs
Normal file
572
swarm-controller/src/archive_access.rs
Normal file
|
|
@ -0,0 +1,572 @@
|
|||
//! Each placed agent's archive credentials, and their revocation.
|
||||
//!
|
||||
//! The archive (`swarm-archive.nix`) keeps one restic repository per agent and
|
||||
//! lets a user reach only its own. Every [`RECONCILE_INTERVAL`] this makes the
|
||||
//! store hold, per `(hive, agent)` that [`crate::jobq::placements`] reports:
|
||||
//!
|
||||
//! - the REST password and the restic password at
|
||||
//! [`archive::grant_path`], which that hive's store identity reads;
|
||||
//! - the htpasswd line, a hash of the REST password, at [`archive::user_path`],
|
||||
//! which the archive host renders its users file from;
|
||||
//! - the restic password at [`archive::restic_path`], created once and never
|
||||
//! replaced: it opens the agent's repository wherever the agent runs, and the
|
||||
//! archive host prunes with it.
|
||||
//!
|
||||
//! A pass is level-triggered and idempotent. A move to another hive mints a new
|
||||
//! REST password, which replaces the htpasswd line, so the old hive's password
|
||||
//! stops working once the archive host re-renders; the old hive's copy is
|
||||
//! deleted in the same pass. An agent no hive places loses its htpasswd line and
|
||||
//! every hive's copy. Its restic password stays, so what was backed up can still
|
||||
//! be read.
|
||||
//!
|
||||
//! An agent that two hives place is left as it is: either grant would replace
|
||||
//! the other's every pass.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use base64::Engine as _;
|
||||
use sha1::{Digest as _, Sha1};
|
||||
use swarm_secret_client::SecretStore;
|
||||
use swarm_secret_client::archive::{self, HTPASSWD, PASSWORD, REST_PASSWORD, RESTIC_PASSWORD};
|
||||
|
||||
use crate::wanted::WantedWriter;
|
||||
|
||||
/// How often [`spawn`] reconciles.
|
||||
const RECONCILE_INTERVAL: Duration = Duration::from_mins(5);
|
||||
|
||||
/// A pass that has not finished by now is given up on; the next tick retries.
|
||||
const PASS_DEADLINE: Duration = Duration::from_mins(2);
|
||||
|
||||
/// An object in the store: field name to value.
|
||||
type Fields = BTreeMap<String, String>;
|
||||
|
||||
/// The store calls a pass makes, so a test can stand in for the store.
|
||||
pub(crate) trait ArchiveStore {
|
||||
/// The object at `path`; `None` when nothing is stored there.
|
||||
async fn read(&self, path: &str) -> Result<Option<Fields>>;
|
||||
/// Store `fields` at `path`, replacing what is there.
|
||||
async fn write(&self, path: &str, fields: &Fields) -> Result<()>;
|
||||
/// Store `fields` at `path` only while nothing is there.
|
||||
async fn create(&self, path: &str, fields: &Fields) -> Result<()>;
|
||||
/// The names stored directly under `path`.
|
||||
async fn list(&self, path: &str) -> Result<Vec<String>>;
|
||||
/// Remove `path` with every version of it; nothing there is not an error.
|
||||
async fn delete(&self, path: &str) -> Result<()>;
|
||||
}
|
||||
|
||||
impl ArchiveStore for SecretStore {
|
||||
async fn read(&self, path: &str) -> Result<Option<Fields>> {
|
||||
Ok(self.read_optional(path).await?)
|
||||
}
|
||||
|
||||
async fn write(&self, path: &str, fields: &Fields) -> Result<()> {
|
||||
Ok(SecretStore::write(self, path, fields).await?)
|
||||
}
|
||||
|
||||
async fn create(&self, path: &str, fields: &Fields) -> Result<()> {
|
||||
Ok(SecretStore::create(self, path, fields).await?)
|
||||
}
|
||||
|
||||
async fn list(&self, path: &str) -> Result<Vec<String>> {
|
||||
Ok(SecretStore::list(self, path).await?)
|
||||
}
|
||||
|
||||
async fn delete(&self, path: &str) -> Result<()> {
|
||||
Ok(self.delete_all_versions(path).await?)
|
||||
}
|
||||
}
|
||||
|
||||
/// What one pass did.
|
||||
#[derive(Debug, Default, PartialEq, Eq)]
|
||||
pub(crate) struct Done {
|
||||
/// Agents whose credentials were written or replaced.
|
||||
pub granted: usize,
|
||||
/// Credentials removed: a hive's copy or an htpasswd line.
|
||||
pub revoked: usize,
|
||||
/// Agents or hives whose credentials could not all be checked or written.
|
||||
pub failed: usize,
|
||||
}
|
||||
|
||||
fn fields(pairs: &[(&str, &str)]) -> Fields {
|
||||
pairs
|
||||
.iter()
|
||||
.map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The non-empty field `name` of `object`.
|
||||
fn field<'a>(object: Option<&'a Fields>, name: &str) -> Option<&'a str> {
|
||||
object
|
||||
.and_then(|o| o.get(name))
|
||||
.map(String::as_str)
|
||||
.filter(|v| !v.is_empty())
|
||||
}
|
||||
|
||||
/// The htpasswd line that accepts `password` for `agent`: apache's `{SHA}`
|
||||
/// form, which the password's 256 random bits make as strong as a slower hash.
|
||||
fn htpasswd_line(agent: &str, password: &str) -> String {
|
||||
let digest = Sha1::digest(password.as_bytes());
|
||||
format!(
|
||||
"{agent}:{{SHA}}{}",
|
||||
base64::engine::general_purpose::STANDARD.encode(digest)
|
||||
)
|
||||
}
|
||||
|
||||
/// `agent`'s restic password: the stored one, else a new one created at
|
||||
/// [`archive::restic_path`]. A create that fails is read back, in case another
|
||||
/// writer's landed first.
|
||||
async fn ensure_restic(store: &impl ArchiveStore, agent: &str) -> Result<String> {
|
||||
let path = archive::restic_path(agent)?;
|
||||
let stored = store
|
||||
.read(&path)
|
||||
.await
|
||||
.with_context(|| format!("reading {path}"))?;
|
||||
if let Some(password) = field(stored.as_ref(), PASSWORD) {
|
||||
return Ok(password.to_owned());
|
||||
}
|
||||
let minted = crate::agent_identity::generate_queue_secret()?;
|
||||
if let Err(e) = store.create(&path, &fields(&[(PASSWORD, &minted)])).await {
|
||||
let winner = store
|
||||
.read(&path)
|
||||
.await
|
||||
.with_context(|| format!("reading {path} back after a failed create"))?;
|
||||
return field(winner.as_ref(), PASSWORD)
|
||||
.map(str::to_owned)
|
||||
.ok_or_else(|| e.context(format!("creating {path}")));
|
||||
}
|
||||
Ok(minted)
|
||||
}
|
||||
|
||||
/// Bring `agent`'s credentials on `hive` and at the archive up to date; whether
|
||||
/// anything was written.
|
||||
async fn grant(store: &impl ArchiveStore, hive: &str, agent: &str) -> Result<bool> {
|
||||
let restic = ensure_restic(store, agent).await?;
|
||||
let grant_path = archive::grant_path(hive, agent)?;
|
||||
let held = store
|
||||
.read(&grant_path)
|
||||
.await
|
||||
.with_context(|| format!("reading {grant_path}"))?;
|
||||
|
||||
let mut written = false;
|
||||
let rest = match field(held.as_ref(), REST_PASSWORD) {
|
||||
Some(rest) if field(held.as_ref(), RESTIC_PASSWORD) == Some(restic.as_str()) => {
|
||||
rest.to_owned()
|
||||
}
|
||||
_ => {
|
||||
let rest = crate::agent_identity::generate_queue_secret()?;
|
||||
store
|
||||
.write(
|
||||
&grant_path,
|
||||
&fields(&[(REST_PASSWORD, &rest), (RESTIC_PASSWORD, &restic)]),
|
||||
)
|
||||
.await
|
||||
.with_context(|| format!("writing {grant_path}"))?;
|
||||
written = true;
|
||||
rest
|
||||
}
|
||||
};
|
||||
|
||||
let user_path = archive::user_path(agent)?;
|
||||
let line = htpasswd_line(agent, &rest);
|
||||
let user = store
|
||||
.read(&user_path)
|
||||
.await
|
||||
.with_context(|| format!("reading {user_path}"))?;
|
||||
if field(user.as_ref(), HTPASSWD) != Some(line.as_str()) {
|
||||
store
|
||||
.write(&user_path, &fields(&[(HTPASSWD, &line)]))
|
||||
.await
|
||||
.with_context(|| format!("writing {user_path}"))?;
|
||||
written = true;
|
||||
}
|
||||
Ok(written)
|
||||
}
|
||||
|
||||
/// The entries under `prefix` that are single leaves.
|
||||
async fn leaves(store: &impl ArchiveStore, prefix: &str) -> Result<Vec<String>> {
|
||||
Ok(store
|
||||
.list(prefix)
|
||||
.await
|
||||
.with_context(|| format!("listing {prefix}"))?
|
||||
.into_iter()
|
||||
.filter(|name| !name.ends_with('/'))
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// One pass: grant every agent in `placed`, then revoke what no placement
|
||||
/// justifies. `hives` is every hive of the swarm, whose copies are checked.
|
||||
///
|
||||
/// # Errors
|
||||
/// Never: a failure is counted in [`Done::failed`] and logged, and the rest of
|
||||
/// the pass still runs.
|
||||
pub(crate) async fn reconcile(
|
||||
store: &impl ArchiveStore,
|
||||
placed: &BTreeSet<(String, String)>,
|
||||
hives: &[String],
|
||||
) -> Done {
|
||||
let mut done = Done::default();
|
||||
let mut holders: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
|
||||
for (hive, agent) in placed {
|
||||
holders.entry(agent).or_default().push(hive);
|
||||
}
|
||||
|
||||
let mut ambiguous = BTreeSet::new();
|
||||
for (agent, on) in &holders {
|
||||
if let [hive] = on.as_slice() {
|
||||
match grant(store, hive, agent).await {
|
||||
Ok(true) => done.granted += 1,
|
||||
Ok(false) => {}
|
||||
Err(e) => {
|
||||
tracing::warn!(hive, agent, error = %format!("{e:#}"), "archive: granting failed");
|
||||
done.failed += 1;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
tracing::warn!(agent, hives = ?on, "archive: placed on more than one hive; credentials left as they are");
|
||||
ambiguous.insert(*agent);
|
||||
}
|
||||
}
|
||||
|
||||
for hive in hives {
|
||||
let copies = match archive::grants_prefix(hive) {
|
||||
Ok(prefix) => leaves(store, &prefix).await,
|
||||
Err(e) => Err(e.into()),
|
||||
};
|
||||
let copies = match copies {
|
||||
Ok(copies) => copies,
|
||||
Err(e) => {
|
||||
tracing::warn!(hive, error = %format!("{e:#}"), "archive: listing the hive's copies failed");
|
||||
done.failed += 1;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
for agent in copies {
|
||||
let keep = ambiguous.contains(agent.as_str())
|
||||
|| placed.contains(&(hive.clone(), agent.clone()));
|
||||
if !keep {
|
||||
revoke(store, archive::grant_path(hive, &agent), &mut done).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
match leaves(store, &archive::users_prefix().unwrap_or_default()).await {
|
||||
Ok(users) => {
|
||||
for agent in users {
|
||||
if !holders.contains_key(agent.as_str()) {
|
||||
revoke(store, archive::user_path(&agent), &mut done).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %format!("{e:#}"), "archive: listing the users failed");
|
||||
done.failed += 1;
|
||||
}
|
||||
}
|
||||
done
|
||||
}
|
||||
|
||||
/// Delete the credential at `path`, counting the outcome in `done`.
|
||||
async fn revoke(
|
||||
store: &impl ArchiveStore,
|
||||
path: Result<String, swarm_secret_client::Error>,
|
||||
done: &mut Done,
|
||||
) {
|
||||
let path = match path {
|
||||
Ok(path) => path,
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %e, "archive: a stored name is not a valid path segment");
|
||||
done.failed += 1;
|
||||
return;
|
||||
}
|
||||
};
|
||||
match store.delete(&path).await {
|
||||
Ok(()) => {
|
||||
tracing::info!(%path, "archive: credential revoked");
|
||||
done.revoked += 1;
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(%path, error = %format!("{e:#}"), "archive: revoking failed");
|
||||
done.failed += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Reconcile now and every [`RECONCILE_INTERVAL`] after. A pass that fails is
|
||||
/// logged and retried on the next tick; it never stops the daemon. Nothing is
|
||||
/// revoked on a pass that could not read where the hives place agents.
|
||||
pub fn spawn(wanted: Arc<WantedWriter>, hives: Vec<String>) {
|
||||
tokio::spawn(async move {
|
||||
let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
|
||||
loop {
|
||||
ticker.tick().await;
|
||||
let pass = async {
|
||||
let placed = crate::jobq::placements(&wanted, &hives).await?;
|
||||
let store = crate::store::connect()
|
||||
.await
|
||||
.context("logging in to the swarm secret store")?;
|
||||
anyhow::Ok(reconcile(&store, &placed, &hives).await)
|
||||
};
|
||||
match tokio::time::timeout(PASS_DEADLINE, pass).await {
|
||||
Ok(Ok(done)) if done != Done::default() => tracing::info!(
|
||||
granted = done.granted,
|
||||
revoked = done.revoked,
|
||||
failed = done.failed,
|
||||
"archive: credentials reconciled"
|
||||
),
|
||||
Ok(Ok(_)) => tracing::debug!("archive: every credential is current"),
|
||||
Ok(Err(e)) => tracing::warn!(
|
||||
error = %format!("{e:#}"),
|
||||
retry_in_s = RECONCILE_INTERVAL.as_secs(),
|
||||
"archive: pass failed; retrying next tick"
|
||||
),
|
||||
Err(_) => tracing::warn!(
|
||||
deadline_s = PASS_DEADLINE.as_secs(),
|
||||
"archive: pass timed out; retrying next tick"
|
||||
),
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::sync::Mutex;
|
||||
|
||||
use super::*;
|
||||
|
||||
/// An in-memory store. `create` refuses a held path, `list` answers like
|
||||
/// the store's: the names directly under a prefix.
|
||||
#[derive(Default)]
|
||||
struct Memory {
|
||||
kv: Mutex<BTreeMap<String, Fields>>,
|
||||
/// Paths under this prefix refuse every write and delete.
|
||||
refuse: Option<&'static str>,
|
||||
}
|
||||
|
||||
impl Memory {
|
||||
fn get(&self, path: &str) -> Option<Fields> {
|
||||
self.kv.lock().unwrap().get(path).cloned()
|
||||
}
|
||||
|
||||
fn value(&self, path: &str, name: &str) -> Option<String> {
|
||||
self.get(path).and_then(|f| f.get(name).cloned())
|
||||
}
|
||||
|
||||
fn refused(&self, path: &str) -> bool {
|
||||
self.refuse.is_some_and(|p| path.starts_with(p))
|
||||
}
|
||||
}
|
||||
|
||||
impl ArchiveStore for Memory {
|
||||
async fn read(&self, path: &str) -> Result<Option<Fields>> {
|
||||
Ok(self.get(path))
|
||||
}
|
||||
|
||||
async fn write(&self, path: &str, fields: &Fields) -> Result<()> {
|
||||
if self.refused(path) {
|
||||
anyhow::bail!("denied");
|
||||
}
|
||||
self.kv
|
||||
.lock()
|
||||
.unwrap()
|
||||
.insert(path.to_owned(), fields.clone());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn create(&self, path: &str, fields: &Fields) -> Result<()> {
|
||||
if self.get(path).is_some() {
|
||||
anyhow::bail!("exists");
|
||||
}
|
||||
self.write(path, fields).await
|
||||
}
|
||||
|
||||
async fn list(&self, path: &str) -> Result<Vec<String>> {
|
||||
let prefix = format!("{path}/");
|
||||
Ok(self
|
||||
.kv
|
||||
.lock()
|
||||
.unwrap()
|
||||
.keys()
|
||||
.filter_map(|k| k.strip_prefix(&prefix))
|
||||
.map(|rest| match rest.split_once('/') {
|
||||
Some((dir, _)) => format!("{dir}/"),
|
||||
None => rest.to_owned(),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
async fn delete(&self, path: &str) -> Result<()> {
|
||||
if self.refused(path) {
|
||||
anyhow::bail!("denied");
|
||||
}
|
||||
self.kv.lock().unwrap().remove(path);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn placed(pairs: &[(&str, &str)]) -> BTreeSet<(String, String)> {
|
||||
pairs
|
||||
.iter()
|
||||
.map(|(h, a)| ((*h).to_owned(), (*a).to_owned()))
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn hives() -> Vec<String> {
|
||||
vec!["alpha".to_owned(), "beta".to_owned()]
|
||||
}
|
||||
|
||||
const ALPHA_ATLAS: &str = "swarm/hives/alpha/archive/atlas";
|
||||
const BETA_ATLAS: &str = "swarm/hives/beta/archive/atlas";
|
||||
const USER_ATLAS: &str = "swarm/services/archive/users/atlas";
|
||||
const RESTIC_ATLAS: &str = "swarm/services/archive/restic/atlas";
|
||||
|
||||
#[test]
|
||||
fn the_htpasswd_line_is_apaches_sha_form() {
|
||||
// `htpasswd -nbs atlas secret`
|
||||
assert_eq!(
|
||||
htpasswd_line("atlas", "secret"),
|
||||
"atlas:{SHA}5en6G6MezRroT3XKqkdPOmY/BfQ="
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_placed_agent_gets_its_hive_a_password_and_the_archive_its_hash() {
|
||||
let store = Memory::default();
|
||||
let done = reconcile(&store, &placed(&[("alpha", "atlas")]), &hives()).await;
|
||||
assert_eq!(
|
||||
done,
|
||||
Done {
|
||||
granted: 1,
|
||||
revoked: 0,
|
||||
failed: 0
|
||||
}
|
||||
);
|
||||
let rest = store.value(ALPHA_ATLAS, REST_PASSWORD).expect("a password");
|
||||
let restic = store.value(ALPHA_ATLAS, RESTIC_PASSWORD).expect("restic");
|
||||
assert_eq!(
|
||||
store.value(USER_ATLAS, HTPASSWD),
|
||||
Some(htpasswd_line("atlas", &rest))
|
||||
);
|
||||
assert_eq!(store.value(RESTIC_ATLAS, PASSWORD), Some(restic));
|
||||
assert_ne!(Some(rest), store.value(ALPHA_ATLAS, RESTIC_PASSWORD));
|
||||
// The hash line is all the archive's user leaf holds.
|
||||
assert!(!store.get(USER_ATLAS).unwrap().contains_key(REST_PASSWORD));
|
||||
// Presence control for the absence checks below: only the placed hive.
|
||||
assert!(store.get(BETA_ATLAS).is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_current_grant_is_not_rewritten() {
|
||||
let store = Memory::default();
|
||||
let one = placed(&[("alpha", "atlas")]);
|
||||
reconcile(&store, &one, &hives()).await;
|
||||
let before = store.get(ALPHA_ATLAS);
|
||||
let done = reconcile(&store, &one, &hives()).await;
|
||||
assert_eq!(done, Done::default());
|
||||
assert_eq!(store.get(ALPHA_ATLAS), before);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_move_replaces_the_password_and_takes_the_old_hives_copy() {
|
||||
let store = Memory::default();
|
||||
reconcile(&store, &placed(&[("alpha", "atlas")]), &hives()).await;
|
||||
let old_rest = store.value(ALPHA_ATLAS, REST_PASSWORD).unwrap();
|
||||
let restic = store.value(RESTIC_ATLAS, PASSWORD).unwrap();
|
||||
|
||||
let done = reconcile(&store, &placed(&[("beta", "atlas")]), &hives()).await;
|
||||
assert_eq!(done.granted, 1);
|
||||
assert_eq!(done.revoked, 1);
|
||||
assert!(store.get(ALPHA_ATLAS).is_none(), "the old hive loses read");
|
||||
let new_rest = store.value(BETA_ATLAS, REST_PASSWORD).expect("new hive");
|
||||
assert_ne!(new_rest, old_rest, "the old password stops working");
|
||||
assert_eq!(
|
||||
store.value(USER_ATLAS, HTPASSWD),
|
||||
Some(htpasswd_line("atlas", &new_rest))
|
||||
);
|
||||
assert_ne!(
|
||||
store.value(USER_ATLAS, HTPASSWD),
|
||||
Some(htpasswd_line("atlas", &old_rest))
|
||||
);
|
||||
assert_eq!(
|
||||
store.value(BETA_ATLAS, RESTIC_PASSWORD).as_deref(),
|
||||
Some(restic.as_str()),
|
||||
"the repository stays readable on the new hive"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_destroyed_agent_loses_every_credential_but_the_restic_password() {
|
||||
let store = Memory::default();
|
||||
reconcile(
|
||||
&store,
|
||||
&placed(&[("alpha", "atlas"), ("beta", "argus")]),
|
||||
&hives(),
|
||||
)
|
||||
.await;
|
||||
let done = reconcile(&store, &placed(&[("beta", "argus")]), &hives()).await;
|
||||
assert_eq!(done.revoked, 2);
|
||||
assert!(store.get(ALPHA_ATLAS).is_none());
|
||||
assert!(store.get(USER_ATLAS).is_none());
|
||||
assert!(
|
||||
store.get(RESTIC_ATLAS).is_some(),
|
||||
"the backups stay readable"
|
||||
);
|
||||
// Control: the agent still placed keeps all of it.
|
||||
assert!(store.get("swarm/hives/beta/archive/argus").is_some());
|
||||
assert!(store.get("swarm/services/archive/users/argus").is_some());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_agent_two_hives_place_is_left_as_it_is() {
|
||||
let store = Memory::default();
|
||||
reconcile(&store, &placed(&[("alpha", "atlas")]), &hives()).await;
|
||||
let before = (store.get(ALPHA_ATLAS), store.get(USER_ATLAS));
|
||||
let both = placed(&[("alpha", "atlas"), ("beta", "atlas")]);
|
||||
let done = reconcile(&store, &both, &hives()).await;
|
||||
assert_eq!(done, Done::default());
|
||||
assert_eq!((store.get(ALPHA_ATLAS), store.get(USER_ATLAS)), before);
|
||||
assert!(store.get(BETA_ATLAS).is_none());
|
||||
|
||||
// Neither hive is granted when the agent is new and both place it.
|
||||
let fresh = Memory::default();
|
||||
let done = reconcile(&fresh, &both, &hives()).await;
|
||||
assert_eq!(done, Done::default());
|
||||
assert!(fresh.kv.lock().unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_lost_user_line_is_rewritten_from_the_hives_password() {
|
||||
let store = Memory::default();
|
||||
let one = placed(&[("alpha", "atlas")]);
|
||||
reconcile(&store, &one, &hives()).await;
|
||||
let line = store.value(USER_ATLAS, HTPASSWD);
|
||||
store.kv.lock().unwrap().remove(USER_ATLAS);
|
||||
let done = reconcile(&store, &one, &hives()).await;
|
||||
assert_eq!(done.granted, 1);
|
||||
assert_eq!(store.value(USER_ATLAS, HTPASSWD), line);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_refused_write_counts_as_failed_and_the_other_agents_still_get_theirs() {
|
||||
let store = Memory {
|
||||
refuse: Some("swarm/hives/alpha/"),
|
||||
..Memory::default()
|
||||
};
|
||||
let done = reconcile(
|
||||
&store,
|
||||
&placed(&[("alpha", "atlas"), ("beta", "argus")]),
|
||||
&hives(),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(done.failed, 1);
|
||||
assert_eq!(done.granted, 1);
|
||||
assert!(store.get("swarm/hives/beta/archive/argus").is_some());
|
||||
assert!(
|
||||
store.get(USER_ATLAS).is_none(),
|
||||
"no hash without the hive's copy"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -39,6 +39,7 @@ mod agents_md;
|
|||
mod agents_md_initial;
|
||||
mod alerts;
|
||||
mod archetypes;
|
||||
mod archive_access;
|
||||
mod attic_copy;
|
||||
mod auth;
|
||||
mod backend_egress;
|
||||
|
|
|
|||
|
|
@ -366,7 +366,8 @@ fn spawn_subagent_streams(status: Option<&Arc<status::StatusReader>>, hives: &[H
|
|||
/// schedule keys, and the permissions bucket and its import, the allowed
|
||||
/// domains derived from backend environments, and the
|
||||
/// archetypes agents are created from. Beside them, the
|
||||
/// copy of the cache facts to each hive, and the proxy credentials squid checks.
|
||||
/// copy of the cache facts to each hive, the archive credentials of each placed
|
||||
/// agent, and the proxy credentials squid checks.
|
||||
/// Lifted out of `main` for `clippy::too_many_lines`.
|
||||
pub(crate) fn spawn_per_agent_upkeep(
|
||||
status: Option<&Arc<status::StatusReader>>,
|
||||
|
|
@ -377,12 +378,22 @@ pub(crate) fn spawn_per_agent_upkeep(
|
|||
spawn_agent_permissions(status, hives);
|
||||
spawn_backend_egress_backfill(status, hives);
|
||||
crate::attic_copy::spawn(hives.iter().map(|h| h.name.clone()).collect());
|
||||
spawn_archive_access(status, hives);
|
||||
crate::proxy_credential::spawn(hives.iter().map(|h| h.name.clone()).collect());
|
||||
if let Some(status) = status {
|
||||
archetypes::spawn(status.queue_client());
|
||||
}
|
||||
}
|
||||
|
||||
/// Grant and revoke each placed agent's archive credentials when a swarm queue
|
||||
/// is wired up. Same shape as [`spawn_subagent_streams`].
|
||||
fn spawn_archive_access(status: Option<&Arc<status::StatusReader>>, hives: &[HiveEntry]) {
|
||||
let Some(wanted) = wanted_writer(status) else {
|
||||
return;
|
||||
};
|
||||
crate::archive_access::spawn(wanted, hives.iter().map(|h| h.name.clone()).collect());
|
||||
}
|
||||
|
||||
/// Create the agent-permissions bucket and answer hives' imports when a swarm
|
||||
/// queue is wired up. Same shape as [`spawn_subagent_streams`].
|
||||
fn spawn_agent_permissions(status: Option<&Arc<status::StatusReader>>, hives: &[HiveEntry]) {
|
||||
|
|
|
|||
Loading…
Reference in a new issue