feat(security): auth-gate cluster routes, enforce GRANT/REVOKE, cap decompression, redact credentials

config.toml.example

@@ -62,6 +62,8 @@ replication_max_retries = 5
 replication_queue_depth = 8192
 replication_max_inflight_batches = 8
 replication_max_coalesce_body_bytes = 8388608
+# Max HTTP body for /internal/replicate (0 = auto: max(4× coalesce, server.max_body_size_bytes))
+# replicate_body_limit_bytes = 0
 replicate_receiver_queue_depth = 1024
 # Deprecated: receiver always uses a single ordered worker.
 replicate_receiver_workers = 1

docs/developer-guide/internals/replication-design.md

@@ -52,7 +52,7 @@ See [`flush_service`](../../../src/application/flush_service.rs): per-peer ack w
 
 ## Operational knobs
 
-`[cluster]` in `config.rs`: `replication_queue_depth`, `replication_max_inflight_batches`, `replication_max_coalesce_body_bytes`, `replicate_receiver_queue_depth`, `replication_truncate_stale_peer_multiplier`, etc.
+`[cluster]` in `config.rs`: `replication_queue_depth`, `replication_max_inflight_batches`, `replication_max_coalesce_body_bytes`, `replicate_body_limit_bytes`, `replicate_receiver_queue_depth`, `replication_truncate_stale_peer_multiplier`, etc.
 
 ## Flow control
 

docs/user-guide/authentication.md

