557 lines
20 KiB
Rust
557 lines
20 KiB
Rust
//! `/api/egress/allowed-domains` — the destinations the swarm lets agents
|
|
//! reach through their hive's egress proxy.
|
|
//!
|
|
//! The whole list is one value under one key of `egress-allow`
|
|
//! (`swarm_wire::egress_allow`), which this daemon alone writes. Every hive
|
|
//! reads it and renders the entries for the agents it runs: a swarm-wide entry
|
|
//! joins the proxy's allow list for everyone, an entry with an agent is that
|
|
//! agent's alone. Once the key exists it replaces the list a hive's own nix
|
|
//! configuration names, so an empty list is a list.
|
|
//!
|
|
//! Each name goes into the proxy's configuration, so it is checked here with
|
|
//! `swarm_wire::egress_allow::normalize_domain` (the hive checks again) and
|
|
//! refused when the proxy already allows it to everyone. The names the proxy
|
|
//! always allows are the ones in [`ENV_BASE_DOMAINS`].
|
|
|
|
use axum::Json;
|
|
use axum::extract::State;
|
|
use axum::http::StatusCode;
|
|
use problem_details::ProblemDetails;
|
|
use swarm_controller_api::egress::{AllowedDomain, AllowedDomains};
|
|
use swarm_queue_client::egress_allow;
|
|
use swarm_wire::egress_allow::{AllowList, Entry, KEY, MAX_ENTRIES, covered_by, normalize_domain};
|
|
|
|
use crate::routes::{AppState, error_problem, wanted_error_status};
|
|
use crate::schedules::{live, queue, read_or_default, refused, write_revision};
|
|
|
|
/// How many times a losing writer re-reads and re-applies, as in
|
|
/// `wanted::MAX_ATTEMPTS`.
|
|
const MAX_ATTEMPTS: usize = 5;
|
|
|
|
/// 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
|
|
/// none are known and nothing is refused on that ground; a hive still skips
|
|
/// the entries its own base list covers.
|
|
pub(crate) const ENV_BASE_DOMAINS: &str = "SWARM_CONTROLLER_SQUID_BASE_DOMAINS";
|
|
|
|
fn base_domains() -> Vec<String> {
|
|
std::env::var(ENV_BASE_DOMAINS)
|
|
.unwrap_or_default()
|
|
.split_whitespace()
|
|
.map(str::to_ascii_lowercase)
|
|
.collect()
|
|
}
|
|
|
|
/// The entry to store for `req`, or why it is refused.
|
|
pub(crate) fn checked(req: &AllowedDomain, base: &[String]) -> Result<Entry, String> {
|
|
let domain = normalize_domain(&req.domain)?;
|
|
if covered_by(&domain, base) {
|
|
return Err(format!(
|
|
"{domain} is already allowed for every agent by the proxy's base list"
|
|
));
|
|
}
|
|
let agent = req
|
|
.agent
|
|
.as_deref()
|
|
.map(|a| {
|
|
hive_types::Ident::parse(a)
|
|
.map(hive_types::Ident::into_string)
|
|
.map_err(|reason| format!("agent {a:?}: {reason}"))
|
|
})
|
|
.transpose()?;
|
|
Ok(Entry { domain, agent })
|
|
}
|
|
|
|
/// What adding an entry to a list came to.
|
|
enum Added {
|
|
/// The entry was there: nothing to write.
|
|
Already(AllowList),
|
|
/// The list with the entry in it.
|
|
Now(AllowList),
|
|
}
|
|
|
|
/// `list` with `entry` added, or why it can't take another entry.
|
|
fn added(mut list: AllowList, entry: &Entry) -> Result<Added, String> {
|
|
if list.entries.contains(entry) {
|
|
return Ok(Added::Already(list));
|
|
}
|
|
if list.entries.len() >= MAX_ENTRIES {
|
|
return Err(format!(
|
|
"the list holds {MAX_ENTRIES} entries, which is the most it takes"
|
|
));
|
|
}
|
|
list.entries.push(entry.clone());
|
|
list.entries.sort();
|
|
Ok(Added::Now(list))
|
|
}
|
|
|
|
fn view(list: AllowList) -> AllowedDomains {
|
|
let mut entries = list.entries;
|
|
entries.sort();
|
|
AllowedDomains {
|
|
entries: entries
|
|
.into_iter()
|
|
.map(|e| AllowedDomain {
|
|
domain: e.domain,
|
|
agent: e.agent,
|
|
})
|
|
.collect(),
|
|
}
|
|
}
|
|
|
|
fn queue_problem(e: &anyhow::Error) -> ProblemDetails {
|
|
tracing::warn!(error = %format!("{e:#}"), "egress allow-list: queue failure");
|
|
error_problem(wanted_error_status(e), &format!("{e:#}"))
|
|
}
|
|
|
|
async fn open_or_create(
|
|
client: &async_nats::Client,
|
|
) -> Result<async_nats::jetstream::kv::Store, ProblemDetails> {
|
|
egress_allow::open_or_create(client)
|
|
.await
|
|
.map_err(|e| queue_problem(&anyhow::Error::new(e)))
|
|
}
|
|
|
|
#[utoipa::path(
|
|
get,
|
|
path = "/api/egress/allowed-domains",
|
|
responses(
|
|
(status = 200, description = "every entry, swarm-wide and per agent, sorted; empty \
|
|
when the swarm has not set a list, in which case each hive keeps its own",
|
|
body = AllowedDomains),
|
|
(status = 503, description = "no swarm queue is configured on this host, or it is \
|
|
not connected (problem+json)", body = String),
|
|
(status = 500, description = "the list could not be read (problem+json)",
|
|
body = String),
|
|
),
|
|
tag = "egress"
|
|
)]
|
|
pub(crate) async fn get_allowed_domains(
|
|
State(state): State<AppState>,
|
|
) -> Result<Json<AllowedDomains>, ProblemDetails> {
|
|
let client = queue(&state).map_err(refused)?;
|
|
let Some(store) = egress_allow::open(&client)
|
|
.await
|
|
.map_err(|e| queue_problem(&anyhow::Error::new(e)))?
|
|
else {
|
|
return Ok(Json(AllowedDomains::default()));
|
|
};
|
|
let (current, _, _) = read_or_default::<AllowList>(&store, KEY)
|
|
.await
|
|
.map_err(|e| queue_problem(&e))?;
|
|
Ok(Json(view(current)))
|
|
}
|
|
|
|
/// Allow one destination, for every agent or for one.
|
|
///
|
|
/// Adding an entry that is already there changes nothing. The first entry
|
|
/// makes the swarm's list replace each hive's own, so a name the hive's nix
|
|
/// configuration allows and this list lacks stops being allowed once hives read
|
|
/// it.
|
|
#[utoipa::path(
|
|
post,
|
|
path = "/api/egress/allowed-domains",
|
|
request_body = AllowedDomain,
|
|
responses(
|
|
(status = 200, description = "the whole list after the add", body = AllowedDomains),
|
|
(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),
|
|
(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),
|
|
(status = 500, description = "the list could not be written (problem+json)",
|
|
body = String),
|
|
),
|
|
tag = "egress"
|
|
)]
|
|
pub(crate) async fn add_allowed_domain(
|
|
State(state): State<AppState>,
|
|
Json(req): Json<AllowedDomain>,
|
|
) -> Result<Json<AllowedDomains>, ProblemDetails> {
|
|
let entry = checked(&req, &base_domains())
|
|
.map_err(|why| error_problem(StatusCode::BAD_REQUEST, &why))?;
|
|
let client = queue(&state).map_err(refused)?;
|
|
if let Some(agent) = &entry.agent
|
|
&& !live(&state).await?.contains(agent)
|
|
{
|
|
return Err(error_problem(
|
|
StatusCode::BAD_REQUEST,
|
|
&format!("no hive's wanted state declares an agent {agent:?} that is not destroyed"),
|
|
));
|
|
}
|
|
let store = open_or_create(&client).await?;
|
|
for _ in 0..MAX_ATTEMPTS {
|
|
let (list, revision, _) = read_or_default::<AllowList>(&store, KEY)
|
|
.await
|
|
.map_err(|e| queue_problem(&e))?;
|
|
let list = match added(list, &entry) {
|
|
Ok(Added::Already(list)) => return Ok(Json(view(list))),
|
|
Ok(Added::Now(list)) => list,
|
|
Err(why) => return Err(error_problem(StatusCode::BAD_REQUEST, &why)),
|
|
};
|
|
if write_revision(&store, KEY, &list, revision)
|
|
.await
|
|
.map_err(|e| queue_problem(&e))?
|
|
.is_some()
|
|
{
|
|
tracing::info!(domain = %entry.domain, agent = ?entry.agent, "allowed a domain");
|
|
return Ok(Json(view(list)));
|
|
}
|
|
}
|
|
Err(queue_problem(&anyhow::anyhow!(
|
|
"gave up adding {} after {MAX_ATTEMPTS} conflicting writes",
|
|
entry.domain
|
|
)))
|
|
}
|
|
|
|
/// Stop allowing one destination for the scope it was added with.
|
|
///
|
|
/// Removing an entry that is not there changes nothing. Removing the last one
|
|
/// leaves an empty list, which still replaces each hive's own.
|
|
#[utoipa::path(
|
|
delete,
|
|
path = "/api/egress/allowed-domains",
|
|
request_body = AllowedDomain,
|
|
responses(
|
|
(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 = 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),
|
|
(status = 500, description = "the list could not be written (problem+json)",
|
|
body = String),
|
|
),
|
|
tag = "egress"
|
|
)]
|
|
pub(crate) async fn remove_allowed_domain(
|
|
State(state): State<AppState>,
|
|
Json(req): Json<AllowedDomain>,
|
|
) -> Result<Json<AllowedDomains>, ProblemDetails> {
|
|
// 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))?;
|
|
let client = queue(&state).map_err(refused)?;
|
|
let store = open_or_create(&client).await?;
|
|
for _ in 0..MAX_ATTEMPTS {
|
|
let (mut list, revision, _) = read_or_default::<AllowList>(&store, KEY)
|
|
.await
|
|
.map_err(|e| queue_problem(&e))?;
|
|
if !list.entries.contains(&entry) {
|
|
return Ok(Json(view(list)));
|
|
}
|
|
list.entries.retain(|e| *e != entry);
|
|
if write_revision(&store, KEY, &list, revision)
|
|
.await
|
|
.map_err(|e| queue_problem(&e))?
|
|
.is_some()
|
|
{
|
|
tracing::info!(domain = %entry.domain, agent = ?entry.agent, "stopped allowing a domain");
|
|
return Ok(Json(view(list)));
|
|
}
|
|
}
|
|
Err(queue_problem(&anyhow::anyhow!(
|
|
"gave up removing {} after {MAX_ATTEMPTS} conflicting writes",
|
|
entry.domain
|
|
)))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::agent_limits::tests::{declared, state};
|
|
use crate::test_nats::TestNats;
|
|
|
|
fn entry(domain: &str, agent: Option<&str>) -> AllowedDomain {
|
|
AllowedDomain {
|
|
domain: domain.to_owned(),
|
|
agent: agent.map(str::to_owned),
|
|
}
|
|
}
|
|
|
|
async fn add(
|
|
state: &AppState,
|
|
domain: &str,
|
|
agent: Option<&str>,
|
|
) -> Result<AllowedDomains, ProblemDetails> {
|
|
add_allowed_domain(State(state.clone()), Json(entry(domain, agent)))
|
|
.await
|
|
.map(|j| j.0)
|
|
}
|
|
|
|
async fn remove(
|
|
state: &AppState,
|
|
domain: &str,
|
|
agent: Option<&str>,
|
|
) -> Result<AllowedDomains, ProblemDetails> {
|
|
remove_allowed_domain(State(state.clone()), Json(entry(domain, agent)))
|
|
.await
|
|
.map(|j| j.0)
|
|
}
|
|
|
|
async fn listed(state: &AppState) -> Vec<(String, Option<String>)> {
|
|
get_allowed_domains(State(state.clone()))
|
|
.await
|
|
.expect("the list is read")
|
|
.0
|
|
.entries
|
|
.into_iter()
|
|
.map(|e| (e.domain, e.agent))
|
|
.collect()
|
|
}
|
|
|
|
/// The raw value a hive reads, or `None` for no key.
|
|
async fn published(client: &async_nats::Client) -> Option<String> {
|
|
egress_allow::open_or_create(client)
|
|
.await
|
|
.expect("bucket")
|
|
.get(KEY)
|
|
.await
|
|
.expect("read")
|
|
.map(|v| String::from_utf8(v.to_vec()).expect("utf-8"))
|
|
}
|
|
|
|
fn base() -> Vec<String> {
|
|
vec!["api.anthropic.com".to_owned(), ".claude.ai".to_owned()]
|
|
}
|
|
|
|
#[test]
|
|
fn a_name_is_stored_normalised_and_an_ip_literal_is_refused() {
|
|
let stored = checked(&entry(" GitHub.com ", Some("atlas")), &base()).expect("a name");
|
|
assert_eq!(stored.domain, "github.com");
|
|
assert_eq!(stored.agent.as_deref(), Some("atlas"));
|
|
let dotted = checked(&entry(".githubusercontent.com", None), &base()).expect("dot form");
|
|
assert_eq!(dotted.domain, ".githubusercontent.com");
|
|
assert_eq!(dotted.agent, None);
|
|
|
|
for bad in [
|
|
"1.2.3.4",
|
|
"127.0.0.1",
|
|
"[::1]",
|
|
"::1",
|
|
"*.example.com",
|
|
"localhost",
|
|
"example.com:443",
|
|
"example.com\nhttp_access allow all",
|
|
] {
|
|
assert!(
|
|
checked(&entry(bad, None), &base()).is_err(),
|
|
"{bad:?} was accepted"
|
|
);
|
|
}
|
|
for agent in ["", "At Las", "a/b", "../argus"] {
|
|
assert!(
|
|
checked(&entry("github.com", Some(agent)), &base()).is_err(),
|
|
"agent {agent:?} was accepted"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_name_the_base_list_already_allows_is_refused() {
|
|
for covered in [
|
|
"api.anthropic.com",
|
|
"claude.ai",
|
|
"a.claude.ai",
|
|
".claude.ai",
|
|
] {
|
|
let why = checked(&entry(covered, None), &base()).expect_err(covered);
|
|
assert!(why.contains("base list"), "{why}");
|
|
}
|
|
// Control: the same names pass when the base list is unknown, and names
|
|
// next to the base ones are not refused.
|
|
assert!(checked(&entry("claude.ai", None), &[]).is_ok());
|
|
for fresh in ["x.api.anthropic.com", ".anthropic.com", "notclaude.ai"] {
|
|
assert!(checked(&entry(fresh, None), &base()).is_ok(), "{fresh}");
|
|
}
|
|
}
|
|
|
|
/// Through the real handlers and a real `nats-server`: an entry lands in
|
|
/// the one value a hive reads, the two scopes of one name are two entries,
|
|
/// a repeat changes nothing, and a removal takes only the scope it names.
|
|
///
|
|
/// The published bytes are pinned literally because hive-c0re's
|
|
/// `workers::squid_allow` tests decode this same string; a change to either
|
|
/// side's reading of it fails one of the two.
|
|
#[tokio::test]
|
|
async fn the_route_writes_the_list_a_hive_reads_and_each_scope_is_its_own_entry() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let state = declared(&client).await;
|
|
|
|
add(&state, "github.com", None).await.expect("swarm-wide");
|
|
add(&state, "crates.io", Some("atlas"))
|
|
.await
|
|
.expect("for atlas");
|
|
add(&state, "github.com", Some("atlas"))
|
|
.await
|
|
.expect("the same name for atlas alone");
|
|
let before = published(&client).await;
|
|
add(&state, "GITHUB.com", None).await.expect("a repeat");
|
|
assert_eq!(published(&client).await, before, "a repeat writes nothing");
|
|
|
|
assert_eq!(
|
|
published(&client).await.as_deref(),
|
|
Some(
|
|
r#"{"entries":[{"domain":"crates.io","agent":"atlas"},{"domain":"github.com"},{"domain":"github.com","agent":"atlas"}]}"#
|
|
)
|
|
);
|
|
assert_eq!(
|
|
listed(&state).await,
|
|
[
|
|
("crates.io".to_owned(), Some("atlas".to_owned())),
|
|
("github.com".to_owned(), None),
|
|
("github.com".to_owned(), Some("atlas".to_owned())),
|
|
]
|
|
);
|
|
|
|
remove(&state, "github.com", Some("atlas"))
|
|
.await
|
|
.expect("removes atlas's");
|
|
assert_eq!(
|
|
listed(&state).await,
|
|
[
|
|
("crates.io".to_owned(), Some("atlas".to_owned())),
|
|
("github.com".to_owned(), None),
|
|
]
|
|
);
|
|
// Removing what is not there is not an error and writes nothing.
|
|
let before = published(&client).await;
|
|
remove(&state, "github.com", Some("atlas"))
|
|
.await
|
|
.expect("already gone");
|
|
assert_eq!(published(&client).await, before);
|
|
|
|
remove(&state, "github.com", None)
|
|
.await
|
|
.expect("swarm-wide");
|
|
remove(&state, "crates.io", Some("atlas"))
|
|
.await
|
|
.expect("atlas's");
|
|
assert_eq!(
|
|
published(&client).await.as_deref(),
|
|
Some(r#"{"entries":[]}"#),
|
|
"the last removal leaves an empty list, which still replaces a hive's own"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn a_full_list_takes_no_new_entry_but_a_present_one_is_not_a_new_one() {
|
|
let full = AllowList {
|
|
entries: (0..MAX_ENTRIES)
|
|
.map(|i| Entry {
|
|
domain: format!("host{i}.example.com"),
|
|
agent: None,
|
|
})
|
|
.collect(),
|
|
};
|
|
let new = Entry {
|
|
domain: "github.com".to_owned(),
|
|
agent: None,
|
|
};
|
|
let why = added(full.clone(), &new).err().expect("a full list");
|
|
assert!(why.contains("1000"), "{why}");
|
|
// Controls: an entry already there is not refused, and one fewer fits.
|
|
let present = full.entries[0].clone();
|
|
assert!(matches!(
|
|
added(full.clone(), &present),
|
|
Ok(Added::Already(_))
|
|
));
|
|
let mut room = full;
|
|
room.entries.pop();
|
|
let Ok(Added::Now(list)) = added(room, &new) else {
|
|
panic!("a list with room takes the entry");
|
|
};
|
|
assert_eq!(list.entries.len(), MAX_ENTRIES);
|
|
}
|
|
|
|
/// A read before anything was added answers an empty list and creates no
|
|
/// bucket, so a hive keeps its own list.
|
|
#[tokio::test]
|
|
async fn a_read_before_any_entry_answers_empty_and_creates_no_bucket() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let state = state(&client);
|
|
|
|
assert_eq!(listed(&state).await, []);
|
|
assert!(
|
|
egress_allow::open(&client)
|
|
.await
|
|
.expect("the queue answers")
|
|
.is_none()
|
|
);
|
|
}
|
|
|
|
/// An agent no hive declares gets no entry, so a typo cannot allow a name
|
|
/// for an agent that does not exist yet; an unscoped add is not asked.
|
|
#[tokio::test]
|
|
async fn an_agent_no_hive_declares_is_refused_and_nothing_is_written() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let state = declared(&client).await;
|
|
|
|
let problem = add(&state, "github.com", Some("iris"))
|
|
.await
|
|
.expect_err("no hive declares iris");
|
|
assert_eq!(problem.status, Some(StatusCode::BAD_REQUEST), "{problem:?}");
|
|
assert_eq!(published(&client).await, None);
|
|
add(&state, "github.com", Some("atlas"))
|
|
.await
|
|
.expect("control: a declared agent is accepted");
|
|
}
|
|
|
|
async fn post_raw(app: &axum::Router, body: &'static str) -> StatusCode {
|
|
use tower::ServiceExt as _;
|
|
app.clone()
|
|
.oneshot(
|
|
axum::http::Request::post("/api/egress/allowed-domains")
|
|
.header(axum::http::header::CONTENT_TYPE, "application/json")
|
|
.body(axum::body::Body::from(body))
|
|
.expect("request"),
|
|
)
|
|
.await
|
|
.expect("router answers")
|
|
.status()
|
|
}
|
|
|
|
/// Through the real router: an IP literal or a wildcard is a 400 and
|
|
/// writes nothing, an unknown field a 422, and the well-formed body after
|
|
/// them is the control that the same route reaches the bucket.
|
|
#[tokio::test]
|
|
async fn the_router_refuses_an_ip_literal_and_a_wildcard_before_the_queue() {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
let app = crate::startup::build_app(declared(&client).await);
|
|
|
|
for body in [
|
|
r#"{"domain":"1.2.3.4"}"#,
|
|
r#"{"domain":"10.0.0.1","agent":"atlas"}"#,
|
|
r#"{"domain":"*.example.com"}"#,
|
|
r#"{"domain":"[::1]"}"#,
|
|
r#"{"domain":""}"#,
|
|
] {
|
|
assert_eq!(
|
|
post_raw(&app, body).await,
|
|
StatusCode::BAD_REQUEST,
|
|
"{body}"
|
|
);
|
|
}
|
|
assert_eq!(
|
|
post_raw(&app, r#"{"domain":"github.com","hive":"pr1ma"}"#).await,
|
|
StatusCode::UNPROCESSABLE_ENTITY
|
|
);
|
|
assert_eq!(published(&client).await, None);
|
|
|
|
assert_eq!(
|
|
post_raw(&app, r#"{"domain":"github.com"}"#).await,
|
|
StatusCode::OK
|
|
);
|
|
assert_eq!(
|
|
published(&client).await.as_deref(),
|
|
Some(r#"{"entries":[{"domain":"github.com"}]}"#)
|
|
);
|
|
}
|
|
}
|