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 09ed6f6..524b65e 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}; @@ -84,12 +85,87 @@ 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. +/// +/// 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. +/// +/// 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.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() } async fn handle_peer_connection(socket: WebSocket, state: BeaconState) { @@ -222,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 { @@ -230,6 +315,30 @@ pub fn spawn_peer_connections(state: BeaconState, peer_urls: Vec) { } } +/// Build the WebSocket handshake request for a peer beacon, attaching the +/// shared federation key. +fn build_federation_request( + ws_url: &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}"))?; + + 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) +} + /// 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 +353,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 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); + 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 +535,59 @@ mod tests { assert!(json.contains("full_sync")); assert!(json.contains("beacon-1")); } + + #[test] + 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(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_requires_a_key() { + assert!( + build_federation_request("ws://beacon.example/api/beacon/ws", "").is_err(), + "dialing without a federation key must fail rather than connect anonymously" + ); + } + + #[test] + 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/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..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]); @@ -969,13 +970,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!(