summaryrefslogtreecommitdiffstats
path: root/src/api
diff options
context:
space:
mode:
Diffstat (limited to 'src/api')
-rw-r--r--src/api/auth.rs25
-rw-r--r--src/api/document.rs5
-rw-r--r--src/api/mod.rs37
-rw-r--r--src/api/peers.rs2
-rw-r--r--src/api/stats.rs4
-rw-r--r--src/api/stats/prometheus.rs1
-rw-r--r--src/api/store.rs12
7 files changed, 67 insertions, 19 deletions
diff --git a/src/api/auth.rs b/src/api/auth.rs
index 70b1af5..de4220c 100644
--- a/src/api/auth.rs
+++ b/src/api/auth.rs
@@ -72,14 +72,18 @@ fn parse_basic(headers: &HeaderMap) -> Option<(String, String)> {
Some((user.to_owned(), pass.to_owned()))
}
-async fn authenticate(state: &AuthState, headers: &HeaderMap) -> bool {
- let Some((user, pass)) = parse_basic(headers) else {
- return false;
- };
- let Some(hash) = state.users.get(&user).cloned() else {
- return false;
- };
+/// The user a request authenticated as (a request extension, set by [`basic_auth`]).
+#[derive(Clone, Debug)]
+pub(crate) struct AuthenticatedUser(pub(crate) String);
+/// The user, if the credentials are valid.
+async fn authenticate(state: &AuthState, headers: &HeaderMap) -> Option<String> {
+ let (user, pass) = parse_basic(headers)?;
+ let hash = state.users.get(&user).cloned()?;
+ verify(state, &user, pass, hash).await.then_some(user)
+}
+
+async fn verify(state: &AuthState, user: &str, pass: String, hash: String) -> bool {
let key = *blake3::hash(format!("{user}\0{pass}\0{hash}").as_bytes()).as_bytes();
// Fast path. The shard guard lives only until the end of this `match`;
@@ -124,10 +128,10 @@ async fn authenticate(state: &AuthState, headers: &HeaderMap) -> bool {
pub async fn basic_auth(
State(state): State<AuthState>,
headers: HeaderMap,
- request: Request<Body>,
+ mut request: Request<Body>,
next: axum::middleware::Next,
) -> Response {
- if !authenticate(&state, &headers).await {
+ let Some(user) = authenticate(&state, &headers).await else {
sleep(Duration::from_millis(1500)).await;
return (
StatusCode::UNAUTHORIZED,
@@ -135,7 +139,8 @@ pub async fn basic_auth(
"Unauthorized",
)
.into_response();
- }
+ };
+ request.extensions_mut().insert(AuthenticatedUser(user));
next.run(request).await
}
diff --git a/src/api/document.rs b/src/api/document.rs
index 303f5c9..6c10c38 100644
--- a/src/api/document.rs
+++ b/src/api/document.rs
@@ -22,6 +22,7 @@ pub(crate) async fn get_document(State(state): State<AppState>) -> Response {
let swarm_id = state.swarm_id().await.unwrap_or_default();
let start = Instant::now();
+ let version = crate::swarm::stats::document_version(&state);
let doc = std::sync::Arc::clone(&state.doc);
// Both steps take time proportional to the document: off the async threads, and
// the lock only for the copy.
@@ -42,7 +43,9 @@ pub(crate) async fn get_document(State(state): State<AppState>) -> Response {
}
};
let took = start.elapsed();
- state.stats.saved_size_measured(bytes.len() as u64, took);
+ state
+ .stats
+ .saved_size_measured(bytes.len() as u64, took, version);
info!(
bytes = bytes.len(),
took_ms = took.as_millis() as u64,
diff --git a/src/api/mod.rs b/src/api/mod.rs
index 3233b16..1fba48b 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -9,7 +9,8 @@ mod stats;
mod store;
use axum::{
- Router,
+ Json, Router,
+ extract::State,
http::{
HeaderValue, StatusCode,
header::{CONTENT_TYPE, ETAG},
@@ -23,7 +24,7 @@ use tracing::{error, warn};
use utoipa::OpenApi;
use utoipa_swagger_ui::SwaggerUi;
-use crate::api::auth::{AuthState, basic_auth};
+use crate::api::auth::{AuthState, AuthenticatedUser, basic_auth};
use crate::document::store::StoreError;
use crate::state::AppState;
@@ -46,6 +47,7 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) ->
)
.route("/admin/stats", get(stats::get_stats))
.route("/admin/document", get(document::get_document))
+ .route("/admin/config", get(get_config))
.merge(crate::remake::router::<AppState>());
if debug_api {
@@ -63,6 +65,12 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) ->
.with_state(state)
}
+/// `GET /admin/config`: the effective configuration with each value's source;
+/// imported values are left out.
+async fn get_config(State(state): State<AppState>) -> Json<JsonValue> {
+ Json((*state.config).clone())
+}
+
// --- shared helpers ----------------------------------------------------------
pub(crate) fn store_status(error: StoreError, context: &str) -> StatusCode {
@@ -96,6 +104,31 @@ fn parse_key_path(key: &str) -> Result<Vec<String>, StatusCode> {
Ok(segments)
}
+/// A path for a write. Protocol data (`_continuum`: remake records, members, forget
+/// records, epoch metadata) is what every node acts on, so a write there acts on the
+/// whole swarm: only users listed in `--protocol-writers` may, and each such write is
+/// logged. Epoch metadata (and `_continuum` as a whole, which contains it) is never
+/// writable: the protocol reads it at the snapshot change, so an edit only misleads
+/// whoever reads it later. `403` otherwise.
+fn parse_write_path(
+ state: &AppState,
+ user: &AuthenticatedUser,
+ key: &str,
+ method: &str,
+) -> Result<Vec<String>, StatusCode> {
+ let path = parse_key_path(key)?;
+ if path.first().map(String::as_str) != Some(crate::remake::snapshot::META_ROOT) {
+ return Ok(path);
+ }
+ let epochs = path.len() < 2 || path[1] == crate::remake::snapshot::EPOCHS;
+ if epochs || !state.protocol_writers.contains(&user.0) {
+ warn!(user = %user.0, method, path = %key, "Refused a write to protocol data");
+ return Err(StatusCode::FORBIDDEN);
+ }
+ warn!(user = %user.0, method, path = %key, "Protocol data written through the store API; this acts on the whole swarm");
+ Ok(path)
+}
+
fn response_for_value(
value: &JsonValue,
accept: Option<&HeaderValue>,
diff --git a/src/api/peers.rs b/src/api/peers.rs
index 7caaaf3..919f187 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -182,7 +182,7 @@ pub(crate) async fn list_peers(
}
peers.sort_by(|a, b| a.public_key.cmp(&b.public_key));
- let local_heads = state.doc.write().await.doc.get_heads().len();
+ let local_heads = state.doc.read().await.heads().len();
Ok(Json(PeerListResponse {
local_alias: state.alias.clone(),
local_epoch: local_epoch.into(),
diff --git a/src/api/stats.rs b/src/api/stats.rs
index bc925ff..838cc0f 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -134,8 +134,8 @@ async fn collect(state: &AppState) -> StatsResponse {
let node = state.stats.snapshot();
let (heads, tier) = {
- let mut replica = state.doc.write().await;
- let heads = replica.doc.get_heads().len();
+ let replica = state.doc.read().await;
+ let heads = replica.heads().len();
(heads, members::tier(&replica.doc, state.local_id()))
};
let saved = state.stats.saved_size();
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index d29207c..f68ca85 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -130,6 +130,7 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou
(direction, "sync", by.sync),
(direction, "ping", by.ping),
(direction, "pong", by.pong),
+ (direction, "blob", by.blob),
]
};
let mut series = Vec::new();
diff --git a/src/api/store.rs b/src/api/store.rs
index d6ff306..aa9465c 100644
--- a/src/api/store.rs
+++ b/src/api/store.rs
@@ -4,6 +4,7 @@ use std::io::Error as IoError;
use automerge::ChangeHash;
use axum::{
+ Extension,
body::{Body, Bytes},
extract::{Path, Query, State},
http::{
@@ -17,7 +18,8 @@ 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 super::auth::AuthenticatedUser;
+use super::{parse_key_path, parse_write_path, response_for_value, store_status};
use crate::document::store::{self, StoreError};
use crate::state::{AppState, ReadSource};
use crate::swarm::sync::paths_related;
@@ -194,6 +196,7 @@ fn mark_source(
responses(
(status = 200, description = "Value written successfully"),
(status = 400, description = "Invalid path, malformed JSON, invalid UTF-8, or unsupported JSON value"),
+ (status = 403, description = "Protocol data (_continuum) and the user isn't a protocol writer, or epoch metadata"),
(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"),
@@ -202,6 +205,7 @@ fn mark_source(
)]
pub(crate) async fn put_store_entry(
State(state): State<AppState>,
+ Extension(user): Extension<AuthenticatedUser>,
Path(key): Path<String>,
headers: HeaderMap,
body: Bytes,
@@ -210,7 +214,7 @@ pub(crate) async fn put_store_entry(
return Err(StatusCode::SERVICE_UNAVAILABLE);
}
- let path = parse_key_path(&key)?;
+ let path = parse_write_path(&state, &user, &key, "PUT")?;
let content_type = headers
.get(CONTENT_TYPE)
@@ -251,6 +255,7 @@ pub(crate) async fn put_store_entry(
responses(
(status = 204, description = "Value deleted successfully"),
(status = 400, description = "Invalid store path"),
+ (status = 403, description = "Protocol data (_continuum) and the user isn't a protocol writer, or epoch metadata"),
(status = 404, description = "Value does not exist"),
(status = 500, description = "Deletion or synchronization notification failed"),
(status = 503, description = "Node is not ready")
@@ -259,10 +264,11 @@ pub(crate) async fn put_store_entry(
#[instrument(skip(state), fields(path, prune = query.prune))]
pub(crate) async fn delete_store_entry(
State(state): State<AppState>,
+ Extension(user): Extension<AuthenticatedUser>,
Path(key): Path<String>,
Query(query): Query<DeleteQuery>,
) -> Result<StatusCode, StatusCode> {
- let path = parse_key_path(&key)?;
+ let path = parse_write_path(&state, &user, &key, "DELETE")?;
tracing::Span::current().record("path", path.join("/").as_str());
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-25 09:12:17 +0000