From 45c59185981bd16b8d8d20f543b4dbf84f2903ed Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 09:39:25 +0000 Subject: [PATCH 1/2] fix: close unauthenticated chunk path traversal, peer/replicate auth, eviction panic, broken downloads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five defects, each with a regression test. 1. Arbitrary file read via GET /api/chunks/{sha} (unauthenticated) ChunkStore::chunk_path never validated its argument, which arrives straight from a URL path segment. hash[2..4] of "../../../etc/passwd" is "/.", and PathBuf::join on an absolute component discards the store prefix, so the path resolved to /etc/passwd. Verified against a running node before the fix. chunk_path now returns Option and rejects anything that is not 64 lowercase hex chars; get/has/verify/delete all handle the rejection. 2. Remote panic on the same endpoint hash[..2] panicked for input shorter than two bytes or on a non-ASCII char boundary, so GET /api/chunks/a reset the connection. Same validation covers it; the endpoint now returns 404. 3. POST /api/replicate and POST /api/peers were unauthenticated /api/replicate takes a caller-supplied peer_url, fetches it and stores the response — an SSRF primitive plus arbitrary store pollution. A registered peer is not passive either: the heartbeat loop contacts it every 60s and auto-replication pulls from it every 5 minutes, so one anonymous request set up a standing ingestion channel from an attacker's server. Both now go through authorize() at AccessLevel::Write. 4. Auto-eviction task panicked on first run eviction::build_replica_map_from_beacon used reqwest::blocking, which builds and drops its own runtime. Called from the tokio::spawn eviction loop in server.rs, that panics with "Cannot drop a runtime in a context where blocking is not allowed" — the task died the first time storage exceeded the limit, so nodes never evicted and grew past their cap indefinitely. Switched to ureq (already a dependency, genuinely synchronous, safe from the CLI too) and wrapped the call in block_in_place so it no longer stalls an async worker. 5. GET /api/download/{collection}/{item} returned 500 for every item reconstruct_cog only understood the tiled layout and bailed on "Missing: earthgrid:width". Nothing writes that metadata: ingest_raster is the sole producer and has no callers, so every item comes from ingest_file, which stores sequential raw byte chunks. Those concatenate back to the original COG exactly, so reconstruct_cog now falls back to reconstruct_raw, which reassembles in order and checks the result against the earthgrid:file_hash recorded at ingest — a missing chunk is a hard error rather than a truncated download. Also: the beacon federation WebSocket (/api/beacon/ws) accepted any client and handed it apply_remote_event, including NodePruned, which deletes any node from the registry by ID. It now requires the shared grid key when one is configured, and the outbound dialer presents that key. With no key set the socket stays open, matching the node's open-mode behaviour elsewhere — federated beacons must set EARTHGRID_API_KEY to the same value on each host. Tests: 156 passing, up from 147. Traversal, panic and the auth changes were also verified end to end against a running node from a non-loopback address. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Qjes4orjaT3k6ShNXj5Axm --- earthgrid-core/src/beacon_federation.rs | 88 ++++++++++++++++- earthgrid-core/src/chunk_store.rs | 120 +++++++++++++++++++++--- earthgrid-core/src/eviction.rs | 86 ++++++++++++++--- earthgrid-core/src/reconstruct.rs | 115 +++++++++++++++++++++++ earthgrid-core/src/routes/chunks.rs | 23 ++++- earthgrid-core/src/routes/federation.rs | 23 ++++- earthgrid-core/src/server.rs | 21 +++-- 7 files changed, 436 insertions(+), 40 deletions(-) diff --git a/earthgrid-core/src/beacon_federation.rs b/earthgrid-core/src/beacon_federation.rs index 09ed6f6..e0a763d 100644 --- a/earthgrid-core/src/beacon_federation.rs +++ b/earthgrid-core/src/beacon_federation.rs @@ -11,6 +11,7 @@ use axum::{ State, ws::{Message, WebSocket, WebSocketUpgrade}, }, + http::{HeaderMap, StatusCode}, response::IntoResponse, }; use futures_util::{SinkExt, StreamExt}; @@ -85,11 +86,38 @@ impl FederationState { // --------------------------------------------------------------------------- /// GET /beacon/ws — upgrade to WebSocket for federation sync. +/// +/// Requires the shared grid key when one is configured. The peer that opens +/// this socket is not a passive reader: `apply_remote_event` lets it upsert +/// nodes into the registry and, via `NodePruned`, **delete** any node by ID. +/// Without a check, anyone who can reach the port could wipe a beacon's +/// registry or fill it with fabricated nodes. +/// +/// When no key is configured the socket stays open, matching the rest of the +/// node's open-mode behaviour — so set `EARTHGRID_API_KEY` on federated +/// beacons, and set it to the *same* value on each, or they cannot peer. pub async fn ws_handler( ws: WebSocketUpgrade, State(state): State, + headers: HeaderMap, ) -> impl IntoResponse { + if state.auth.is_enabled() { + let key = headers + .get("x-api-key") + .and_then(|v| v.to_str().ok()) + .or_else(|| { + headers + .get("authorization") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + }); + if state.auth.check_write(key).is_err() { + warn!("Federation: rejecting unauthenticated WebSocket upgrade"); + return (StatusCode::UNAUTHORIZED, "federation key required").into_response(); + } + } ws.on_upgrade(move |socket| handle_peer_connection(socket, state)) + .into_response() } async fn handle_peer_connection(socket: WebSocket, state: BeaconState) { @@ -230,6 +258,28 @@ pub fn spawn_peer_connections(state: BeaconState, peer_urls: Vec) { } } +/// Build the WebSocket handshake request for a peer beacon, attaching the +/// shared grid key as `x-api-key` when one is configured. +fn build_federation_request( + ws_url: &str, + api_key: &str, +) -> std::result::Result { + use tokio_tungstenite::tungstenite::client::IntoClientRequest; + + let mut request = ws_url + .into_client_request() + .map_err(|e| format!("invalid websocket url: {e}"))?; + + if !api_key.is_empty() { + let value = api_key + .parse() + .map_err(|_| "grid key is not a valid header value".to_string())?; + request.headers_mut().insert("x-api-key", value); + } + + Ok(request) +} + /// Connect to a single peer beacon, reconnecting with exponential backoff. async fn connect_to_peer_loop(state: BeaconState, peer_url: String) { let ws_url = peer_url @@ -244,7 +294,17 @@ async fn connect_to_peer_loop(state: BeaconState, peer_url: String) { loop { info!("Federation: connecting to peer {}", ws_url); - match tokio_tungstenite::connect_async(&ws_url).await { + // Present the shared grid key so the peer's `ws_handler` accepts us. + // Rebuilt each attempt because `connect_async` consumes the request. + let request = match build_federation_request(&ws_url, &state.auth.api_key) { + Ok(r) => r, + Err(e) => { + warn!("Federation: cannot build request for {}: {}", ws_url, e); + return; + } + }; + + match tokio_tungstenite::connect_async(request).await { Ok((ws_stream, _)) => { info!("Federation: connected to {}", ws_url); backoff = Duration::from_secs(1); // Reset backoff on success @@ -416,4 +476,30 @@ mod tests { assert!(json.contains("full_sync")); assert!(json.contains("beacon-1")); } + + #[test] + fn federation_request_carries_grid_key() { + let req = build_federation_request("ws://beacon.example/api/beacon/ws", "grid-secret") + .expect("request should build"); + assert_eq!( + req.headers().get("x-api-key").map(|v| v.to_str().unwrap()), + Some("grid-secret"), + "dialer must present the key the peer's ws_handler now requires" + ); + } + + #[test] + fn federation_request_omits_empty_key() { + let req = build_federation_request("ws://beacon.example/api/beacon/ws", "") + .expect("request should build"); + assert!( + req.headers().get("x-api-key").is_none(), + "keyless deployments must dial exactly as before" + ); + } + + #[test] + fn federation_request_rejects_bad_url() { + assert!(build_federation_request("not a url", "k").is_err()); + } } diff --git a/earthgrid-core/src/chunk_store.rs b/earthgrid-core/src/chunk_store.rs index e59c443..212ba04 100644 --- a/earthgrid-core/src/chunk_store.rs +++ b/earthgrid-core/src/chunk_store.rs @@ -171,14 +171,34 @@ impl ChunkStore { hex::encode(hasher.finalize()) } + /// Whether `hash` is a canonical SHA-256 digest: exactly 64 lowercase hex + /// characters, the form `hash_bytes` produces. + /// + /// Uppercase is rejected on purpose. Content addresses have one canonical + /// spelling, and on case-insensitive filesystems accepting both would let + /// two distinct hash strings resolve to the same file. + pub fn is_valid_hash(hash: &str) -> bool { + hash.len() == 64 + && hash + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) + } + /// Store a chunk. Returns its SHA-256 hash. /// /// If the chunk already exists, this is a no-op (content-addressed dedup). pub fn put(&mut self, data: &[u8]) -> Result { let hash = Self::hash_bytes(data); - // Fast dedup check via chunk path - let path = self.chunk_path(&hash); + // Fast dedup check via chunk path. + // `hash` comes from `hash_bytes`, so it is canonical by construction and + // `chunk_path` cannot reject it — but degrade rather than panic if that + // ever stops holding. + let Some(path) = self.chunk_path(&hash) else { + return Err(EarthGridError::Other(format!( + "computed hash is not a canonical digest: {hash}" + ))); + }; if path.exists() { return Ok(hash); } @@ -226,7 +246,9 @@ impl ChunkStore { /// /// Returns `None` if the chunk doesn't exist. pub fn get(&self, hash: &str) -> Result>> { - let path = self.chunk_path(hash); + let Some(path) = self.chunk_path(hash) else { + return Ok(None); // Malformed hash — treat as "not found", never touch the FS + }; if !path.exists() { return Ok(None); } @@ -242,7 +264,7 @@ impl ChunkStore { /// Check if a chunk exists. pub fn has(&self, hash: &str) -> bool { - self.chunk_path(hash).exists() + self.chunk_path(hash).is_some_and(|p| p.exists()) } /// Get the size in bytes of a single chunk from the DB index. @@ -259,7 +281,9 @@ impl ChunkStore { /// /// Returns `Ok(true)` if valid, `Err(IntegrityViolation)` if corrupted. pub fn verify(&self, hash: &str) -> Result { - let path = self.chunk_path(hash); + let Some(path) = self.chunk_path(hash) else { + return Err(EarthGridError::ChunkNotFound(hash.to_string())); + }; if !path.exists() { return Err(EarthGridError::ChunkNotFound(hash.to_string())); } @@ -276,7 +300,9 @@ impl ChunkStore { /// Delete a chunk by hash. Returns `true` if it was deleted. pub fn delete(&mut self, hash: &str) -> Result { - let path = self.chunk_path(hash); + let Some(path) = self.chunk_path(hash) else { + return Ok(false); // Malformed hash — nothing to delete + }; if path.exists() { let size = self.chunk_size(hash).unwrap_or(0); fs::remove_file(&path)?; @@ -339,11 +365,25 @@ impl ChunkStore { // --- Private --- - fn chunk_path(&self, hash: &str) -> PathBuf { - self.store_path - .join(&hash[..2]) - .join(&hash[2..4]) - .join(hash) + /// Build the on-disk path for a chunk, or `None` if `hash` is not a + /// canonical SHA-256 digest. + /// + /// Validation is mandatory, not cosmetic: `hash` reaches this function + /// straight from URL path segments (`GET /api/chunks/{sha}`). Without the + /// check, `hash[2..4]` of `../../../etc/passwd` is `"/."`, and `PathBuf::join` + /// on an absolute component *discards* `store_path` entirely — turning the + /// chunk endpoint into an arbitrary-file read. Short or non-ASCII input + /// would also panic on the byte slices. + fn chunk_path(&self, hash: &str) -> Option { + if !Self::is_valid_hash(hash) { + return None; + } + Some( + self.store_path + .join(&hash[..2]) + .join(&hash[2..4]) + .join(hash), + ) } fn db_chunk_count(db: &Connection) -> usize { @@ -405,8 +445,10 @@ impl ChunkStore { { if entry.file_type().is_file() { if let Some(name) = entry.path().file_name().and_then(|n| n.to_str()) { - // Skip non-hash files (tmp files, etc.) - if name.len() == 64 && name.chars().all(|c| c.is_ascii_hexdigit()) { + // Skip non-hash files (tmp files, etc.). Uses the same + // validator as `chunk_path`, so we never index a name that + // could not later be read back. + if Self::is_valid_hash(name) { let size = entry.metadata().map(|m| m.len()).unwrap_or(0) as i64; tx.execute( "INSERT OR IGNORE INTO chunks (hash, size_bytes, created_at) VALUES (?1, ?2, ?3)", @@ -521,6 +563,56 @@ mod tests { assert_eq!(s.chunks_stored, 1); } + #[test] + fn test_is_valid_hash() { + let valid = "a".repeat(64); + assert!(ChunkStore::is_valid_hash(&valid)); + assert!(ChunkStore::is_valid_hash(&ChunkStore::hash_bytes(b"anything"))); + + assert!(!ChunkStore::is_valid_hash("")); + assert!(!ChunkStore::is_valid_hash("abc")); + assert!(!ChunkStore::is_valid_hash(&"a".repeat(63))); + assert!(!ChunkStore::is_valid_hash(&"a".repeat(65))); + assert!(!ChunkStore::is_valid_hash(&"A".repeat(64)), "uppercase is not canonical"); + assert!(!ChunkStore::is_valid_hash(&"g".repeat(64)), "non-hex rejected"); + } + + /// A hash from a URL path segment must never escape the store directory. + /// `"../../../etc/passwd"` splits to `".."` / `"/."`; because `"/."` is + /// absolute, `PathBuf::join` used to drop `store_path` and resolve to + /// `/etc/passwd`. + #[test] + fn test_traversal_rejected() { + let dir = tempdir().unwrap(); + let store = ChunkStore::new(&dir.path().join("store"), 0.0).unwrap(); + + for evil in [ + "../../../etc/passwd", + "..%2F..%2Fetc%2Fpasswd", + "/etc/passwd", + "....//....//etc/passwd", + ] { + assert!(store.chunk_path(evil).is_none(), "chunk_path accepted {evil:?}"); + assert_eq!(store.get(evil).unwrap(), None, "get() served {evil:?}"); + assert!(!store.has(evil)); + } + } + + /// Short or non-ASCII hashes used to panic on the `hash[..2]` byte slice, + /// killing the connection for `GET /api/chunks/a`. + #[test] + fn test_malformed_hash_does_not_panic() { + let dir = tempdir().unwrap(); + let mut store = ChunkStore::new(&dir.path().join("store"), 0.0).unwrap(); + + for bad in ["", "a", "ab", "abc", "é", "aé", "日本語"] { + assert_eq!(store.get(bad).unwrap(), None); + assert!(!store.has(bad)); + assert!(!store.delete(bad).unwrap()); + assert!(store.verify(bad).is_err()); + } + } + #[test] fn test_atomic_write() { let dir = tempdir().unwrap(); @@ -528,7 +620,7 @@ mod tests { let mut store = ChunkStore::new(&store_path, 0.0).unwrap(); let hash = store.put(b"atomic write").unwrap(); - let path = store.chunk_path(&hash); + let path = store.chunk_path(&hash).unwrap(); let tmp = path.with_extension("tmp"); // Temp file should not exist after put (renamed) assert!(!tmp.exists()); diff --git a/earthgrid-core/src/eviction.rs b/earthgrid-core/src/eviction.rs index cb06375..fc054f3 100644 --- a/earthgrid-core/src/eviction.rs +++ b/earthgrid-core/src/eviction.rs @@ -204,25 +204,31 @@ fn build_replica_map_from_beacon(beacon_url: &str) -> std::collections::HashMap< let mut map = std::collections::HashMap::new(); let url = format!("{}/api/beacon/nodes", beacon_url.trim_end_matches('/')); - // Blocking HTTP request (eviction runs in a background task) - let client = reqwest::blocking::Client::builder() - .timeout(std::time::Duration::from_secs(10)) - .build(); - let Ok(client) = client else { return map }; - - let resp = match client.get(&url).send() { - Ok(r) if r.status().is_success() => r, - _ => { + // Synchronous HTTP via ureq, NOT reqwest::blocking. + // + // `evict_with_beacon_url` is called from both sync contexts (the CLI) and + // from inside a `tokio::spawn` (the auto-eviction loop in server.rs). + // `reqwest::blocking` builds and drops its own runtime, which panics with + // "Cannot drop a runtime in a context where blocking is not allowed" when + // called from an async context — killing the auto-eviction task on its + // first run, so nodes silently grew past their storage limit forever. + // ureq is genuinely synchronous and safe from either context. + let body: serde_json::Value = match ureq::get(&url) + .config() + .timeout_global(Some(std::time::Duration::from_secs(10))) + .build() + .call() + { + Ok(mut resp) => match resp.body_mut().read_json() { + Ok(v) => v, + Err(_) => return map, + }, + Err(_) => { eprintln!("⚠️ Eviction: could not reach beacon at {}", url); return map; } }; - let body: serde_json::Value = match resp.json() { - Ok(v) => v, - Err(_) => return map, - }; - // Count how many nodes have each collection let mut collection_node_count: std::collections::HashMap = std::collections::HashMap::new(); if let Some(nodes) = body.get("nodes").and_then(|n| n.as_array()) { @@ -285,3 +291,55 @@ fn now_ts() -> f64 { .unwrap_or_default() .as_secs_f64() } + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + /// Regression: eviction must be callable from inside an async task. + /// + /// `build_replica_map_from_beacon` used `reqwest::blocking`, which builds + /// and drops its own runtime. Called from the `tokio::spawn` auto-eviction + /// loop in `server.rs`, that panicked with "Cannot drop a runtime in a + /// context where blocking is not allowed", so the task died on its first + /// run and nodes never evicted anything — they just grew past their limit. + /// + /// The beacon URL points at a closed port: we are asserting that the call + /// *returns* (a failed lookup yields an empty replica map) rather than + /// unwinding. + #[test] + fn evict_from_async_context_does_not_panic() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let joined = rt.block_on(async { + tokio::spawn(async { + let dir = tempfile::tempdir().unwrap(); + let catalog = Catalog::new(&dir.path().join("catalog.db")).unwrap(); + let mut store = ChunkStore::new(&dir.path().join("store"), 0.0).unwrap(); + store.put(b"some bytes worth evicting").unwrap(); + + // target_gb below current usage forces the eviction path, + // which reaches out to the beacon for the replica map. + evict_with_beacon_url( + &catalog, + &mut store, + 0.0000000001, + None, + Some("http://127.0.0.1:1"), + ) + .map(|r| r.items_deleted) + }) + .await + }); + + assert!( + joined.is_ok(), + "eviction panicked inside a tokio task: {:?}", + joined.err() + ); + assert!(joined.unwrap().is_ok(), "eviction returned an error"); + } +} diff --git a/earthgrid-core/src/reconstruct.rs b/earthgrid-core/src/reconstruct.rs index 3548bcd..ca0f0a9 100644 --- a/earthgrid-core/src/reconstruct.rs +++ b/earthgrid-core/src/reconstruct.rs @@ -148,6 +148,19 @@ pub fn reconstruct_cog( store: &mut ChunkStore, bands: Option<&[String]>, ) -> Result> { + // Items produced by `ingest::ingest_file` — which is every item this node + // can currently create, since `ingest_raster` is not wired to any caller — + // carry no tile metadata: no `earthgrid:width`, `tile_size`, `tile_cols`. + // The tiled path below would fail them all with "Missing: earthgrid:width", + // which is why `GET /api/download/...` returned 500 for every item. + // + // `ingest_file` splits the source file into sequential raw byte chunks, so + // concatenating them in order reproduces the original COG byte-for-byte. + // Serve that directly. + if props_u32(&item.properties, "earthgrid:width").is_err() { + return reconstruct_raw(item, store); + } + let band_data = reconstruct_bands(item, store, bands)?; if band_data.is_empty() { return Err(crate::error::EarthGridError::Other( @@ -176,6 +189,50 @@ pub fn reconstruct_cog( ) } +/// Reassemble an item stored as sequential raw byte chunks (the `ingest_file` +/// layout) by concatenating its chunks in order. +/// +/// When the item records `earthgrid:file_hash` — which `ingest_file` always +/// writes — the result is checked against it, so a missing or corrupted chunk +/// surfaces as an integrity error instead of a silently truncated download. +pub fn reconstruct_raw(item: &StacItem, store: &mut ChunkStore) -> Result> { + if item.chunk_hashes.is_empty() { + return Err(crate::error::EarthGridError::Other(format!( + "Item {} has no chunks", + item.id + ))); + } + + let mut out = Vec::new(); + for sha in &item.chunk_hashes { + match store.get(sha)? { + Some(data) => out.extend_from_slice(&data), + None => { + return Err(crate::error::EarthGridError::ChunkNotFound(format!( + "{} (item {})", + sha, item.id + ))) + } + } + } + + if let Some(expected) = item + .properties + .get("earthgrid:file_hash") + .and_then(|v| v.as_str()) + { + let actual = ChunkStore::hash_bytes(&out); + if actual != expected { + return Err(crate::error::EarthGridError::IntegrityViolation { + expected: expected.to_string(), + actual, + }); + } + } + + Ok(out) +} + /// Compute NDVI and return as COG. pub fn ndvi_cog( red_data: &[u8], @@ -357,3 +414,61 @@ fn dtype_size(dtype: &str) -> usize { _ => 2, } } + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use crate::ingest; + + /// End-to-end: an item ingested the way `/api/fetch` ingests one must be + /// downloadable again, byte-for-byte. + /// + /// Regression for `GET /api/download/{collection}/{item}` returning 500 on + /// every item. `ingest_file` writes no `earthgrid:width`, so `reconstruct_cog` + /// hit `props_u32(..., "earthgrid:width")?` and bailed with + /// "Missing: earthgrid:width" — there was no item in existence it could serve. + #[test] + fn reconstruct_roundtrips_ingest_file_items() { + let dir = tempfile::tempdir().unwrap(); + let mut store = ChunkStore::new(&dir.path().join("store"), 0.0).unwrap(); + + // Multi-chunk payload so ordering is actually exercised. + let original: Vec = (0..300_000u32).map(|i| (i % 251) as u8).collect(); + let src = dir.path().join("scene.tif"); + std::fs::write(&src, &original).unwrap(); + + let item = ingest::ingest_file(&src, "test-collection", 64 * 1024, &mut store).unwrap(); + assert!(item.chunk_hashes.len() > 1, "expected a multi-chunk item"); + assert!( + item.properties.get("earthgrid:width").is_none(), + "ingest_file is not expected to record tile metadata" + ); + + let rebuilt = reconstruct_cog(&item, &mut store, None).unwrap(); + assert_eq!(rebuilt, original, "download must return the original bytes"); + } + + /// A missing chunk must be a hard error, never a truncated file. + #[test] + fn reconstruct_raw_rejects_missing_chunk() { + let dir = tempfile::tempdir().unwrap(); + let mut store = ChunkStore::new(&dir.path().join("store"), 0.0).unwrap(); + + let original: Vec = (0..200_000u32).map(|i| (i % 197) as u8).collect(); + let src = dir.path().join("scene.tif"); + std::fs::write(&src, &original).unwrap(); + let item = ingest::ingest_file(&src, "c", 64 * 1024, &mut store).unwrap(); + + store.delete(&item.chunk_hashes[1]).unwrap(); + + let err = reconstruct_cog(&item, &mut store, None).unwrap_err(); + assert!( + matches!(err, crate::error::EarthGridError::ChunkNotFound(_)), + "expected ChunkNotFound, got {err:?}" + ); + } +} diff --git a/earthgrid-core/src/routes/chunks.rs b/earthgrid-core/src/routes/chunks.rs index f2944dc..bc9acd3 100644 --- a/earthgrid-core/src/routes/chunks.rs +++ b/earthgrid-core/src/routes/chunks.rs @@ -1,13 +1,15 @@ use axum::{ - extract::{Path, Query, State}, + extract::{ConnectInfo, Path, Query, State}, http::{HeaderMap, StatusCode}, response::IntoResponse, Json, }; use serde::Deserialize; use sha2::{Digest, Sha256}; +use std::net::SocketAddr; -use crate::server::{AppState, api_key, err, LimitQuery}; +use crate::auth::AccessLevel; +use crate::server::{AppState, api_key, authorize, err, LimitQuery}; use crate::replication::Replicator; @@ -143,8 +145,25 @@ pub struct ReplicateQuery { pub(crate) async fn replicate( State(state): State, + headers: HeaderMap, + ConnectInfo(addr): ConnectInfo, Query(q): Query, ) -> impl IntoResponse { + // The caller picks `peer_url` and this node then fetches it and stores the + // response. Unauthenticated, that is both a server-side request forgery + // primitive (probe any host the node can reach) and a way to fill the store + // with arbitrary attacker-supplied content. + if let Err(e) = authorize( + &state.auth, + state.user_auth.as_deref(), + &headers, + addr, + AccessLevel::Write, + &state.data_dir, + ) { + return err(StatusCode::UNAUTHORIZED, &e.to_string()).into_response(); + } + if q.peer_url.is_empty() { return err(StatusCode::BAD_REQUEST, "peer_url is required").into_response(); } diff --git a/earthgrid-core/src/routes/federation.rs b/earthgrid-core/src/routes/federation.rs index 45520dc..257222f 100644 --- a/earthgrid-core/src/routes/federation.rs +++ b/earthgrid-core/src/routes/federation.rs @@ -1,12 +1,14 @@ use axum::{ - extract::{Query, State}, + extract::{ConnectInfo, Query, State}, http::{HeaderMap, StatusCode}, response::IntoResponse, Json, }; use serde::Deserialize; +use std::net::SocketAddr; -use crate::server::{AppState, api_key, err}; +use crate::auth::AccessLevel; +use crate::server::{AppState, api_key, authorize, err}; use crate::peers::NodeInfo; @@ -46,8 +48,25 @@ pub(crate) async fn list_peers(State(state): State) -> Json, + headers: HeaderMap, + ConnectInfo(addr): ConnectInfo, Query(q): Query, ) -> impl IntoResponse { + // A registered peer is not passive: the heartbeat/gossip loop contacts it + // every 60s and the auto-replication loop pulls items and chunks from it + // every 5 minutes. Letting anyone register a URL turns one unauthenticated + // request into a standing data-ingestion channel from an attacker's server. + if let Err(e) = authorize( + &state.auth, + state.user_auth.as_deref(), + &headers, + addr, + AccessLevel::Write, + &state.data_dir, + ) { + return err(StatusCode::UNAUTHORIZED, &e.to_string()).into_response(); + } + if q.url.is_empty() { return err(StatusCode::BAD_REQUEST, "url is required").into_response(); } diff --git a/earthgrid-core/src/server.rs b/earthgrid-core/src/server.rs index 0aea8a9..6f4d1c6 100644 --- a/earthgrid-core/src/server.rs +++ b/earthgrid-core/src/server.rs @@ -969,13 +969,20 @@ pub async fn serve( }; let catalog = evict_catalog.lock().await; let mut store = evict_store.lock().await; - match crate::eviction::evict_with_beacon_url( - &catalog, - &mut store, - limit_gb, - beacon_db.as_deref(), - evict_beacon_url.as_deref(), - ) { + // Eviction is synchronous and slow: an HTTP call to the + // beacon plus SQLite writes and per-chunk file deletes. + // `block_in_place` hands the work to a blocking thread + // instead of stalling an async worker, while keeping the + // two lock guards in scope. + match tokio::task::block_in_place(|| { + crate::eviction::evict_with_beacon_url( + &catalog, + &mut store, + limit_gb, + beacon_db.as_deref(), + evict_beacon_url.as_deref(), + ) + }) { Ok(result) => { if result.items_deleted > 0 { eprintln!( From 984ad4d65d3b434a06d42b8de884bd9f2756dd6b Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 6 Aug 2026 10:57:19 +0000 Subject: [PATCH 2/2] fix: give beacon federation its own key, and fail closed without one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit gated /api/beacon/ws on EARTHGRID_API_KEY. That was wrong in two ways. 1. It reused the node's write credential. The dialer has to transmit whatever the listener checks, so every beacon you federated with would learn your EARTHGRID_API_KEY and could then drive your whole node API — /api/fetch, /api/replicate, the fetch queue. A federated peer needs exactly one capability, registry sync, so it now gets exactly one credential: EARTHGRID_FEDERATION_KEY, sent in x-earthgrid-federation-key. 2. It failed open. The check only engaged when a key happened to be set, so a beacon in the default keyless configuration still accepted any client — and a client on that socket can add nodes to the registry and, via NodePruned, delete any node by ID. Federation now fails closed: with no key configured the endpoint returns 503 and spawn_peer_connections refuses to dial. An unconfigured beacon federates with nobody instead of with everybody. Key comparison is constant-time. Verified against two live beacons: no key configured — 503 for no credential, for a guessed one, and for the old x-api-key header key configured — 401 for no credential, 401 for a wrong key, 101 Switching Protocols for the right one docs/beacon-guide.md documents the new variable, that it must match across beacons, that it is not the grid key, and that peer URLs should be https since the key travels in a handshake header. Tests: 158 passing. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Qjes4orjaT3k6ShNXj5Axm --- docs/beacon-guide.md | 28 +++- earthgrid-core/src/auth.rs | 2 +- earthgrid-core/src/beacon.rs | 3 + earthgrid-core/src/beacon_federation.rs | 174 ++++++++++++++++++------ earthgrid-core/src/server.rs | 1 + 5 files changed, 162 insertions(+), 46 deletions(-) diff --git a/docs/beacon-guide.md b/docs/beacon-guide.md index 142cc75..9bb5462 100644 --- a/docs/beacon-guide.md +++ b/docs/beacon-guide.md @@ -88,14 +88,38 @@ curl http://:8400/node-info Beacons federate with each other via **real-time WebSocket connections** — sharing their node registries so every beacon has the complete grid view. +Federation requires a shared secret, `EARTHGRID_FEDERATION_KEY`. **Every beacon +that federates must set the same value.** + ```bash +# Generate one secret and copy it to every federated beacon +openssl rand -hex 32 + +# On each beacon: +export EARTHGRID_FEDERATION_KEY= + # Add a peer beacon -export EARTHGRID_BEACON_PEERS=http://other-beacon.example.com:8400 +export EARTHGRID_BEACON_PEERS=https://other-beacon.example.com:8400 # Or add multiple (comma-separated) -export EARTHGRID_BEACON_PEERS=http://beacon1.example.com:8400,http://beacon2.example.com:8400 +export EARTHGRID_BEACON_PEERS=https://beacon1.example.com:8400,https://beacon2.example.com:8400 ``` +Notes: + +- **Without the key, federation is off.** `/api/beacon/ws` returns 503 and no + outbound connections are attempted. This is deliberate: a peer on that socket + can add nodes to your registry *and delete them*, so an unconfigured beacon + federates with nobody rather than with anybody. +- **The federation key is separate from `EARTHGRID_API_KEY` on purpose.** A + federated peer gets registry sync and nothing else — it cannot write to your + node's API. Do not reuse your grid key here. +- **Use `https://` peer URLs.** The key travels in a request header on the + WebSocket handshake; over plain `http://` it is sent in the clear. +- Mismatched keys are not silent: the dialer logs + `Federation: failed to connect to ` once a minute and retries with + backoff. + ### How it works 1. On startup, each beacon connects to its peers via WebSocket (`/beacon/ws`) diff --git a/earthgrid-core/src/auth.rs b/earthgrid-core/src/auth.rs index 4ae35ff..cfa9559 100644 --- a/earthgrid-core/src/auth.rs +++ b/earthgrid-core/src/auth.rs @@ -83,7 +83,7 @@ impl AuthConfig { } /// Constant-time string comparison to prevent timing attacks on key checks. -fn constant_time_eq_str(a: &str, b: &str) -> bool { +pub(crate) fn constant_time_eq_str(a: &str, b: &str) -> bool { if a.len() != b.len() { return false; } diff --git a/earthgrid-core/src/beacon.rs b/earthgrid-core/src/beacon.rs index 4860cc3..2c43ecc 100644 --- a/earthgrid-core/src/beacon.rs +++ b/earthgrid-core/src/beacon.rs @@ -975,6 +975,9 @@ pub struct BeaconState { pub registry: Arc>, pub federation: Option, pub auth: AuthConfig, + /// Credential for beacon-to-beacon federation. Separate from `auth` on + /// purpose: a federated peer gets registry sync only, not the node API. + pub federation_auth: crate::beacon_federation::FederationAuth, } async fn register_node( diff --git a/earthgrid-core/src/beacon_federation.rs b/earthgrid-core/src/beacon_federation.rs index e0a763d..524b65e 100644 --- a/earthgrid-core/src/beacon_federation.rs +++ b/earthgrid-core/src/beacon_federation.rs @@ -85,37 +85,85 @@ impl FederationState { // WebSocket handler (inbound connections from peer beacons) // --------------------------------------------------------------------------- +/// Header carrying the federation credential on the WebSocket handshake. +pub const FEDERATION_KEY_HEADER: &str = "x-earthgrid-federation-key"; + +/// The credential two beacons share in order to federate. +/// +/// Deliberately *not* the node's `EARTHGRID_API_KEY`. A federated peer needs +/// exactly one capability — beacon registry sync — and nothing else. Reusing +/// the grid key would hand every peer write access to the whole node API +/// (`/api/fetch`, `/api/replicate`, the fetch queue, …), because the dialer +/// must transmit whatever key the listener checks. +#[derive(Debug, Clone, Default)] +pub struct FederationAuth { + key: String, +} + +impl FederationAuth { + /// Read `EARTHGRID_FEDERATION_KEY` from the environment. + pub fn from_env() -> Self { + Self { + key: std::env::var("EARTHGRID_FEDERATION_KEY").unwrap_or_default(), + } + } + + /// Whether a federation key has been configured. + pub fn is_configured(&self) -> bool { + !self.key.is_empty() + } + + /// The configured key, for the outbound dialer. + pub fn key(&self) -> &str { + &self.key + } + + /// Whether `presented` matches the configured key. + /// + /// Returns false when nothing is configured: federation **fails closed**, + /// so an unconfigured beacon refuses to federate rather than accepting + /// everyone. Comparison is constant-time. + pub fn verify(&self, presented: Option<&str>) -> bool { + if !self.is_configured() { + return false; + } + presented.is_some_and(|p| crate::auth::constant_time_eq_str(p, &self.key)) + } +} + /// GET /beacon/ws — upgrade to WebSocket for federation sync. /// -/// Requires the shared grid key when one is configured. The peer that opens -/// this socket is not a passive reader: `apply_remote_event` lets it upsert -/// nodes into the registry and, via `NodePruned`, **delete** any node by ID. -/// Without a check, anyone who can reach the port could wipe a beacon's -/// registry or fill it with fabricated nodes. +/// A peer on this socket is not a passive reader: `apply_remote_event` lets it +/// upsert nodes into the registry and, via `NodePruned`, **delete** any node by +/// ID. Previously the upgrade was unauthenticated, so anyone who could reach +/// the port could wipe a beacon's registry or fill it with fabricated nodes. /// -/// When no key is configured the socket stays open, matching the rest of the -/// node's open-mode behaviour — so set `EARTHGRID_API_KEY` on federated -/// beacons, and set it to the *same* value on each, or they cannot peer. +/// Fails closed. With no `EARTHGRID_FEDERATION_KEY` configured the endpoint is +/// disabled outright — an unconfigured beacon federates with nobody instead of +/// with everybody. pub async fn ws_handler( ws: WebSocketUpgrade, State(state): State, headers: HeaderMap, ) -> impl IntoResponse { - if state.auth.is_enabled() { - let key = headers - .get("x-api-key") - .and_then(|v| v.to_str().ok()) - .or_else(|| { - headers - .get("authorization") - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.strip_prefix("Bearer ")) - }); - if state.auth.check_write(key).is_err() { - warn!("Federation: rejecting unauthenticated WebSocket upgrade"); - return (StatusCode::UNAUTHORIZED, "federation key required").into_response(); - } + if !state.federation_auth.is_configured() { + warn!("Federation: WS refused — EARTHGRID_FEDERATION_KEY is not set"); + return ( + StatusCode::SERVICE_UNAVAILABLE, + "beacon federation is disabled: set EARTHGRID_FEDERATION_KEY to enable it", + ) + .into_response(); + } + + let presented = headers + .get(FEDERATION_KEY_HEADER) + .and_then(|v| v.to_str().ok()); + + if !state.federation_auth.verify(presented) { + warn!("Federation: rejecting WS upgrade with missing or invalid federation key"); + return (StatusCode::UNAUTHORIZED, "invalid federation key").into_response(); } + ws.on_upgrade(move |socket| handle_peer_connection(socket, state)) .into_response() } @@ -250,6 +298,15 @@ fn upsert_if_newer(registry: &crate::beacon::BeaconRegistry, node: &BeaconNode) /// Spawn background tasks that connect to each peer beacon via WebSocket. pub fn spawn_peer_connections(state: BeaconState, peer_urls: Vec) { + if !state.federation_auth.is_configured() { + warn!( + "Federation: {} peer(s) configured but EARTHGRID_FEDERATION_KEY is not set — \ + not connecting. Set the same key on every federated beacon.", + peer_urls.len() + ); + return; + } + for url in peer_urls { let state = state.clone(); tokio::spawn(async move { @@ -259,23 +316,25 @@ pub fn spawn_peer_connections(state: BeaconState, peer_urls: Vec) { } /// Build the WebSocket handshake request for a peer beacon, attaching the -/// shared grid key as `x-api-key` when one is configured. +/// shared federation key. fn build_federation_request( ws_url: &str, - api_key: &str, + federation_key: &str, ) -> std::result::Result { use tokio_tungstenite::tungstenite::client::IntoClientRequest; + if federation_key.is_empty() { + return Err("no federation key configured".to_string()); + } + let mut request = ws_url .into_client_request() .map_err(|e| format!("invalid websocket url: {e}"))?; - if !api_key.is_empty() { - let value = api_key - .parse() - .map_err(|_| "grid key is not a valid header value".to_string())?; - request.headers_mut().insert("x-api-key", value); - } + let value = federation_key + .parse() + .map_err(|_| "federation key is not a valid header value".to_string())?; + request.headers_mut().insert(FEDERATION_KEY_HEADER, value); Ok(request) } @@ -294,9 +353,9 @@ async fn connect_to_peer_loop(state: BeaconState, peer_url: String) { loop { info!("Federation: connecting to peer {}", ws_url); - // Present the shared grid key so the peer's `ws_handler` accepts us. - // Rebuilt each attempt because `connect_async` consumes the request. - let request = match build_federation_request(&ws_url, &state.auth.api_key) { + // Present the shared federation key so the peer's `ws_handler` accepts + // us. Rebuilt each attempt because `connect_async` consumes the request. + let request = match build_federation_request(&ws_url, state.federation_auth.key()) { Ok(r) => r, Err(e) => { warn!("Federation: cannot build request for {}: {}", ws_url, e); @@ -478,23 +537,27 @@ mod tests { } #[test] - fn federation_request_carries_grid_key() { - let req = build_federation_request("ws://beacon.example/api/beacon/ws", "grid-secret") + fn federation_request_carries_federation_key() { + let req = build_federation_request("ws://beacon.example/api/beacon/ws", "fed-secret") .expect("request should build"); assert_eq!( - req.headers().get("x-api-key").map(|v| v.to_str().unwrap()), - Some("grid-secret"), - "dialer must present the key the peer's ws_handler now requires" + req.headers() + .get(FEDERATION_KEY_HEADER) + .map(|v| v.to_str().unwrap()), + Some("fed-secret"), + "dialer must present the key the peer's ws_handler requires" + ); + assert!( + req.headers().get("x-api-key").is_none(), + "the node's grid API key must never be sent to a federated peer" ); } #[test] - fn federation_request_omits_empty_key() { - let req = build_federation_request("ws://beacon.example/api/beacon/ws", "") - .expect("request should build"); + fn federation_request_requires_a_key() { assert!( - req.headers().get("x-api-key").is_none(), - "keyless deployments must dial exactly as before" + build_federation_request("ws://beacon.example/api/beacon/ws", "").is_err(), + "dialing without a federation key must fail rather than connect anonymously" ); } @@ -502,4 +565,29 @@ mod tests { fn federation_request_rejects_bad_url() { assert!(build_federation_request("not a url", "k").is_err()); } + + /// Federation must fail closed: an unconfigured beacon federates with + /// nobody, rather than accepting every anonymous client. + #[test] + fn unconfigured_federation_auth_rejects_everything() { + let auth = FederationAuth::default(); + assert!(!auth.is_configured()); + assert!(!auth.verify(None)); + assert!(!auth.verify(Some(""))); + assert!(!auth.verify(Some("anything"))); + } + + #[test] + fn configured_federation_auth_accepts_only_the_key() { + let auth = FederationAuth { key: "correct-horse".to_string() }; + assert!(auth.is_configured()); + assert!(auth.verify(Some("correct-horse"))); + + assert!(!auth.verify(None)); + assert!(!auth.verify(Some(""))); + assert!(!auth.verify(Some("wrong"))); + assert!(!auth.verify(Some("correct-horse ")), "no trimming"); + assert!(!auth.verify(Some("correct-hors")), "prefix must not pass"); + assert!(!auth.verify(Some("correct-horse-battery")), "extension must not pass"); + } } diff --git a/earthgrid-core/src/server.rs b/earthgrid-core/src/server.rs index 6f4d1c6..c5f41e5 100644 --- a/earthgrid-core/src/server.rs +++ b/earthgrid-core/src/server.rs @@ -703,6 +703,7 @@ pub async fn serve( registry: Arc::new(Mutex::new(registry)), federation: Some(federation), auth: auth.clone(), + federation_auth: crate::beacon_federation::FederationAuth::from_env(), }; app = app.merge(beacon_router(beacon_state.clone())); println!("🔦 Beacon registry enabled ({}) [beacon_id={}]", beacon_db_path.display(), &beacon_id[..8]);