derive an agent's backend hosts into the egress allow list at the controller
This commit is contained in:
parent
4f32948a97
commit
c5a876b845
8 changed files with 950 additions and 25 deletions
|
|
@ -67,7 +67,8 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients.
|
|||
`egress-allow` KV bucket, which this daemon alone writes
|
||||
(`GET`/`POST`/`DELETE /api/egress/allowed-domains`); every hive reads it and
|
||||
renders the proxy's allow-list files, and it replaces the list the hive's own
|
||||
configuration names →
|
||||
configuration names. It also keeps a `derived` entry per agent for the host
|
||||
of its stored backend environment's `ANTHROPIC_BASE_URL` →
|
||||
[`docs/networking/network.md`](../docs/networking/network.md#egress-proxy-squid).
|
||||
- **Agent instructions** — each agent's `AGENTS.md` as its harness last
|
||||
published it into the `agents-md` KV bucket
|
||||
|
|
|
|||
703
swarm-controller/src/backend_egress.rs
Normal file
703
swarm-controller/src/backend_egress.rs
Normal file
|
|
@ -0,0 +1,703 @@
|
|||
//! The hosts an agent's backend environment names, kept as derived entries of
|
||||
//! the egress allow list (`crate::egress_allow`), so a restricted agent whose
|
||||
//! `ANTHROPIC_BASE_URL` points at a provider other than Anthropic reaches that
|
||||
//! provider through its hive's proxy.
|
||||
//!
|
||||
//! The controller sees the environment where it is written (`backend_env`) and
|
||||
//! in the store, so it derives here and a hive renders the entries like any
|
||||
//! other. An entry is marked `derived`: [`follow_env`] replaces an agent's
|
||||
//! derived entries on every write or delete of its environment, and
|
||||
//! [`spawn_backfill`] reconciles all of them from the stored environments at
|
||||
//! start. Each host goes through `egress_allow::checked`, the check an
|
||||
//! operator's own entry gets.
|
||||
//!
|
||||
//! Only the store's copy is read. An agent still on the hand-placed
|
||||
//! `backendEnvironmentFile` alone has no derived entry. The opencode preset's
|
||||
//! provider URL (`acp.opencode.provider.baseUrl`) is host configuration, not
|
||||
//! part of the environment, so it is never derived either.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use problem_details::ProblemDetails;
|
||||
use swarm_secret_client::backend_env::{self, BackendEnv};
|
||||
use swarm_wire::egress_allow::{AllowList, Entry, MAX_ENTRIES};
|
||||
|
||||
use crate::egress_allow::{base_domains, checked, modify};
|
||||
use crate::linked_accounts::AccountStore;
|
||||
use crate::routes::{AppState, error_problem, wanted_error_status};
|
||||
use crate::schedules::queue;
|
||||
use crate::wanted::WantedWriter;
|
||||
|
||||
/// `ANTHROPIC_BASE_URL` and every variable named `<PROVIDER>_BASE_URL`, which
|
||||
/// is where a provider's address sits for an ACP agent that loads the same
|
||||
/// file.
|
||||
fn is_url_variable(name: &str) -> bool {
|
||||
name == "ANTHROPIC_BASE_URL" || name.ends_with("_BASE_URL")
|
||||
}
|
||||
|
||||
/// The URL variables `env` assigns, as `(name, value)`. A later assignment of a
|
||||
/// name replaces an earlier one, as systemd's `EnvironmentFile=` does.
|
||||
fn url_assignments(env: &str) -> BTreeMap<&str, &str> {
|
||||
let mut found = BTreeMap::new();
|
||||
for line in env.lines().map(str::trim) {
|
||||
if line.is_empty() || line.starts_with(['#', ';']) {
|
||||
continue;
|
||||
}
|
||||
let Some((name, value)) = line.split_once('=') else {
|
||||
continue;
|
||||
};
|
||||
let name = name.trim();
|
||||
if is_url_variable(name) {
|
||||
found.insert(name, unquoted(value.trim()));
|
||||
}
|
||||
}
|
||||
found
|
||||
}
|
||||
|
||||
fn unquoted(value: &str) -> &str {
|
||||
for quote in ['"', '\''] {
|
||||
if let Some(inner) = value
|
||||
.strip_prefix(quote)
|
||||
.and_then(|v| v.strip_suffix(quote))
|
||||
{
|
||||
return inner;
|
||||
}
|
||||
}
|
||||
value
|
||||
}
|
||||
|
||||
/// The host of `url`, as written: its scheme (`http` or `https`, or none),
|
||||
/// user information, port, path, query and fragment are dropped. Nothing is
|
||||
/// validated; the result goes through `egress_allow::checked`.
|
||||
fn host_of(url: &str) -> Option<&str> {
|
||||
let rest = match url.split_once("://") {
|
||||
Some(("http" | "https", rest)) => rest,
|
||||
Some(_) => return None,
|
||||
None => url,
|
||||
};
|
||||
let authority = rest.split(['/', '?', '#']).next()?;
|
||||
let authority = authority.rsplit('@').next()?;
|
||||
Some(match authority.rsplit_once(':') {
|
||||
Some((host, port)) if port.bytes().all(|b| b.is_ascii_digit()) => host,
|
||||
_ => authority,
|
||||
})
|
||||
}
|
||||
|
||||
/// The derived entries for `agent`'s `env`: one per distinct host that passes
|
||||
/// `egress_allow::checked`. A host that does not (an IP address, `localhost`,
|
||||
/// a name the base list already allows) is logged by variable name and left
|
||||
/// out. The value is never logged: a URL can carry credentials.
|
||||
fn derive(agent: &str, env: &str, base: &[String]) -> Vec<Entry> {
|
||||
let mut entries = BTreeSet::new();
|
||||
for (name, value) in url_assignments(env) {
|
||||
let Some(host) = host_of(value) else {
|
||||
tracing::info!(%agent, variable = name, "backend egress: not an http(s) URL; no entry");
|
||||
continue;
|
||||
};
|
||||
let request = swarm_controller_api::egress::AllowedDomain {
|
||||
domain: host.to_owned(),
|
||||
agent: Some(agent.to_owned()),
|
||||
derived: false,
|
||||
};
|
||||
match checked(&request, base) {
|
||||
Ok(entry) => {
|
||||
entries.insert(Entry {
|
||||
derived: true,
|
||||
..entry
|
||||
});
|
||||
}
|
||||
Err(why) => {
|
||||
tracing::info!(%agent, variable = name, %why, "backend egress: no entry for this host");
|
||||
}
|
||||
}
|
||||
}
|
||||
entries.into_iter().collect()
|
||||
}
|
||||
|
||||
/// `list` with, for each agent in `desired`, its derived entries replaced by
|
||||
/// the given ones. Entries of other agents and operator-added entries are
|
||||
/// untouched. A full list takes no more derived entries.
|
||||
fn reconciled(mut list: AllowList, desired: &BTreeMap<String, Vec<Entry>>) -> AllowList {
|
||||
list.entries
|
||||
.retain(|e| !(e.derived && e.agent.as_ref().is_some_and(|a| desired.contains_key(a))));
|
||||
for entry in desired.values().flatten() {
|
||||
if list.entries.len() >= MAX_ENTRIES {
|
||||
tracing::warn!(
|
||||
domain = %entry.domain,
|
||||
"backend egress: the allow list is full; no entry for this host"
|
||||
);
|
||||
break;
|
||||
}
|
||||
list.entries.push(entry.clone());
|
||||
}
|
||||
list
|
||||
}
|
||||
|
||||
/// Replace `agent`'s derived entries with those of `env`, or with none when
|
||||
/// `env` is `None` (the environment was deleted). A host without a queue
|
||||
/// (`state.status` unset) has no allow list, so nothing happens.
|
||||
///
|
||||
/// # Errors
|
||||
/// A problem naming that the environment is stored but the list was not
|
||||
/// updated; saving the environment again retries.
|
||||
pub(crate) async fn follow_env(
|
||||
state: &AppState,
|
||||
agent: &str,
|
||||
env: Option<&str>,
|
||||
) -> Result<(), ProblemDetails> {
|
||||
if state.status.is_none() {
|
||||
return Ok(());
|
||||
}
|
||||
let base = base_domains();
|
||||
let desired = BTreeMap::from([(
|
||||
agent.to_owned(),
|
||||
env.map(|text| derive(agent, text, &base))
|
||||
.unwrap_or_default(),
|
||||
)]);
|
||||
let updated = async {
|
||||
let client = queue(state).map_err(|(_, why)| anyhow::anyhow!(why))?;
|
||||
modify(&client, |list| reconciled(list, &desired)).await
|
||||
}
|
||||
.await;
|
||||
match updated {
|
||||
Ok(true) => tracing::info!(%agent, "backend egress: derived entries updated"),
|
||||
Ok(false) => {}
|
||||
Err(e) => {
|
||||
tracing::warn!(%agent, error = %format!("{e:#}"), "backend egress: updating the allow list failed");
|
||||
return Err(error_problem(
|
||||
wanted_error_status(&e),
|
||||
&format!(
|
||||
"the backend environment is stored, but the allowed domains were not \
|
||||
updated ({e:#}); saving it again retries"
|
||||
),
|
||||
));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Reconcile every agent's derived entries from the environments in `store`:
|
||||
/// the live agents, and every agent that has a derived entry now. An agent
|
||||
/// whose environment cannot be read keeps its entries.
|
||||
///
|
||||
/// Answers how many agents could not be read.
|
||||
///
|
||||
/// # Errors
|
||||
/// A queue failure reading or writing the list.
|
||||
pub(crate) async fn reconcile(
|
||||
client: &async_nats::Client,
|
||||
store: &impl AccountStore,
|
||||
live: &BTreeSet<String>,
|
||||
) -> anyhow::Result<usize> {
|
||||
let mut agents = live.clone();
|
||||
if let Some(list) = crate::egress_allow::current(client).await? {
|
||||
agents.extend(
|
||||
list.entries
|
||||
.iter()
|
||||
.filter(|e| e.derived)
|
||||
.filter_map(|e| e.agent.clone()),
|
||||
);
|
||||
}
|
||||
let base = base_domains();
|
||||
let mut desired = BTreeMap::new();
|
||||
let mut unreadable = 0;
|
||||
for agent in agents {
|
||||
let Ok(path) = backend_env::env_path(&agent) else {
|
||||
continue;
|
||||
};
|
||||
match store.read_optional::<BackendEnv>(&path).await {
|
||||
Ok(stored) => {
|
||||
let entries = stored.map_or_else(Vec::new, |s| derive(&agent, &s.env, &base));
|
||||
desired.insert(agent, entries);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(%agent, error = %e, "backend egress: reading the backend env failed; its entries stay");
|
||||
unreadable += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
if modify(client, |list| reconciled(list, &desired)).await? {
|
||||
let derived: usize = desired.values().map(Vec::len).sum();
|
||||
tracing::info!(
|
||||
agents = desired.len(),
|
||||
derived,
|
||||
"backend egress: backfilled the allow list"
|
||||
);
|
||||
}
|
||||
Ok(unreadable)
|
||||
}
|
||||
|
||||
/// Run [`reconcile`] once the queue and the store answer, retrying with a
|
||||
/// growing wait until a pass reads every agent. One pass at start: the
|
||||
/// routes keep the entries after it.
|
||||
pub(crate) fn spawn_backfill(
|
||||
client: async_nats::Client,
|
||||
wanted: Arc<WantedWriter>,
|
||||
hives: Vec<String>,
|
||||
) {
|
||||
tokio::spawn(async move {
|
||||
let mut wait = Duration::from_secs(5);
|
||||
loop {
|
||||
let pass = async {
|
||||
swarm_queue_client::ensure_connected(&client)?;
|
||||
let store = crate::store::connect().await?;
|
||||
let live = crate::subagent_term::live_agents(&wanted, &hives).await;
|
||||
anyhow::ensure!(!live.is_empty(), "no live agent is declared yet");
|
||||
reconcile(&client, &store, &live).await
|
||||
}
|
||||
.await;
|
||||
match pass {
|
||||
Ok(0) => return,
|
||||
Ok(n) => tracing::warn!(
|
||||
unreadable = n,
|
||||
"backend egress: backfill incomplete; retrying"
|
||||
),
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %format!("{e:#}"), "backend egress: backfill failed; retrying");
|
||||
}
|
||||
}
|
||||
tokio::time::sleep(wait).await;
|
||||
wait = (wait * 2).min(Duration::from_mins(5));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::agent_limits::tests::declared;
|
||||
use crate::linked_accounts::tests::FakeStore;
|
||||
use crate::test_nats::TestNats;
|
||||
use serde_json::json;
|
||||
|
||||
fn base() -> Vec<String> {
|
||||
vec!["api.anthropic.com".to_owned(), ".claude.ai".to_owned()]
|
||||
}
|
||||
|
||||
fn hosts(env: &str) -> Vec<String> {
|
||||
derive("atlas", env, &base())
|
||||
.into_iter()
|
||||
.map(|e| e.domain)
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_url_is_reduced_to_its_host() {
|
||||
for (value, host) in [
|
||||
("https://openrouter.ai/api/v1", "openrouter.ai"),
|
||||
("http://llm.example.com", "llm.example.com"),
|
||||
("https://LLM.Example.com:8443/v1?x=1#f", "llm.example.com"),
|
||||
("llm.example.com/v1", "llm.example.com"),
|
||||
("llm.example.com:8080", "llm.example.com"),
|
||||
("\"https://llm.example.com/v1\"", "llm.example.com"),
|
||||
("'https://llm.example.com/v1'", "llm.example.com"),
|
||||
("https://user:pw@llm.example.com/v1", "llm.example.com"),
|
||||
] {
|
||||
assert_eq!(
|
||||
hosts(&format!("ANTHROPIC_BASE_URL={value}\n")),
|
||||
[host],
|
||||
"{value}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_host_the_swarm_would_refuse_gets_no_entry() {
|
||||
for value in [
|
||||
"http://10.0.0.5:8080/v1",
|
||||
"https://127.0.0.1",
|
||||
"http://[::1]:11434",
|
||||
"http://localhost:11434/v1",
|
||||
"https://metadata",
|
||||
"ftp://llm.example.com",
|
||||
"https://*.example.com",
|
||||
"",
|
||||
"https://",
|
||||
// The base list already allows it.
|
||||
"https://api.anthropic.com/v1",
|
||||
"https://a.claude.ai",
|
||||
] {
|
||||
assert_eq!(
|
||||
hosts(&format!("ANTHROPIC_BASE_URL={value}\n")),
|
||||
Vec::<String>::new(),
|
||||
"{value:?}"
|
||||
);
|
||||
}
|
||||
// Control: the same shape with a public name gets one.
|
||||
assert_eq!(
|
||||
hosts("ANTHROPIC_BASE_URL=https://llm.example.com\n"),
|
||||
["llm.example.com"]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn only_url_variables_count_and_the_last_assignment_wins() {
|
||||
let env = "# ANTHROPIC_BASE_URL=https://commented.example.com\n\
|
||||
ANTHROPIC_API_KEY=https://not-a-url-var.example.com\n\
|
||||
OPENAI_BASE_URL=https://acp.example.org/v1\n\
|
||||
ANTHROPIC_BASE_URL=https://first.example.com\n\
|
||||
ANTHROPIC_BASE_URL=https://second.example.com\n";
|
||||
assert_eq!(hosts(env), ["acp.example.org", "second.example.com"]);
|
||||
assert_eq!(hosts("ANTHROPIC_API_KEY=sk-x\n"), Vec::<String>::new());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_derived_entry_is_marked_and_scoped_to_the_agent() {
|
||||
let entries = derive(
|
||||
"atlas",
|
||||
"ANTHROPIC_BASE_URL=https://llm.example.com\n",
|
||||
&base(),
|
||||
);
|
||||
assert_eq!(
|
||||
entries,
|
||||
[Entry {
|
||||
domain: "llm.example.com".to_owned(),
|
||||
agent: Some("atlas".to_owned()),
|
||||
derived: true,
|
||||
}]
|
||||
);
|
||||
}
|
||||
|
||||
fn manual(domain: &str, agent: &str) -> Entry {
|
||||
Entry {
|
||||
domain: domain.to_owned(),
|
||||
agent: Some(agent.to_owned()),
|
||||
derived: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn derived(domain: &str, agent: &str) -> Entry {
|
||||
Entry {
|
||||
derived: true,
|
||||
..manual(domain, agent)
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reconciling_replaces_only_the_named_agents_derived_entries() {
|
||||
let list = AllowList {
|
||||
entries: vec![
|
||||
derived("old.example.com", "atlas"),
|
||||
manual("old.example.com", "atlas"),
|
||||
manual("github.com", "atlas"),
|
||||
derived("other.example.com", "iris"),
|
||||
],
|
||||
};
|
||||
let desired = BTreeMap::from([(
|
||||
"atlas".to_owned(),
|
||||
vec![derived("new.example.com", "atlas")],
|
||||
)]);
|
||||
let after = reconciled(list, &desired);
|
||||
assert_eq!(
|
||||
after.entries,
|
||||
[
|
||||
manual("old.example.com", "atlas"),
|
||||
manual("github.com", "atlas"),
|
||||
derived("other.example.com", "iris"),
|
||||
derived("new.example.com", "atlas"),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
/// The state of a test swarm whose `pr1ma` declares `atlas`, with `store`
|
||||
/// as the secret store.
|
||||
async fn swarm(client: &async_nats::Client, store: &Arc<FakeStore>) -> AppState {
|
||||
AppState {
|
||||
accounts: crate::linked_accounts::Accounts::Fake(Arc::clone(store)),
|
||||
..declared(client).await
|
||||
}
|
||||
}
|
||||
|
||||
async fn list_of(client: &async_nats::Client) -> Vec<Entry> {
|
||||
crate::egress_allow::current(client)
|
||||
.await
|
||||
.expect("the queue answers")
|
||||
.map(|l| {
|
||||
let mut e = l.entries;
|
||||
e.sort();
|
||||
e
|
||||
})
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
async fn put(state: &AppState, agent: &str, env: &str) {
|
||||
crate::backend_env::put_backend_env(
|
||||
axum::extract::State(state.clone()),
|
||||
axum::extract::Path(("pr1ma".to_owned(), agent.to_owned())),
|
||||
axum::Json(serde_json::from_value(json!({"env": env})).expect("a request body")),
|
||||
)
|
||||
.await
|
||||
.expect("stored");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn saving_changing_and_deleting_the_env_moves_the_derived_entries() {
|
||||
let nats = TestNats::start();
|
||||
let client = nats.client().await;
|
||||
let store = Arc::default();
|
||||
let state = swarm(&client, &store).await;
|
||||
|
||||
put(
|
||||
&state,
|
||||
"atlas",
|
||||
"ANTHROPIC_BASE_URL=https://one.example.com/v1\n",
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
list_of(&client).await,
|
||||
[derived("one.example.com", "atlas")]
|
||||
);
|
||||
|
||||
// A change replaces the host, an operator's own entry stays.
|
||||
let _ = crate::egress_allow::add_allowed_domain(
|
||||
axum::extract::State(state.clone()),
|
||||
axum::Json(swarm_controller_api::egress::AllowedDomain {
|
||||
domain: "github.com".to_owned(),
|
||||
agent: Some("atlas".to_owned()),
|
||||
derived: false,
|
||||
}),
|
||||
)
|
||||
.await
|
||||
.expect("manual entry");
|
||||
put(
|
||||
&state,
|
||||
"atlas",
|
||||
"ANTHROPIC_BASE_URL=https://two.example.com/v1\n",
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
list_of(&client).await,
|
||||
[
|
||||
manual("github.com", "atlas"),
|
||||
derived("two.example.com", "atlas")
|
||||
]
|
||||
);
|
||||
|
||||
// Removing the variable removes the entry.
|
||||
put(&state, "atlas", "ANTHROPIC_API_KEY=sk-live\n").await;
|
||||
assert_eq!(list_of(&client).await, [manual("github.com", "atlas")]);
|
||||
|
||||
// Deleting the env takes the entries of the env it held.
|
||||
put(
|
||||
&state,
|
||||
"atlas",
|
||||
"ANTHROPIC_BASE_URL=https://three.example.com\n",
|
||||
)
|
||||
.await;
|
||||
crate::linked_accounts::delete_backend_env(
|
||||
axum::extract::State(state.clone()),
|
||||
axum::extract::Path(("pr1ma".to_owned(), "atlas".to_owned())),
|
||||
)
|
||||
.await
|
||||
.expect("deleted");
|
||||
assert_eq!(list_of(&client).await, [manual("github.com", "atlas")]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_env_with_no_host_never_creates_the_list() {
|
||||
let nats = TestNats::start();
|
||||
let client = nats.client().await;
|
||||
let store = Arc::default();
|
||||
let state = swarm(&client, &store).await;
|
||||
|
||||
put(&state, "atlas", "ANTHROPIC_API_KEY=sk-live\n").await;
|
||||
assert!(
|
||||
swarm_queue_client::egress_allow::open(&client)
|
||||
.await
|
||||
.expect("the queue answers")
|
||||
.is_none(),
|
||||
"no bucket: each hive keeps its own list"
|
||||
);
|
||||
// Control: a host does create it.
|
||||
put(
|
||||
&state,
|
||||
"atlas",
|
||||
"ANTHROPIC_BASE_URL=https://llm.example.com\n",
|
||||
)
|
||||
.await;
|
||||
assert_eq!(list_of(&client).await.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_backfill_derives_from_stored_envs_and_a_second_pass_changes_nothing() {
|
||||
let nats = TestNats::start();
|
||||
let client = nats.client().await;
|
||||
let store = Arc::new(
|
||||
FakeStore::default()
|
||||
.with(
|
||||
"swarm/agents/atlas/backend-env",
|
||||
json!({"env": "ANTHROPIC_BASE_URL=https://llm.example.com/v1\n"}),
|
||||
)
|
||||
.with(
|
||||
"swarm/agents/iris/backend-env",
|
||||
json!({"env": "ANTHROPIC_API_KEY=sk-live\n"}),
|
||||
),
|
||||
);
|
||||
let live: BTreeSet<String> = ["atlas", "iris", "scribe"].map(str::to_owned).into();
|
||||
|
||||
assert_eq!(reconcile(&client, &*store, &live).await.expect("pass"), 0);
|
||||
assert_eq!(
|
||||
list_of(&client).await,
|
||||
[derived("llm.example.com", "atlas")]
|
||||
);
|
||||
|
||||
let published = published(&client).await;
|
||||
assert_eq!(reconcile(&client, &*store, &live).await.expect("again"), 0);
|
||||
assert_eq!(
|
||||
self::published(&client).await,
|
||||
published,
|
||||
"the second pass writes nothing"
|
||||
);
|
||||
|
||||
// An agent that stops being live but keeps its leaf keeps its entry;
|
||||
// one whose leaf is gone loses it.
|
||||
let gone: BTreeSet<String> = BTreeSet::new();
|
||||
reconcile(&client, &*store, &gone).await.expect("pass");
|
||||
assert_eq!(
|
||||
list_of(&client).await,
|
||||
[derived("llm.example.com", "atlas")]
|
||||
);
|
||||
let empty = FakeStore::default();
|
||||
reconcile(&client, &empty, &gone).await.expect("pass");
|
||||
assert_eq!(list_of(&client).await, []);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_unreadable_env_keeps_its_entries() {
|
||||
let nats = TestNats::start();
|
||||
let client = nats.client().await;
|
||||
let state = swarm(&client, &Arc::default()).await;
|
||||
put(
|
||||
&state,
|
||||
"atlas",
|
||||
"ANTHROPIC_BASE_URL=https://llm.example.com\n",
|
||||
)
|
||||
.await;
|
||||
|
||||
let live: BTreeSet<String> = ["atlas"].map(str::to_owned).into();
|
||||
let unreadable = reconcile(&client, &FakeStore::denying(), &live)
|
||||
.await
|
||||
.expect("pass");
|
||||
assert_eq!(unreadable, 1);
|
||||
assert_eq!(
|
||||
list_of(&client).await,
|
||||
[derived("llm.example.com", "atlas")]
|
||||
);
|
||||
}
|
||||
|
||||
async fn published(client: &async_nats::Client) -> Option<String> {
|
||||
swarm_queue_client::egress_allow::open_or_create(client)
|
||||
.await
|
||||
.expect("bucket")
|
||||
.get(swarm_wire::egress_allow::KEY)
|
||||
.await
|
||||
.expect("read")
|
||||
.map(|v| String::from_utf8(v.to_vec()).expect("utf-8"))
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_operator_cannot_add_or_remove_a_derived_entry() {
|
||||
let nats = TestNats::start();
|
||||
let client = nats.client().await;
|
||||
let store = Arc::default();
|
||||
let state = swarm(&client, &store).await;
|
||||
put(
|
||||
&state,
|
||||
"atlas",
|
||||
"ANTHROPIC_BASE_URL=https://llm.example.com\n",
|
||||
)
|
||||
.await;
|
||||
|
||||
let request = |derived: bool| swarm_controller_api::egress::AllowedDomain {
|
||||
domain: "llm.example.com".to_owned(),
|
||||
agent: Some("atlas".to_owned()),
|
||||
derived,
|
||||
};
|
||||
for flagged in [true, false] {
|
||||
let problem = crate::egress_allow::remove_allowed_domain(
|
||||
axum::extract::State(state.clone()),
|
||||
axum::Json(request(flagged)),
|
||||
)
|
||||
.await
|
||||
.expect_err("refused");
|
||||
assert_eq!(
|
||||
problem.status,
|
||||
Some(axum::http::StatusCode::CONFLICT),
|
||||
"{problem:?}"
|
||||
);
|
||||
}
|
||||
let problem = crate::egress_allow::add_allowed_domain(
|
||||
axum::extract::State(state.clone()),
|
||||
axum::Json(request(true)),
|
||||
)
|
||||
.await
|
||||
.expect_err("refused");
|
||||
assert_eq!(problem.status, Some(axum::http::StatusCode::BAD_REQUEST));
|
||||
assert_eq!(
|
||||
list_of(&client).await,
|
||||
[derived("llm.example.com", "atlas")]
|
||||
);
|
||||
|
||||
// Control: the same name added by the operator is a separate entry that
|
||||
// the operator can remove, and the derived one stays.
|
||||
let manual_request = request(false);
|
||||
let _ = crate::egress_allow::add_allowed_domain(
|
||||
axum::extract::State(state.clone()),
|
||||
axum::Json(manual_request.clone()),
|
||||
)
|
||||
.await
|
||||
.expect("a manual entry");
|
||||
let _ = crate::egress_allow::remove_allowed_domain(
|
||||
axum::extract::State(state.clone()),
|
||||
axum::Json(manual_request),
|
||||
)
|
||||
.await
|
||||
.expect("removes the manual one");
|
||||
assert_eq!(
|
||||
list_of(&client).await,
|
||||
[derived("llm.example.com", "atlas")]
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn destroying_an_agent_drops_all_its_entries_and_only_its_own() {
|
||||
let nats = TestNats::start();
|
||||
let client = nats.client().await;
|
||||
let state = swarm(&client, &Arc::default()).await;
|
||||
put(
|
||||
&state,
|
||||
"atlas",
|
||||
"ANTHROPIC_BASE_URL=https://llm.example.com\n",
|
||||
)
|
||||
.await;
|
||||
modify(&client, |mut list| {
|
||||
list.entries.push(manual("github.com", "atlas"));
|
||||
list.entries.push(manual("crates.io", "iris"));
|
||||
list
|
||||
})
|
||||
.await
|
||||
.expect("seed");
|
||||
|
||||
crate::egress_allow::clear_on_destroy(&state, "pr1ma", "atlas").await;
|
||||
assert_eq!(list_of(&client).await, [manual("crates.io", "iris")]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_host_without_a_queue_stores_the_env_and_skips_the_list() {
|
||||
let nats = TestNats::start();
|
||||
let client = nats.client().await;
|
||||
let store = Arc::new(FakeStore::default());
|
||||
let mut state = swarm(&client, &store).await;
|
||||
state.status = None;
|
||||
put(
|
||||
&state,
|
||||
"atlas",
|
||||
"ANTHROPIC_BASE_URL=https://llm.example.com\n",
|
||||
)
|
||||
.await;
|
||||
assert_eq!(store.written(), ["swarm/agents/atlas/backend-env"]);
|
||||
assert_eq!(list_of(&client).await, []);
|
||||
}
|
||||
}
|
||||
|
|
@ -76,6 +76,8 @@ pub async fn put_backend_env(
|
|||
error_problem(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string())
|
||||
})?;
|
||||
|
||||
crate::backend_egress::follow_env(&state, &agent, Some(&stored.env)).await?;
|
||||
|
||||
tracing::info!(%hive, %agent, "backend env stored");
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,6 +28,10 @@ use crate::schedules::{live, queue, read_or_default, refused, write_revision};
|
|||
/// `wanted::MAX_ATTEMPTS`.
|
||||
const MAX_ATTEMPTS: usize = 5;
|
||||
|
||||
/// Why an operator can neither add nor remove an entry the controller derived.
|
||||
const DERIVED_REFUSED: &str = "this entry is kept automatically from the agent's backend environment; \
|
||||
change or remove that variable instead";
|
||||
|
||||
/// Set from the hive module's `baseAllowedDomains` default by
|
||||
/// `nix/host-modules/swarm-controller.nix`: the names, separated by spaces,
|
||||
/// that every hive's proxy lets every authenticated agent reach. Unset means
|
||||
|
|
@ -35,7 +39,7 @@ const MAX_ATTEMPTS: usize = 5;
|
|||
/// the entries its own base list covers.
|
||||
pub(crate) const ENV_BASE_DOMAINS: &str = "SWARM_CONTROLLER_SQUID_BASE_DOMAINS";
|
||||
|
||||
fn base_domains() -> Vec<String> {
|
||||
pub(crate) fn base_domains() -> Vec<String> {
|
||||
std::env::var(ENV_BASE_DOMAINS)
|
||||
.unwrap_or_default()
|
||||
.split_whitespace()
|
||||
|
|
@ -60,7 +64,11 @@ pub(crate) fn checked(req: &AllowedDomain, base: &[String]) -> Result<Entry, Str
|
|||
.map_err(|reason| format!("agent {a:?}: {reason}"))
|
||||
})
|
||||
.transpose()?;
|
||||
Ok(Entry { domain, agent })
|
||||
Ok(Entry {
|
||||
domain,
|
||||
agent,
|
||||
derived: false,
|
||||
})
|
||||
}
|
||||
|
||||
/// What adding an entry to a list came to.
|
||||
|
|
@ -95,6 +103,7 @@ fn view(list: AllowList) -> AllowedDomains {
|
|||
.map(|e| AllowedDomain {
|
||||
domain: e.domain,
|
||||
agent: e.agent,
|
||||
derived: e.derived,
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
|
|
@ -158,7 +167,7 @@ pub(crate) async fn get_allowed_domains(
|
|||
(status = 400, description = "the domain is not a DNS name (an IP address, a \
|
||||
wildcard, a single label and a port are refused), the proxy already allows \
|
||||
it to everyone, the agent is not an identifier or not declared on any hive, \
|
||||
or the list is full (problem+json)", body = String),
|
||||
or the list is full, or the entry is marked derived (problem+json)", body = String),
|
||||
(status = 422, description = "an unknown field"),
|
||||
(status = 503, description = "no swarm queue is configured on this host, or it is \
|
||||
not connected (problem+json)", body = String),
|
||||
|
|
@ -171,6 +180,9 @@ pub(crate) async fn add_allowed_domain(
|
|||
State(state): State<AppState>,
|
||||
Json(req): Json<AllowedDomain>,
|
||||
) -> Result<Json<AllowedDomains>, ProblemDetails> {
|
||||
if req.derived {
|
||||
return Err(error_problem(StatusCode::BAD_REQUEST, DERIVED_REFUSED));
|
||||
}
|
||||
let entry = checked(&req, &base_domains())
|
||||
.map_err(|why| error_problem(StatusCode::BAD_REQUEST, &why))?;
|
||||
let client = queue(&state).map_err(refused)?;
|
||||
|
|
@ -219,6 +231,9 @@ pub(crate) async fn add_allowed_domain(
|
|||
(status = 200, description = "the whole list after the removal", body = AllowedDomains),
|
||||
(status = 400, description = "the domain is not a DNS name, or the agent is not an \
|
||||
identifier (problem+json)", body = String),
|
||||
(status = 409, description = "the entry is one the controller keeps from the agent's \
|
||||
backend environment, which an operator cannot remove (problem+json)",
|
||||
body = String),
|
||||
(status = 422, description = "an unknown field"),
|
||||
(status = 503, description = "no swarm queue is configured on this host, or it is \
|
||||
not connected (problem+json)", body = String),
|
||||
|
|
@ -231,6 +246,9 @@ pub(crate) async fn remove_allowed_domain(
|
|||
State(state): State<AppState>,
|
||||
Json(req): Json<AllowedDomain>,
|
||||
) -> Result<Json<AllowedDomains>, ProblemDetails> {
|
||||
if req.derived {
|
||||
return Err(error_problem(StatusCode::CONFLICT, DERIVED_REFUSED));
|
||||
}
|
||||
// The base list is not consulted: an entry stored before a name joined it
|
||||
// can still be removed.
|
||||
let entry = checked(&req, &[]).map_err(|why| error_problem(StatusCode::BAD_REQUEST, &why))?;
|
||||
|
|
@ -241,6 +259,13 @@ pub(crate) async fn remove_allowed_domain(
|
|||
.await
|
||||
.map_err(|e| queue_problem(&e))?;
|
||||
if !list.entries.contains(&entry) {
|
||||
if list
|
||||
.entries
|
||||
.iter()
|
||||
.any(|e| e.derived && e.domain == entry.domain && e.agent == entry.agent)
|
||||
{
|
||||
return Err(error_problem(StatusCode::CONFLICT, DERIVED_REFUSED));
|
||||
}
|
||||
return Ok(Json(view(list)));
|
||||
}
|
||||
list.entries.retain(|e| *e != entry);
|
||||
|
|
@ -259,6 +284,87 @@ pub(crate) async fn remove_allowed_domain(
|
|||
)))
|
||||
}
|
||||
|
||||
/// The stored list, or `None` when no one has written it.
|
||||
///
|
||||
/// # Errors
|
||||
/// A queue failure.
|
||||
pub(crate) async fn current(client: &async_nats::Client) -> anyhow::Result<Option<AllowList>> {
|
||||
let Some(store) = egress_allow::open(client).await? else {
|
||||
return Ok(None);
|
||||
};
|
||||
let (list, _, _) = read_or_default::<AllowList>(&store, KEY).await?;
|
||||
Ok(Some(list))
|
||||
}
|
||||
|
||||
/// Re-read the list and write `change` of it, retrying a lost write, until it
|
||||
/// is written or `change` leaves it as it is. Answers whether a write landed.
|
||||
///
|
||||
/// A list that was never written is written only when `change` makes it
|
||||
/// non-empty: the key's existence is what makes the swarm's list replace each
|
||||
/// hive's own, so a change that adds nothing must not create it.
|
||||
///
|
||||
/// # Errors
|
||||
/// A queue failure, or `MAX_ATTEMPTS` conflicting writes.
|
||||
pub(crate) async fn modify(
|
||||
client: &async_nats::Client,
|
||||
change: impl Fn(AllowList) -> AllowList,
|
||||
) -> anyhow::Result<bool> {
|
||||
let mut store = egress_allow::open(client).await?;
|
||||
for _ in 0..MAX_ATTEMPTS {
|
||||
let (mut list, revision) = match &store {
|
||||
Some(store) => {
|
||||
let (list, revision, _) = read_or_default::<AllowList>(store, KEY).await?;
|
||||
(list, revision)
|
||||
}
|
||||
None => (AllowList::default(), None),
|
||||
};
|
||||
list.entries.sort();
|
||||
let mut changed = change(list.clone());
|
||||
changed.entries.sort();
|
||||
changed.entries.dedup();
|
||||
if changed == list && (revision.is_some() || changed.entries.is_empty()) {
|
||||
return Ok(false);
|
||||
}
|
||||
if store.is_none() {
|
||||
store = Some(egress_allow::open_or_create(client).await?);
|
||||
}
|
||||
let target = store.as_ref().expect("opened above");
|
||||
if write_revision(target, KEY, &changed, revision)
|
||||
.await?
|
||||
.is_some()
|
||||
{
|
||||
return Ok(true);
|
||||
}
|
||||
}
|
||||
anyhow::bail!("gave up after {MAX_ATTEMPTS} conflicting writes")
|
||||
}
|
||||
|
||||
/// Drop every entry of a destroyed agent, derived and operator-added: a name
|
||||
/// is free again once its agent is destroyed, and the next agent of that name
|
||||
/// would inherit what it left. A failure is logged and does not stop the
|
||||
/// destroy, as in `agent_permissions::clear_on_destroy`.
|
||||
pub(crate) async fn clear_on_destroy(state: &AppState, hive: &str, agent: &str) {
|
||||
let cleared = async {
|
||||
let client = queue(state).map_err(|(_, why)| anyhow::anyhow!(why))?;
|
||||
modify(&client, |mut list| {
|
||||
list.entries.retain(|e| e.agent.as_deref() != Some(agent));
|
||||
list
|
||||
})
|
||||
.await
|
||||
}
|
||||
.await;
|
||||
match cleared {
|
||||
Ok(_) => tracing::info!(hive, agent, "destroyed agent's allowed domains cleared"),
|
||||
Err(e) => tracing::error!(
|
||||
hive,
|
||||
agent,
|
||||
error = %format!("{e:#}"),
|
||||
"clearing this agent's allowed domains FAILED; it was destroyed anyway, so a new \
|
||||
agent of the same name inherits them. Declaring the agent destroyed again re-runs this."
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
|
@ -269,6 +375,7 @@ mod tests {
|
|||
AllowedDomain {
|
||||
domain: domain.to_owned(),
|
||||
agent: agent.map(str::to_owned),
|
||||
derived: false,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -445,12 +552,14 @@ mod tests {
|
|||
.map(|i| Entry {
|
||||
domain: format!("host{i}.example.com"),
|
||||
agent: None,
|
||||
derived: false,
|
||||
})
|
||||
.collect(),
|
||||
};
|
||||
let new = Entry {
|
||||
domain: "github.com".to_owned(),
|
||||
agent: None,
|
||||
derived: false,
|
||||
};
|
||||
let why = added(full.clone(), &new).err().expect("a full list");
|
||||
assert!(why.contains("1000"), "{why}");
|
||||
|
|
|
|||
|
|
@ -621,6 +621,8 @@ pub async fn delete_backend_env(
|
|||
e.problem()
|
||||
})?;
|
||||
|
||||
crate::backend_egress::follow_env(&state, &agent, None).await?;
|
||||
|
||||
tracing::info!(%hive, %agent, "backend env deleted");
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -39,6 +39,7 @@ mod alerts;
|
|||
mod archetypes;
|
||||
mod attic_copy;
|
||||
mod auth;
|
||||
mod backend_egress;
|
||||
mod backend_env;
|
||||
mod builder;
|
||||
mod cache_miss;
|
||||
|
|
|
|||
|
|
@ -17,7 +17,7 @@ use crate::routes::hives::HiveEntry;
|
|||
use crate::routes::{AppState, StatusUnavailable, error_problem, wanted_error_status};
|
||||
use crate::{
|
||||
agent_archetype, agent_identity, agent_limits, agent_permissions, agent_status,
|
||||
agents_md_initial, auth, inference, schedules, wanted,
|
||||
agents_md_initial, auth, egress_allow, inference, schedules, wanted,
|
||||
};
|
||||
|
||||
/// The swarm's agent roster — every agent the swarm holds an identity for.
|
||||
|
|
@ -214,6 +214,7 @@ pub(crate) async fn set_agent_state(
|
|||
agent_permissions::clear_on_destroy(&state, &hive, &agent).await;
|
||||
agent_archetype::clear_on_destroy(&state, &hive, &agent).await;
|
||||
agents_md_initial::clear_on_destroy(&state, &hive, &agent).await;
|
||||
egress_allow::clear_on_destroy(&state, &hive, &agent).await;
|
||||
}
|
||||
Ok(Json(render(&declaration)))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -375,7 +375,8 @@ fn spawn_subagent_streams(status: Option<&Arc<status::StatusReader>>, hives: &[H
|
|||
}
|
||||
|
||||
/// The per-agent background passes: subagent and terminal streams, and
|
||||
/// schedule keys, and the permissions bucket and its import, and the
|
||||
/// 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.
|
||||
/// Lifted out of `main` for `clippy::too_many_lines`.
|
||||
|
|
@ -386,6 +387,7 @@ pub(crate) fn spawn_per_agent_upkeep(
|
|||
spawn_subagent_streams(status, hives);
|
||||
spawn_schedule_reconcile(status, hives);
|
||||
spawn_agent_permissions(status, hives);
|
||||
spawn_backend_egress_backfill(status, hives);
|
||||
crate::attic_copy::spawn(hives.iter().map(|h| h.name.clone()).collect());
|
||||
crate::proxy_credential::spawn(hives.iter().map(|h| h.name.clone()).collect());
|
||||
if let Some(status) = status {
|
||||
|
|
@ -403,6 +405,17 @@ fn spawn_agent_permissions(status: Option<&Arc<status::StatusReader>>, hives: &[
|
|||
agent_permissions::spawn(status.queue_client(), wanted, hives);
|
||||
}
|
||||
|
||||
/// Derive each agent's allowed-domain entries from the backend environments in
|
||||
/// the store when a swarm queue is wired up. Same shape as
|
||||
/// [`spawn_subagent_streams`].
|
||||
fn spawn_backend_egress_backfill(status: Option<&Arc<status::StatusReader>>, hives: &[HiveEntry]) {
|
||||
let (Some(status), Some(wanted)) = (status, wanted_writer(status)) else {
|
||||
return;
|
||||
};
|
||||
let hives = hives.iter().map(|h| h.name.clone()).collect();
|
||||
crate::backend_egress::spawn_backfill(status.queue_client(), wanted, hives);
|
||||
}
|
||||
|
||||
/// Start rewriting every agent's schedule key from the definitions when a
|
||||
/// swarm queue is wired up. Same shape as [`spawn_subagent_streams`].
|
||||
fn spawn_schedule_reconcile(status: Option<&Arc<status::StatusReader>>, hives: &[HiveEntry]) {
|
||||
|
|
|
|||
Loading…
Reference in a new issue