Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/swarm-controller/src/egress_allow.rs

667 lines
24 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. A hive always allows the names its own nix configuration
//! lists (`services.hyperhive.deploy.hive-controller.squid.allowedDomains`) as well, so these entries
//! only add to them.
//!
//! 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;
/// 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 or a linked \
account; change or remove that variable, or unlink the account, 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
/// 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";
pub(crate) 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,
derived: false,
source: None,
})
}
/// 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,
derived: e.derived,
})
.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",
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 entry adds to
/// the names each hive's nix configuration already allows.
#[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, 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),
(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> {
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)?;
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. A name a hive's nix
/// configuration lists stays allowed on that hive.
#[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 = 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),
(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> {
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))?;
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) {
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);
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
)))
}
/// 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, so a change that adds nothing creates no bucket or key.
///
/// # 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::*;
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),
derived: false,
}
}
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"
);
}
#[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,
derived: false,
source: None,
})
.collect(),
};
let new = Entry {
domain: "github.com".to_owned(),
agent: None,
derived: false,
source: 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.
#[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"}]}"#)
);
}
}