From d3a9d5d3ebff98e342dff1df0ebad86f346719f5 Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 19 Oct 2026 23:44:20 +0200 Subject: initial commit --- src/api/debug.rs | 41 +++++++ src/api/mod.rs | 160 +++++++++++++++++++++++++ src/api/peers.rs | 161 +++++++++++++++++++++++++ src/api/store.rs | 359 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 721 insertions(+) create mode 100644 src/api/debug.rs create mode 100644 src/api/mod.rs create mode 100644 src/api/peers.rs create mode 100644 src/api/store.rs (limited to 'src/api') diff --git a/src/api/debug.rs b/src/api/debug.rs new file mode 100644 index 0000000..3c24f62 --- /dev/null +++ b/src/api/debug.rs @@ -0,0 +1,41 @@ +//! /debug/* handlers (only routed with --debug-api). + +use std::collections::HashSet; + +use axum::{body::Bytes, extract::State, http::StatusCode, Json}; +use tracing::{info, warn}; + +use crate::firewall::FirewallRules; +use crate::state::AppState; + +pub(crate) async fn get_debug_firewall(State(state): State) -> Json { + Json(state.firewall.rules()) +} + +pub(crate) async fn put_debug_firewall( + State(state): State, + Json(rules): Json, +) -> StatusCode { + info!(inbound = ?rules.inbound, outbound = ?rules.outbound, "Firewall rules replaced"); + state.firewall.set_rules(rules); + StatusCode::NO_CONTENT +} + +/// Compatibility for the existing harness: block a set of peers in both directions. +/// Parses the body without requiring a Content-Type header, as before. +pub(crate) async fn put_debug_blacklist(State(state): State, body: Bytes) -> StatusCode { + match serde_json::from_slice::>(&body) { + Ok(peers) => { + info!(peers = ?peers, "Blacklist replaced (both directions)"); + state.firewall.set_rules(FirewallRules { + inbound: peers.clone(), + outbound: peers, + }); + StatusCode::NO_CONTENT + } + Err(error) => { + warn!(error = %error, "Invalid blacklist body"); + StatusCode::BAD_REQUEST + } + } +} diff --git a/src/api/mod.rs b/src/api/mod.rs new file mode 100644 index 0000000..8038f16 --- /dev/null +++ b/src/api/mod.rs @@ -0,0 +1,160 @@ +//! HTTP API: router, OpenAPI document and shared response helpers. + +mod debug; +mod peers; +mod store; + +use axum::{ + http::{ + header::{CONTENT_TYPE, ETAG}, + HeaderValue, StatusCode, + }, + response::{IntoResponse, Response}, + routing::{delete, get, put}, + Router, +}; +use serde_json::Value as JsonValue; +use sha2::{Digest, Sha256}; +use tracing::{error, warn}; +use utoipa::OpenApi; +use utoipa_swagger_ui::SwaggerUi; + +use crate::auth::{basic_auth, AuthState}; +use crate::state::AppState; +use crate::store::StoreError; + +/// Builds the complete HTTP router. Everything except Swagger is authenticated. +pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) -> Router { + let mut api: Router = Router::new() + .route("/store", get(store::get_store_root)) + .route( + "/store/{*key}", + get(store::get_store_entry) + .put(store::put_store_entry) + .delete(store::delete_store_entry), + ) + .route("/stream/{*key}", get(store::stream_store_entry)) + .route("/peers", get(peers::list_peers)) + .route("/peers/{peer}", delete(peers::delete_peer)); + + if debug_api { + warn!("Debug API enabled"); + api = api + .route( + "/debug/firewall", + get(debug::get_debug_firewall).put(debug::put_debug_firewall), + ) + .route("/debug/blacklist", put(debug::put_debug_blacklist)); + } + + api.layer(axum::middleware::from_fn_with_state(auth_state, basic_auth)) + .merge(SwaggerUi::new("/swagger-ui").url("/openapi.json", ApiDoc::openapi())) + .with_state(state) +} + +// --- shared helpers ---------------------------------------------------------- + +pub(crate) fn store_status(error: StoreError, context: &str) -> StatusCode { + match error { + StoreError::Conflict => StatusCode::CONFLICT, + StoreError::NotFound => StatusCode::NOT_FOUND, + other => { + error!(error = %format!("{other:#}"), "{context}"); + StatusCode::INTERNAL_SERVER_ERROR + } + } +} + +fn etag_for(bytes: &[u8]) -> String { + let digest = Sha256::digest(bytes); + let hex = digest + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + + format!("\"{hex}\"") +} + +fn parse_key_path(key: &str) -> Result, StatusCode> { + let segments: Vec = key.split('/').map(str::to_owned).collect(); + + if segments.is_empty() || segments.iter().any(|segment| segment.is_empty()) { + return Err(StatusCode::BAD_REQUEST); + } + + Ok(segments) +} + +fn response_for_value( + value: &JsonValue, + accept: Option<&HeaderValue>, +) -> Result { + let wants_text = accept + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| { + value + .split(',') + .any(|item| item.trim().starts_with("text/plain")) + }); + + let (content_type, body) = if wants_text { + let body = match value { + JsonValue::String(value) => value.clone(), + other => serde_json::to_string(other).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?, + }; + + ("text/plain; charset=utf-8", body) + } else { + ( + "application/json", + serde_json::to_string(value).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?, + ) + }; + + let etag = etag_for(body.as_bytes()); + + let mut response = body.into_response(); + + response + .headers_mut() + .insert(CONTENT_TYPE, HeaderValue::from_static(content_type)); + + response.headers_mut().insert( + ETAG, + HeaderValue::from_str(&etag).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?, + ); + + Ok(response) +} + +// --- openapi spec ------------------------------------------------------------ +// TODO spec is mostly out of sync + +#[derive(OpenApi)] +#[openapi( + paths( + store::get_store_root, + store::get_store_entry, + store::put_store_entry, + store::delete_store_entry, + peers::list_peers, + peers::delete_peer, + ), + components( + schemas( + peers::PeerStatus, + peers::PeerListResponse, + peers::PeerLivenessStatus, + ) + ), + tags( + (name = "Store", description = "Key-value store operations"), + (name = "Peers", description = "Peer discovery and retirement") + ), + info( + title = "Automerge Replication API", + version = "1.0.0", + description = "HTTP API for a replicated Automerge key-value store" + ) +)] +struct ApiDoc; diff --git a/src/api/peers.rs b/src/api/peers.rs new file mode 100644 index 0000000..b2658a5 --- /dev/null +++ b/src/api/peers.rs @@ -0,0 +1,161 @@ +//! /peers handlers. + +use std::sync::Arc; + +use axum::{ + extract::{Path, Query, State}, + http::StatusCode, + Json, +}; +use serde::{Deserialize, Serialize}; +use tokio::sync::Mutex; +use tokio::time::Instant; +use tracing::warn; +use utoipa::ToSchema; + +use super::store_status; +use crate::peer::forgotten_path; +use crate::state::{unix_millis, AppState, HelloState, Liveness, PeerMembership, PeerState}; +use crate::store; + +#[derive(Debug, Serialize, ToSchema)] +#[serde(rename_all = "lowercase")] +pub(crate) enum PeerLivenessStatus { + Active, + Dormant, +} + +#[derive(Debug, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub(crate) struct PeerStatus { + public_key: String, + validated_cluster: Option, + status: PeerLivenessStatus, + handshake_complete: bool, + last_seen_secs_ago: Option, + dormant_for_secs: Option, + has_pending_delivery: bool, +} + +#[derive(Debug, Serialize, ToSchema)] +#[serde(rename_all = "camelCase")] +pub(crate) struct PeerListResponse { + peers: Vec, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct DeletePeerQuery { + #[serde(default)] + force: bool, +} + +#[utoipa::path( + get, + path = "/peers", + tag = "Peers", + responses( + ( + status = 200, + description = "Lists known peers and their local replication status", + body = PeerListResponse + ) + ) +)] +pub(crate) async fn list_peers( + State(state): State, +) -> Result, StatusCode> { + let snapshot: Vec<(String, Arc>)> = state + .peer_states + .read() + .await + .iter() + .map(|(pk, m)| (pk.clone(), Arc::clone(m))) + .collect(); + + let now = Instant::now(); + let mut peers = Vec::with_capacity(snapshot.len()); + + for (public_key, peer_state_mutex) in snapshot { + let ps = peer_state_mutex.lock().await; + + let (status, dormant_for_secs) = match ps.liveness { + Liveness::Active => (PeerLivenessStatus::Active, None), + Liveness::Dormant { since } => ( + PeerLivenessStatus::Dormant, + Some(now.saturating_duration_since(since).as_secs()), + ), + }; + + peers.push(PeerStatus { + public_key, + validated_cluster: match &ps.membership { + PeerMembership::Unknown => None, + PeerMembership::Member { cluster_id } => Some(cluster_id.clone()), + }, + status, + handshake_complete: ps.hello_state == HelloState::Established, + last_seen_secs_ago: ps + .last_seen + .map(|t| now.saturating_duration_since(t).as_secs()), + dormant_for_secs, + has_pending_delivery: ps.pending_outbound_sync.is_some(), + }); + } + + peers.sort_by(|a, b| a.public_key.cmp(&b.public_key)); + Ok(Json(PeerListResponse { peers })) +} + +#[utoipa::path( + delete, + path = "/peers/{peer}", + tag = "Peers", + params( + ("peer" = String, Path, description = "Peer public key"), + ("force" = Option, Query, description = "Also forget a peer that is currently alive; it returns on its next message") + ), + responses( + (status = 204, description = "Cluster-wide forget recorded"), + (status = 400, description = "Not a public key, or the local node's own key"), + (status = 409, description = "Peer is alive; pass force=true"), + (status = 503, description = "Node is not ready") + ) +)] +pub(crate) async fn delete_peer( + State(state): State, + Path(peer): Path, + Query(query): Query, +) -> Result { + if !state.is_ready().await { + return Err(StatusCode::SERVICE_UNAVAILABLE); + } + if peer.is_empty() || peer == state.local_id() { + return Err(StatusCode::BAD_REQUEST); + } + + // Unknown locally is fine: other nodes may still know the peer. + if let Some(peer_state_mutex) = state.get_peer_state(&peer).await { + if peer_state_mutex + .lock() + .await + .is_alive(&state.timing, Instant::now()) + && !query.force + { + return Err(StatusCode::CONFLICT); + } + } + + let record = serde_json::json!({ "at": unix_millis(), "by": state.local_id() }); + + let peer_clone = peer.clone(); + state + .mutate_doc(move |doc| store::write_path(doc, &forgotten_path(&peer_clone), &record)) + .await + .map_err(|e| store_status(e, "Failed to write forget record"))?; + + if state.peer_states.write().await.remove(&peer).is_some() { + warn!(peer = %peer, "Peer forgotten cluster-wide by operator"); + } + + Ok(StatusCode::NO_CONTENT) +} diff --git a/src/api/store.rs b/src/api/store.rs new file mode 100644 index 0000000..dd5be94 --- /dev/null +++ b/src/api/store.rs @@ -0,0 +1,359 @@ +//! /store and /stream handlers. + +use std::io::Error as IoError; + +use axum::{ + body::{Body, Bytes}, + extract::{Path, Query, State}, + http::{ + header::{ACCEPT, CONTENT_TYPE}, + HeaderMap, HeaderValue, StatusCode, + }, + response::{IntoResponse, Response}, +}; +use serde::Deserialize; +use serde_json::Value as JsonValue; +use tokio::sync::broadcast; +use tracing::{debug, error, instrument}; + +use super::{parse_key_path, response_for_value, store_status}; +use crate::state::AppState; +use crate::store::{self, StoreError}; +use crate::sync::paths_related; + +#[derive(Debug, Default, Deserialize)] +pub(crate) struct ReadQuery { + #[serde(default)] + conflicts: bool, +} + +#[derive(Debug, Deserialize)] +pub(crate) struct DeleteQuery { + #[serde(default)] + prune: bool, +} + +fn conflicts_response( + value: JsonValue, + conflicts: Vec, +) -> Result { + let body = serde_json::json!({ "value": value, "conflicts": conflicts }); + response_for_value(&body, None) // always JSON +} + +#[utoipa::path( + get, + path = "/store", + tag = "Store", + params( + ("conflicts" = Option, Query, description = "Wrap the value and list every conflict resolved on the way"), + ), + responses( + (status = 200, description = "Returns the complete store as JSON or text, depending on Accept"), + (status = 503, description = "Node is not ready"), + (status = 500, description = "Failed to read or serialize the document") + ) +)] +pub(crate) async fn get_store_root( + State(state): State, + headers: HeaderMap, + Query(query): Query, +) -> Result { + let accept = headers.get(ACCEPT).cloned(); + let want_conflicts = query.conflicts; + + state + .render( + move |doc| { + if want_conflicts { + store::read_path_with_conflicts(doc, &Vec::new()) + .map(|found| found.map(|(value, conflicts)| (value, Some(conflicts)))) + } else { + store::read_path(doc, &Vec::new()).map(|found| found.map(|value| (value, None))) + } + }, + move |result| { + let (value, conflicts) = result + .map_err(|e| store_status(e, "Failed to read store path"))? + .unwrap_or_else(|| (JsonValue::Object(Default::default()), Some(Vec::new()))); + match conflicts { + Some(conflicts) => conflicts_response(value, conflicts), + None => response_for_value(&value, accept.as_ref()), + } + }, + ) + .await +} + +#[utoipa::path( + get, + path = "/store/{key}", + tag = "Store", + params( + ("key" = String, Path, description = "Slash-separated store path, for example settings/theme"), + ("conflicts" = Option, Query, description = "Wrap the value and list every conflict resolved on the way"), + ), + responses( + (status = 200, description = "Returns the value as JSON or text, depending on Accept"), + (status = 400, description = "Invalid store path"), + (status = 404, description = "Value does not exist"), + (status = 500, description = "Failed to read or serialize the value"), + (status = 503, description = "Node is not ready") + ) +)] +pub(crate) async fn get_store_entry( + State(state): State, + Path(key): Path, + headers: HeaderMap, + Query(query): Query, +) -> Result { + if !state.is_ready().await { + return Err(StatusCode::SERVICE_UNAVAILABLE); + } + + let path = parse_key_path(&key)?; + + let accept = headers.get(ACCEPT).cloned(); + let want_conflicts = query.conflicts; + + state + .render( + move |doc| { + if want_conflicts { + store::read_path_with_conflicts(doc, &path) + .map(|found| found.map(|(value, conflicts)| (value, Some(conflicts)))) + } else { + store::read_path(doc, &path).map(|found| found.map(|value| (value, None))) + } + }, + move |result| { + let (value, conflicts) = result + .map_err(|e| store_status(e, "Failed to read store path"))? + .ok_or(StatusCode::NOT_FOUND)?; + match conflicts { + Some(conflicts) => conflicts_response(value, conflicts), + None => response_for_value(&value, accept.as_ref()), + } + }, + ) + .await +} + +#[utoipa::path( + put, + path = "/store/{key}", + tag = "Store", + params( + ("key" = String, Path, description = "Slash-separated store path") + ), + request_body( + content = String, + description = "Value to write. Use application/json for JSON values or text/plain for strings.", + content_type = "application/json" + ), + responses( + (status = 200, description = "Value written successfully"), + (status = 400, description = "Invalid path, malformed JSON, invalid UTF-8, or unsupported JSON value"), + (status = 409, description = "Path conflicts with an existing non-map value"), + (status = 415, description = "Unsupported Content-Type"), + (status = 500, description = "Internal error or synchronization notification failure"), + (status = 503, description = "Node is not ready") + ) +)] +pub(crate) async fn put_store_entry( + State(state): State, + Path(key): Path, + headers: HeaderMap, + body: Bytes, +) -> Result { + if !state.is_ready().await { + return Err(StatusCode::SERVICE_UNAVAILABLE); + } + + let path = parse_key_path(&key)?; + + let content_type = headers + .get(CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .unwrap_or("application/json") + .split(';') + .next() + .unwrap_or("") + .trim(); + + let value = match content_type { + "application/json" => { + serde_json::from_slice::(&body).map_err(|_| StatusCode::BAD_REQUEST)? + } + "text/plain" => JsonValue::String( + String::from_utf8(body.to_vec()).map_err(|_| StatusCode::BAD_REQUEST)?, + ), + _ => return Err(StatusCode::UNSUPPORTED_MEDIA_TYPE), + }; + + let written = value.clone(); + state + .mutate_doc(move |doc| store::write_path(doc, &path, &written)) + .await + .map_err(|e| store_status(e, "Failed to write store value"))?; + + response_for_value(&value, headers.get(ACCEPT)) +} + +#[utoipa::path( + delete, + path = "/store/{key}", + tag = "Store", + params( + ("key" = String, Path, description = "Slash-separated store path"), + ("prune" = Option, Query, description = "Remove empty ancestor maps after deletion") + ), + responses( + (status = 204, description = "Value deleted successfully"), + (status = 400, description = "Invalid store path"), + (status = 404, description = "Value does not exist"), + (status = 500, description = "Deletion or synchronization notification failed"), + (status = 503, description = "Node is not ready") + ) +)] +#[instrument(skip(state), fields(path, prune = query.prune))] +pub(crate) async fn delete_store_entry( + State(state): State, + Path(key): Path, + Query(query): Query, +) -> Result { + let path = parse_key_path(&key)?; + + tracing::Span::current().record("path", path.join("/").as_str()); + + if !state.is_ready().await { + return Err(StatusCode::SERVICE_UNAVAILABLE); + } + + if path.is_empty() { + return Err(StatusCode::BAD_REQUEST); + } + + let prune = query.prune; + state + .mutate_doc(move |doc| store::delete_path(doc, &path, prune)) + .await + .map_err(|e| store_status(e, "Failed to delete store value"))?; + + Ok(StatusCode::NO_CONTENT) +} + +// --- streams ----------------------------------------------------------------- + +async fn render_traced( + state: &AppState, + path: Vec, +) -> Result<(Option, bool), StoreError> { + state + .render(move |doc| store::read_path_traced(doc, &path), |result| result) + .await +} + +pub(crate) async fn stream_store_entry( + State(state): State, + Path(key): Path, +) -> Result { + if !state.is_ready().await { + return Err(StatusCode::SERVICE_UNAVAILABLE); + } + + let path = parse_key_path(&key)?; + + // Subscribe first, so nothing between the initial read and the loop is missed. + let mut changes = state.change_events.subscribe(); + + let initial = render_traced(&state, path.clone()) + .await + .map_err(|e| store_status(e, "Failed to read initial stream value"))?; + + let stream_state = state.clone(); + let stream_key = key.clone(); + + let output = async_stream::stream! { + let (mut last_value, mut conflicted) = initial; + + match stream_event_line(&stream_key, last_value.as_ref()) { + Ok(line) => yield Ok::(line), + Err(error) => { + yield Err(error); + return; + } + } + + loop { + let relevant = match changes.recv().await { + // Conflicted paths can change without a patch on their path; re-read always. + Ok(set) => conflicted || set.paths.iter().any(|changed| paths_related(changed, &path)), + Err(broadcast::error::RecvError::Lagged(skipped)) => { + debug!(skipped, key = %stream_key, "Stream lagged; re-reading"); + true + } + Err(broadcast::error::RecvError::Closed) => break, + }; + + if !relevant { + continue; + } + + let (next_value, next_conflicted) = match render_traced(&stream_state, path.clone()).await { + Ok(result) => result, + Err(error) => { + error!(error = %format!("{error:#}"), key = %stream_key, "Failed to read updated stream value"); + yield Err(IoError::other("Failed to read updated store value")); + return; + } + }; + conflicted = next_conflicted; + + // A related change can still leave the resolved value unchanged. + if next_value == last_value { + continue; + } + + match stream_event_line(&stream_key, next_value.as_ref()) { + Ok(line) => yield Ok(line), + Err(error) => { + yield Err(error); + return; + } + } + last_value = next_value; + } + }; + + let mut response = Body::from_stream(output).into_response(); + response.headers_mut().insert( + CONTENT_TYPE, + HeaderValue::from_static("application/x-ndjson"), + ); + response.headers_mut().insert( + axum::http::header::CACHE_CONTROL, + HeaderValue::from_static("no-cache, no-transform"), + ); + Ok(response) +} + +fn stream_event_line(key: &str, value: Option<&JsonValue>) -> std::io::Result { + let mut event = serde_json::Map::new(); + + event.insert("key".to_owned(), JsonValue::String(key.to_owned())); + + match value { + Some(value) => { + event.insert("value".to_owned(), value.clone()); + } + None => { + event.insert("deleted".to_owned(), JsonValue::Bool(true)); + } + } + + let mut line = serde_json::to_vec(&JsonValue::Object(event)).map_err(IoError::other)?; + + line.push(b'\n'); + Ok(Bytes::from(line)) +} -- cgit v1.3.1