@@ -29,8 +29,9 @@ The following are **not** wrapped in the user auth middleware when `auth.enabled
 | `GET` / `HEAD` `/health` | JSON health |
 | `GET` / `HEAD` `/health/ready` | Readiness (chDB probe) |
 | `GET` `/metrics` | Prometheus text |
+| `GET` / `DELETE` `/api/v1/statements` | Only when `statement_summary.require_auth = false` (default: **requires auth**) |
 
-`GET` / `DELETE` **`/api/v1/statements`** (statement summary) is also **unauthenticated** today. It only exposes recent query digests and timings when [statement summary](configuration.md#statement_summary) is enabled, but in locked-down environments you should still restrict access at the network (or proxy) layer.
+When `statement_summary.require_auth` is `true` (the default) and `[auth] enabled = true`, statement summary requires the same credentials as `/query`.
 
 ---
 
@@ -88,7 +89,7 @@ Run through `/query` (or `POST` with form `q=...`).
 
 **Admin flag:** the parser sets `admin` if the `CREATE USER` text contains `ALL PRIVILEGES` or `ADMIN` (case-insensitive). Only admin users can access **internal** cluster routes when auth is on.
 
-**Authorization model:** no per-database GRANT/REVOKE. `GRANT` / `REVOKE` are accepted as no-ops for compatibility. Plan around **admin** vs **non-admin** only.
+**Authorization model:** per-database `GRANT ALL ON <db> TO <user>` / `REVOKE ALL ON <db> FROM <user>` control write and query access for non-admin users. Non-admin users without a grant on a database cannot read or write it. Admin users bypass per-database checks. `GRANT ALL PRIVILEGES TO <user>` (no `ON` clause) promotes an existing user to admin; `REVOKE ALL PRIVILEGES FROM <user>` removes admin.
 
 ---
 

docs/user-guide/configuration.md

@@ -32,7 +32,7 @@ HTTP server settings.
 |-----|------|---------|-------------|
 | `bind_address` | string | `"0.0.0.0"` | Network interface to bind to |
 | `port` | integer | `8086` | HTTP listen port |
-| `max_body_size_bytes` | integer | `26214400` | Maximum request body size (25 MB) |
+| `max_body_size_bytes` | integer | `26214400` | Maximum request body size for client `/write` (25 MB). Also used when `[cluster] replicate_body_limit_bytes = 0` to compute the auto replicate HTTP cap (see [cluster](#cluster)). |
 | `request_timeout_secs` | integer | `30` | HTTP request timeout |
 | `query_timeout_secs` | integer | `30` | TimeseriesQL query execution timeout |
 | `max_concurrent_queries` | integer | `0` | Max concurrent TimeseriesQL executions; `0` = unlimited (bounded by work-stealing / resources). Use with single chDB session. |
@@ -126,6 +126,7 @@ Master-master peer-to-peer clustering with Raft consensus for schema mutations.
 | `replication_queue_depth` | integer | `8192` | Bounded outbound replication queue (ingest-sized batches) |
 | `replication_max_inflight_batches` | integer | `8` | Max concurrent outbound replication fan-out rounds |
 | `replication_max_coalesce_body_bytes` | integer | `8388608` | Max bytes for coalescing consecutive WAL batches (same db/rp/precision) |
+| `replicate_body_limit_bytes` | integer | `0` (auto) | Max HTTP body size for `/internal/replicate` and `/internal/replicate-mutation`. When `0`, resolves to `max(4 × replication_max_coalesce_body_bytes, server.max_body_size_bytes)` at startup (default **33554432** / 32 MiB with stock settings). Must be ≥ `replication_max_coalesce_body_bytes` or coalesced peer batches may receive HTTP 413. |
 | `replicate_receiver_queue_depth` | integer | `1024` | Bounded apply queue on the replicate receiver |
 | `replicate_receiver_workers` | integer | `1` | **Ignored.** Receiver uses a single ordered worker |
 | `replication_truncate_stale_peer_multiplier` | integer | `2` | When >0, peers with ack 0 and stale heartbeats are omitted from truncate barrier (× heartbeat interval) |
@@ -162,8 +163,9 @@ Query statement tracking for debugging and observability.
 |-----|------|---------|-------------|
 | `enabled` | boolean | `true` | Enable statement summary tracking |
 | `max_entries` | integer | `1000` | Max recent statements kept in the ring buffer |
+| `require_auth` | boolean | `true` | Require auth for `GET`/`DELETE` `/api/v1/statements` when `[auth] enabled = true` |
 
-When enabled, recently executed statements are accessible via `GET /api/v1/statements`.
+When enabled, recently executed statements are accessible via `GET /api/v1/statements`. Password literals in stored query samples are redacted.
 
 ---
 

hyperbytedb/Cargo.toml

@@ -81,6 +81,7 @@ flate2 = "1"
 
 # Auth
 argon2 = "0.5"
+base64 = "0.22"
 
 # Regex
 regex = "1"

hyperbytedb/src/adapters/auth.rs

@@ -1,7 +1,7 @@
 use async_trait::async_trait;
 use std::collections::HashMap;
 use std::hash::{Hash, Hasher};
-use std::sync::{Arc, RwLock};
+use std::sync::{Arc, LazyLock, RwLock};
 use std::time::Instant;
 
 use crate::domain::user::StoredUser;
@@ -11,6 +11,21 @@ use crate::ports::metadata::MetadataPort;
 
 const CREDENTIAL_CACHE_TTL_SECS: u64 = 60;
 
+/// Argon2 hash used for constant-time rejection of unknown usernames.
+static DUMMY_PASSWORD_HASH: LazyLock<String> = LazyLock::new(|| {
+    use argon2::Argon2;
+    use argon2::password_hash::{PasswordHasher, SaltString};
+    // Hardcoded valid salt and dummy password; constant-time dummy is a
+    // compile-time invariant, not user input.
+    #[allow(clippy::expect_used)]
+    let salt = SaltString::from_b64("dGVzdHNhbHR0ZXN0c2FsdA").expect("valid salt b64");
+    #[allow(clippy::expect_used)]
+    Argon2::default()
+        .hash_password(b"timing-constant-dummy", &salt)
+        .expect("dummy hash")
+        .to_string()
+});
+
 /// Fast non-crypto hash of the input password for cache keying.
 fn password_fingerprint(password: &str) -> u64 {
     let mut h = std::collections::hash_map::DefaultHasher::new();
@@ -52,7 +67,10 @@ impl AuthPort for MetadataAuthAdapter {
     ) -> Result<Option<StoredUser>, HyperbytedbError> {
         let stored = match self.metadata.get_user(username).await? {
             Some(s) => s,
-            None => return Ok(None),
+            None => {
+                let _ = verify_password(password, &DUMMY_PASSWORD_HASH);
+                return Ok(None);
+            }
         };
 
         let pw_fp = password_fingerprint(password);

hyperbytedb/src/adapters/http/auth_middleware.rs

@@ -82,32 +82,13 @@ fn extract_credentials(headers: &HeaderMap, query: &AuthParams) -> Option<(Strin
 }
 
 fn base64_decode(input: &str) -> Result<String, ()> {
-    let bytes = input.as_bytes();
-    let decoded = base64_decode_bytes(bytes).map_err(|_| ())?;
+    use base64::Engine;
+    let decoded = base64::engine::general_purpose::STANDARD
+        .decode(input.trim())
+        .map_err(|_| ())?;
     String::from_utf8(decoded).map_err(|_| ())
 }
 
-fn base64_decode_bytes(input: &[u8]) -> Result<Vec<u8>, ()> {
-    const TABLE: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
-    let mut output = Vec::new();
-    let mut buf = 0u32;
-    let mut bits = 0;
-    for &b in input {
-        if b == b'=' || b == b'\n' || b == b'\r' {
-            continue;
-        }
-        let val = TABLE.iter().position(|&c| c == b).ok_or(())? as u32;
-        buf = (buf << 6) | val;
-        bits += 6;
-        if bits >= 8 {
-            bits -= 8;
-            output.push((buf >> bits) as u8);
-            buf &= (1 << bits) - 1;
-        }
-    }
-    Ok(output)
-}
-
 /// Auth layer for internal cluster routes (/internal/*, /cluster/*).
 /// When auth is enabled, requires valid admin credentials.
 /// When auth is disabled, allows all requests (assumes network isolation).

hyperbytedb/src/adapters/http/error.rs

@@ -31,7 +31,8 @@ fn error_to_status_and_message(err: &HyperbytedbError) -> (StatusCode, String) {
         | HyperbytedbError::MissingParameter(_)
         | HyperbytedbError::QueryParse(_) => StatusCode::BAD_REQUEST,
         HyperbytedbError::CardinalityExceeded { .. } => StatusCode::UNPROCESSABLE_ENTITY,
-        HyperbytedbError::RequestPointLimitExceeded { .. } => StatusCode::PAYLOAD_TOO_LARGE,
+        HyperbytedbError::RequestPointLimitExceeded { .. }
+        | HyperbytedbError::PayloadTooLarge(_) => StatusCode::PAYLOAD_TOO_LARGE,
         HyperbytedbError::WalBackpressure { .. } => StatusCode::SERVICE_UNAVAILABLE,
         HyperbytedbError::QueryTimeout => StatusCode::REQUEST_TIMEOUT,
         HyperbytedbError::ClusterUnavailable(_) => StatusCode::SERVICE_UNAVAILABLE,

hyperbytedb/src/adapters/http/peer_handlers.rs

@@ -23,6 +23,22 @@ use crate::ports::metadata::MetadataPort;
 
 use super::router::AppState;
 
+/// Reject spoofed origin node IDs not present in cluster membership.
+async fn validate_origin_node_id(state: &AppState, origin_node_id: u64) -> Result<(), StatusCode> {
+    if origin_node_id == 0 {
+        return Ok(());
+    }
+    let Some(ref membership) = state.membership else {
+        return Err(StatusCode::BAD_REQUEST);
+    };
+    let m = membership.read().await;
+    if m.get_node(origin_node_id).is_some() {
+        Ok(())
+    } else {
+        Err(StatusCode::BAD_REQUEST)
+    }
+}
+
 /// Receives replicated line-protocol writes from a peer (`POST /internal/replicate`).
 pub async fn handle_replicate_write(
     State(state): State<Arc<AppState>>,
@@ -47,6 +63,13 @@ pub async fn handle_replicate_write(
         .and_then(|s| s.parse::<u64>().ok())
         .unwrap_or(0);
 
+    if let Err(status) = validate_origin_node_id(&state, origin_node_id).await {
+        return (
+            status,
+            Json(serde_json::json!({"error": "invalid origin node id"})),
+        );
+    }
+
     let ct = headers
         .get(CONTENT_TYPE)
         .and_then(|v| v.to_str().ok())
@@ -174,6 +197,13 @@ pub async fn handle_replicate_mutation(
     let sender_seq = req.seq;
     let origin = req.origin_node_id;
 
+    if let Err(status) = validate_origin_node_id(&state, origin).await {
+        return (
+            status,
+            Json(serde_json::json!({"error": "invalid origin node id"})),
+        );
+    }
+
     if origin != 0
         && let Some(ref rl) = state.replication_log
         && !rl.check_and_record_mutation(origin, sender_seq)

hyperbytedb/src/adapters/http/query.rs

@@ -237,10 +237,12 @@ async fn handle_query_impl(
 
         if let Some(ref summary) = state.statement_summary {
             let latency_us = elapsed.as_micros() as u64;
+            let sample_query =
+                crate::timeseriesql::digest::redact_credentials(&q);
             summary.record(
                 &digest_hex,
                 &normalized_query,
-                &q,
+                &sample_query,
                 db,
                 stmt_type_label,
                 latency_us,

hyperbytedb/src/adapters/http/router.rs

@@ -46,12 +46,16 @@ pub struct AppState {
     pub auth_enabled: bool,
     pub prometheus_handle: Option<metrics_exporter_prometheus::PrometheusHandle>,
     pub statement_summary: Option<Arc<StatementSummary>>,
+    /// When true and auth is enabled, `/api/v1/statements` requires credentials.
+    pub statement_summary_require_auth: bool,
     pub mv_service: Arc<MaterializedViewService>,
     /// Applies `/internal/replicate` payloads off the HTTP thread (bounded).
     pub replication_apply: Option<Arc<ReplicationApplyQueue>>,
     pub chdb_session_data_path: String,
     pub node_id: u64,
     pub max_body_size_bytes: usize,
+    /// HTTP body cap for `/internal/replicate` (resolved from `[cluster]` at startup).
+    pub replicate_body_limit_bytes: usize,
     pub max_points_per_request: usize,
     pub request_timeout_secs: u64,
     pub rate_limiter: Option<Arc<rate_limit::EndpointRateLimiters>>,
@@ -60,9 +64,21 @@ pub struct AppState {
 pub fn build_router(state: Arc<AppState>) -> Router {
     let auth_state = state.clone();
     let body_limit = state.max_body_size_bytes;
+    let replicate_body_limit = state.replicate_body_limit_bytes;
     let _timeout_duration = std::time::Duration::from_secs(state.request_timeout_secs);
 
-    let mut router = Router::new()
+    let mut statements_router = Router::new().route(
+        "/api/v1/statements",
+        get(statements::handle_list).delete(statements::handle_reset),
+    );
+    if state.auth_enabled && state.statement_summary_require_auth {
+        statements_router = statements_router.route_layer(axum::middleware::from_fn_with_state(
+            auth_state.clone(),
+            auth_middleware::auth_layer,
+        ));
+    }
+
+    let public_router = Router::new()
         .route("/ping", get(ping::ping).head(ping::ping))
         .route("/health", get(ping::health).head(ping::health))
         .route(
@@ -97,10 +113,7 @@ pub fn build_router(state: Arc<AppState>) -> Router {
                 )),
         )
         .route("/metrics", get(metrics::handle_metrics))
-        .route(
-            "/api/v1/statements",
-            get(statements::handle_list).delete(statements::handle_reset),
-        )
+        .merge(statements_router)
         .route(
             "/api/v1/chdb",
             post(chdb::handle_chdb).layer(axum::middleware::from_fn_with_state(
@@ -109,16 +122,19 @@ pub fn build_router(state: Arc<AppState>) -> Router {
             )),
         );
 
+    let mut cluster_router = Router::new();
+
     if state.peer_client.is_some() {
-        let internal_auth = state.clone();
-        router = router
+        cluster_router = cluster_router
             .route(
                 "/internal/replicate",
-                post(peer_handlers::handle_replicate_write).layer(DefaultBodyLimit::disable()),
+                post(peer_handlers::handle_replicate_write)
+                    .layer(DefaultBodyLimit::max(replicate_body_limit)),
             )
             .route(
                 "/internal/replicate-mutation",
-                post(peer_handlers::handle_replicate_mutation).layer(DefaultBodyLimit::disable()),
+                post(peer_handlers::handle_replicate_mutation)
+                    .layer(DefaultBodyLimit::max(replicate_body_limit)),
             )
             .route("/cluster/metrics", get(cluster::handle_cluster_metrics))
             .route("/cluster/nodes", get(cluster::handle_cluster_nodes))
@@ -147,15 +163,11 @@ pub fn build_router(state: Arc<AppState>) -> Router {
                 "/internal/sync/trigger",
                 post(peer_handlers::handle_sync_trigger),
             )
-            .route("/internal/drain", post(peer_handlers::handle_drain))
-            .layer(axum::middleware::from_fn_with_state(
-                internal_auth,
-                auth_middleware::internal_auth_layer,
-            ));
+            .route("/internal/drain", post(peer_handlers::handle_drain));
     }
 
     if state.raft.is_some() {
-        router = router
+        cluster_router = cluster_router
             .route("/internal/raft/vote", post(raft_handlers::handle_raft_vote))
             .route(
                 "/internal/raft/append",
@@ -192,6 +204,17 @@ pub fn build_router(state: Arc<AppState>) -> Router {
             );
     }
 
+    let router = if state.peer_client.is_some() || state.raft.is_some() {
+        let internal_auth = state.clone();
+        let cluster_router = cluster_router.route_layer(axum::middleware::from_fn_with_state(
+            internal_auth,
+            auth_middleware::internal_auth_layer,
+        ));
+        public_router.merge(cluster_router)
+    } else {
+        public_router
+    };
+
     router
         .layer(ServiceBuilder::new().layer(middleware::map_response(
             http_middleware::add_version_headers,

hyperbytedb/src/adapters/http/write.rs

@@ -95,7 +95,7 @@ pub async fn handle_write(
                 .into_response());
         }
 
-        let decompressed = maybe_decompress_gzip(&headers, &body)?;
+        let decompressed = maybe_decompress_gzip(&headers, &body, state.max_body_size_bytes)?;
         let payload: &[u8] = decompressed.as_deref().unwrap_or(&body);
         histogram!("hyperbytedb_write_payload_bytes").record(payload.len() as f64);
 
@@ -161,6 +161,7 @@ fn write_payload_format_from_headers(ct_norm: &str) -> WritePayloadFormat {
 fn maybe_decompress_gzip(
     headers: &HeaderMap,
     body: &[u8],
+    max_decompressed_bytes: usize,
 ) -> Result<Option<Vec<u8>>, HyperbytedbError> {
     let is_gzip = headers
         .get("content-encoding")
@@ -169,15 +170,21 @@ fn maybe_decompress_gzip(
 
     if is_gzip {
         use flate2::read::GzDecoder;
-        use std::io::Read;
-        let mut decoder = GzDecoder::new(body);
-        let mut decompressed = Vec::with_capacity(body.len() * 2);
-        decoder.read_to_end(&mut decompressed).map_err(|e| {
+        use std::io::{Read, Take};
+        let decoder = GzDecoder::new(body);
+        let mut limited: Take<GzDecoder<&[u8]>> = decoder.take(max_decompressed_bytes as u64 + 1);
+        let mut decompressed = Vec::new();
+        limited.read_to_end(&mut decompressed).map_err(|e| {
             HyperbytedbError::LineProtocolParse {
                 line: String::new(),
                 reason: format!("gzip decompression failed: {e}"),
             }
         })?;
+        if decompressed.len() > max_decompressed_bytes {
+            return Err(HyperbytedbError::PayloadTooLarge(format!(
+                "decompressed body exceeds {max_decompressed_bytes} bytes"
+            )));
+        }
         Ok(Some(decompressed))
     } else {
         Ok(None)