//! 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. //! //! An entry is marked `derived` and a hive renders it like any other. Only the //! operator's save or delete of the environment through the controller creates //! one ([`follow_env`]): the agent can create its own `backend-env` leaf while //! it is empty, so what the store holds is not evidence that an operator chose //! it. [`spawn_backfill`] therefore only drops: it keeps a derived entry while //! the stored environment still names its exact host, and never adds one. Each //! host goes through `egress_allow::checked`, the check an operator's own //! entry gets. //! //! An agent still on the hand-placed `backendEnvironmentFile` alone has no //! derived entry, and neither does an environment saved before this existed //! until the operator saves it once more. 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, BACKEND_ENV_SOURCE, 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 `_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`. pub(crate) 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 { let mut entries: Vec = derive_named(agent, env, base) .into_iter() .map(|(_, entry)| entry) .collect(); entries.sort(); entries.dedup(); entries } /// [`fn@derive`] with the variable each entry came from, one per variable. fn derive_named<'a>(agent: &str, env: &'a str, base: &[String]) -> Vec<(&'a str, Entry)> { let mut entries = Vec::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.push(( name, Entry { derived: true, source: Some(BACKEND_ENV_SOURCE.to_owned()), ..entry }, )), Err(why) => { tracing::info!(%agent, variable = name, %why, "backend egress: no entry for this host"); } } } entries } /// `list` with, for each agent in `desired`, its entries derived from the backend /// environment replaced by the given ones. Entries of other agents, /// operator-added entries and entries derived from linked accounts are /// untouched. A full list takes no more derived entries. fn replaced(mut list: AllowList, desired: &BTreeMap>) -> AllowList { list.entries.retain(|e| { !(e.from_backend_env() && 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| replaced(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(()) } /// `list` without the backend-env entries of the agents in `still_named` whose /// host is not in that agent's set. Nothing is added; entries of agents not in /// the map, operator-added entries and link-derived entries are untouched. fn pruned(mut list: AllowList, still_named: &BTreeMap>) -> AllowList { list.entries.retain(|e| { let Some(hosts) = e.agent.as_ref().and_then(|a| still_named.get(a)) else { return true; }; !e.from_backend_env() || hosts.contains(&e.domain) }); list } /// Drop the derived entries whose host the stored environment no longer names. /// /// Only the agents that have a derived entry are looked at. An agent whose /// stored environment is gone loses all of them; one whose environment cannot /// be read keeps them. A host the stored environment names that is not /// already derived is left out and logged by variable name, since whoever /// wrote the store may be the agent itself. /// /// 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, ) -> anyhow::Result { let Some(current) = crate::egress_allow::current(client).await? else { return Ok(0); }; let mut derived: BTreeMap> = BTreeMap::new(); for entry in current.entries.iter().filter(|e| e.from_backend_env()) { if let Some(agent) = &entry.agent { derived .entry(agent.clone()) .or_default() .insert(entry.domain.clone()); } } let base = base_domains(); let mut still_named = BTreeMap::new(); let mut unreadable = 0; for (agent, kept) in derived { let Ok(path) = backend_env::env_path(&agent) else { continue; }; match store.read_optional::(&path).await { Ok(stored) => { let named = stored.map_or_else(Vec::new, |s| { derive_named(&agent, &s.env, &base) .into_iter() .map(|(variable, entry)| (variable.to_owned(), entry.domain)) .collect() }); for (variable, domain) in &named { if !kept.contains(domain) { tracing::warn!( %agent, %variable, "backend egress: the agent's stored env names a host not derived \ (only an operator save derives)" ); } } still_named.insert(agent, named.into_iter().map(|(_, d)| d).collect()); } Err(e) => { tracing::warn!(%agent, error = %e, "backend egress: reading the backend env failed; its entries stay"); unreadable += 1; } } } if modify(client, |list| pruned(list, &still_named)).await? { tracing::info!( agents = still_named.len(), "backend egress: dropped derived entries the stored env no longer names" ); } Ok(unreadable) } /// Run [`reconcile`] and `link_egress::reconcile` once the queue and the store /// answer, retrying with a growing wait until a pass reads every agent that /// has an entry and some hive has declared an agent. pub(crate) fn spawn_backfill( client: async_nats::Client, wanted: Arc, hives: Vec, ) { 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; let unreadable = reconcile(&client, &store).await? + crate::link_egress::reconcile(&client, &store, &live).await?; Ok::<_, anyhow::Error>(unreadable + usize::from(live.is_empty())) } .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 std::sync::Arc; 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 { vec!["api.anthropic.com".to_owned(), ".claude.ai".to_owned()] } fn hosts(env: &str) -> Vec { 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::::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::::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, source: Some(BACKEND_ENV_SOURCE.to_owned()), }] ); } fn manual(domain: &str, agent: &str) -> Entry { Entry { domain: domain.to_owned(), agent: Some(agent.to_owned()), derived: false, source: None, } } fn derived(domain: &str, agent: &str) -> Entry { Entry { derived: true, source: Some(BACKEND_ENV_SOURCE.to_owned()), ..manual(domain, agent) } } #[test] fn replacing_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 = replaced(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"), ] ); } fn linked(domain: &str, agent: &str, source: &str) -> Entry { Entry { derived: true, source: Some(source.to_owned()), ..manual(domain, agent) } } /// A derived entry written before sources existed. fn untagged(domain: &str, agent: &str) -> Entry { Entry { source: None, ..derived(domain, agent) } } #[test] fn a_backend_env_save_leaves_link_entries_and_replaces_untagged_ones() { let list = AllowList { entries: vec![ linked("github.com", "atlas", "link:github"), linked("hs.example.org", "atlas", "link:matrix:work"), untagged("old.example.com", "atlas"), ], }; let desired = BTreeMap::from([( "atlas".to_owned(), vec![derived("new.example.com", "atlas")], )]); let after = replaced(list, &desired); assert_eq!( after.entries, [ linked("github.com", "atlas", "link:github"), linked("hs.example.org", "atlas", "link:matrix:work"), derived("new.example.com", "atlas"), ] ); } #[test] fn pruning_the_backend_env_leaves_link_entries_and_drops_untagged_ones() { let list = AllowList { entries: vec![ linked("github.com", "atlas", "link:github"), untagged("old.example.com", "atlas"), derived("kept.example.com", "atlas"), ], }; let still_named = BTreeMap::from([( "atlas".to_owned(), BTreeSet::from(["kept.example.com".to_owned()]), )]); assert_eq!( pruned(list, &still_named).entries, [ linked("github.com", "atlas", "link:github"), derived("kept.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) -> AppState { AppState { accounts: crate::linked_accounts::Accounts::Fake(Arc::clone(store)), ..declared(client).await } } async fn list_of(client: &async_nats::Client) -> Vec { 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: nothing was written" ); // 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); } const OPERATOR: &str = "ANTHROPIC_BASE_URL=https://llm.example.com/v1\n"; fn leaf(agent: &str) -> String { format!("swarm/agents/{agent}/backend-env") } async fn write_leaf(store: &FakeStore, agent: &str, env: &str) { store .write( &leaf(agent), &BackendEnv { env: env.to_owned(), }, ) .await .expect("written"); } /// The agent creates its own leaf while it is empty, so a stored value is /// not evidence of an operator's choice: the backfill adds nothing for it. #[tokio::test] async fn a_leaf_created_outside_the_route_yields_no_entry() { let nats = TestNats::start(); let client = nats.client().await; let store = Arc::new(FakeStore::default()); let state = swarm(&client, &store).await; write_leaf( &store, "atlas", "ANTHROPIC_BASE_URL=https://evil.example.net\n", ) .await; // With no list at all, the pass creates nothing. assert_eq!(reconcile(&client, &*store).await.expect("pass"), 0); assert!( swarm_queue_client::egress_allow::open(&client) .await .expect("the queue answers") .is_none() ); // With a list that holds only an operator's entry, nothing is added. modify(&client, |mut list| { list.entries.push(manual("github.com", "atlas")); list }) .await .expect("seed"); assert_eq!(reconcile(&client, &*store).await.expect("pass"), 0); assert_eq!(list_of(&client).await, [manual("github.com", "atlas")]); // Control: the operator's save of the same text derives the host. put( &state, "atlas", "ANTHROPIC_BASE_URL=https://evil.example.net\n", ) .await; assert_eq!( list_of(&client).await, [ derived("evil.example.net", "atlas"), manual("github.com", "atlas") ] ); } /// A pass keeps an entry whose exact host the stored env still names, /// drops one it no longer names, and never adds the new host. #[tokio::test] async fn the_backfill_keeps_a_named_host_drops_a_stale_one_and_adds_nothing() { let nats = TestNats::start(); let client = nats.client().await; let store = Arc::new(FakeStore::default()); let state = swarm(&client, &store).await; put(&state, "atlas", OPERATOR).await; let entry = derived("llm.example.com", "atlas"); // Control: unchanged content keeps the entry, and a second pass writes // nothing. assert_eq!(reconcile(&client, &*store).await.expect("pass"), 0); assert_eq!(list_of(&client).await, [entry]); let before = published(&client).await; reconcile(&client, &*store).await.expect("again"); assert_eq!(published(&client).await, before); // The stored text changed outside the route: the old host goes, the // new one does not come. write_leaf( &store, "atlas", "ANTHROPIC_BASE_URL=https://other.example.net\n", ) .await; reconcile(&client, &*store).await.expect("pass"); assert_eq!(list_of(&client).await, []); } /// The leaf is deleted outside the controller and created again by the /// agent with another host before the pass: the old entry goes and the new /// host is not derived. A leaf that is simply gone drops its entries too. #[tokio::test] async fn a_leaf_deleted_and_recreated_by_the_agent_loses_the_entry_and_gains_none() { let nats = TestNats::start(); let client = nats.client().await; let state = swarm(&client, &Arc::default()).await; put(&state, "atlas", OPERATOR).await; put(&state, "iris", OPERATOR).await; assert_eq!(list_of(&client).await.len(), 2, "control: both derived"); let recreated = FakeStore::default(); write_leaf( &recreated, "atlas", "ANTHROPIC_BASE_URL=https://evil.example.net\n", ) .await; // iris's leaf is gone, atlas's was created again by atlas. reconcile(&client, &recreated).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 unreadable = reconcile(&client, &FakeStore::denying()) .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 { 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, []); } }