diff --git a/Cargo.lock b/Cargo.lock
index 3d81d5d..f8be4a8 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1840,6 +1840,7 @@ dependencies = [
  "async-trait",
  "axum",
  "axum-server",
+ "base64 0.22.1",
  "bincode",
  "bytes",
  "chdb-rust",
diff --git a/config.toml.example b/config.toml.example
index f9f5505..32bb6e7 100644
--- a/config.toml.example
+++ b/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
diff --git a/docs/developer-guide/internals/replication-design.md b/docs/developer-guide/internals/replication-design.md
index 3ca2bbb..60b0a4e 100644
--- a/docs/developer-guide/internals/replication-design.md
+++ b/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
 
diff --git a/docs/user-guide/authentication.md b/docs/user-guide/authentication.md
index 0297a60..e8ab713 100644
--- a/docs/user-guide/authentication.md
+++ b/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.
 
 ---
 
diff --git a/docs/user-guide/configuration.md b/docs/user-guide/configuration.md
index c22f5ca..57522c8 100644
--- a/docs/user-guide/configuration.md
+++ b/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.
 
 ---
 
diff --git a/hyperbytedb/Cargo.toml b/hyperbytedb/Cargo.toml
index c52fffc..b25100c 100644
--- a/hyperbytedb/Cargo.toml
+++ b/hyperbytedb/Cargo.toml
@@ -81,6 +81,7 @@ flate2 = "1"
 
 # Auth
 argon2 = "0.5"
+base64 = "0.22"
 
 # Regex
 regex = "1"
diff --git a/hyperbytedb/src/adapters/auth.rs b/hyperbytedb/src/adapters/auth.rs
index 2811856..dfdf1d0 100644
--- a/hyperbytedb/src/adapters/auth.rs
+++ b/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);
diff --git a/hyperbytedb/src/adapters/http/auth_middleware.rs b/hyperbytedb/src/adapters/http/auth_middleware.rs
index 700535f..221cfbe 100644
--- a/hyperbytedb/src/adapters/http/auth_middleware.rs
+++ b/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).
diff --git a/hyperbytedb/src/adapters/http/error.rs b/hyperbytedb/src/adapters/http/error.rs
index af2cb15..90b0345 100644
--- a/hyperbytedb/src/adapters/http/error.rs
+++ b/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,
diff --git a/hyperbytedb/src/adapters/http/peer_handlers.rs b/hyperbytedb/src/adapters/http/peer_handlers.rs
index 7de0f8f..dd2af41 100644
--- a/hyperbytedb/src/adapters/http/peer_handlers.rs
+++ b/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)
diff --git a/hyperbytedb/src/adapters/http/query.rs b/hyperbytedb/src/adapters/http/query.rs
index e5524fd..0388596 100644
--- a/hyperbytedb/src/adapters/http/query.rs
+++ b/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,
diff --git a/hyperbytedb/src/adapters/http/router.rs b/hyperbytedb/src/adapters/http/router.rs
index 7c89858..65f5a79 100644
--- a/hyperbytedb/src/adapters/http/router.rs
+++ b/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,
diff --git a/hyperbytedb/src/adapters/http/write.rs b/hyperbytedb/src/adapters/http/write.rs
index dbb6174..4dd9d30 100644
--- a/hyperbytedb/src/adapters/http/write.rs
+++ b/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)
diff --git a/hyperbytedb/src/adapters/wal/wal_ipc.rs b/hyperbytedb/src/adapters/wal/wal_ipc.rs
index 89c2ff4..54dbc2c 100644
--- a/hyperbytedb/src/adapters/wal/wal_ipc.rs
+++ b/hyperbytedb/src/adapters/wal/wal_ipc.rs
@@ -169,6 +169,15 @@ pub fn decode_prepared_slot(
             .read_exact(&mut len_buf)
             .map_err(|e| HyperbytedbError::Wal(e.to_string()))?;
         let fact_len = u32::from_le_bytes(len_buf) as usize;
+        let remaining = cursor
+            .get_ref()
+            .len()
+            .saturating_sub(cursor.position() as usize);
+        if fact_len > remaining {
+            return Err(HyperbytedbError::Wal(format!(
+                "invalid fact_ipc length {fact_len} (only {remaining} bytes remaining)"
+            )));
+        }
         let mut fact_ipc = vec![0u8; fact_len];
         cursor
             .read_exact(&mut fact_ipc)
@@ -180,6 +189,15 @@ pub fn decode_prepared_slot(
             .map_err(|e| HyperbytedbError::Wal(e.to_string()))?;
         let series_len = u32::from_le_bytes(len_buf) as usize;
         let new_series_batch = if series_len > 0 {
+            let remaining = cursor
+                .get_ref()
+                .len()
+                .saturating_sub(cursor.position() as usize);
+            if series_len > remaining {
+                return Err(HyperbytedbError::Wal(format!(
+                    "invalid series_ipc length {series_len} (only {remaining} bytes remaining)"
+                )));
+            }
             let mut series_ipc = vec![0u8; series_len];
             cursor
                 .read_exact(&mut series_ipc)
@@ -206,6 +224,15 @@ pub fn decode_prepared_slot(
         .map_err(|e| HyperbytedbError::Wal(e.to_string()))?;
     let legacy_len = u32::from_le_bytes(mc_buf) as usize;
     let legacy_entry = if legacy_len > 0 {
+        let remaining = cursor
+            .get_ref()
+            .len()
+            .saturating_sub(cursor.position() as usize);
+        if legacy_len > remaining {
+            return Err(HyperbytedbError::Wal(format!(
+                "invalid legacy entry length {legacy_len} (only {remaining} bytes remaining)"
+            )));
+        }
         let mut legacy = vec![0u8; legacy_len];
         cursor
             .read_exact(&mut legacy)
diff --git a/hyperbytedb/src/application/query_service.rs b/hyperbytedb/src/application/query_service.rs
index 946d133..917c2f1 100644
--- a/hyperbytedb/src/application/query_service.rs
+++ b/hyperbytedb/src/application/query_service.rs
@@ -1360,11 +1360,12 @@ async fn execute_statement(
                         .await?;
                 }
                 None => {
-                    if let Some(user) = svc.metadata.get_user(username).await? {
-                        svc.metadata
-                            .create_user(username, &user.password_hash, true)
-                            .await?;
-                    }
+                    let user = svc.metadata.get_user(username).await?.ok_or_else(|| {
+                        HyperbytedbError::QueryParse(format!("user '{username}' does not exist"))
+                    })?;
+                    svc.metadata
+                        .create_user(username, &user.password_hash, true)
+                        .await?;
                 }
             }
             Ok(StatementResult {
@@ -1379,11 +1380,12 @@ async fn execute_statement(
                     svc.metadata.revoke_privilege(username, db).await?;
                 }
                 None => {
-                    if let Some(user) = svc.metadata.get_user(username).await? {
-                        svc.metadata
-                            .create_user(username, &user.password_hash, false)
-                            .await?;
-                    }
+                    let user = svc.metadata.get_user(username).await?.ok_or_else(|| {
+                        HyperbytedbError::QueryParse(format!("user '{username}' does not exist"))
+                    })?;
+                    svc.metadata
+                        .create_user(username, &user.password_hash, false)
+                        .await?;
                 }
             }
             Ok(StatementResult {
diff --git a/hyperbytedb/src/bootstrap.rs b/hyperbytedb/src/bootstrap.rs
index ff7c091..3b51be1 100644
--- a/hyperbytedb/src/bootstrap.rs
+++ b/hyperbytedb/src/bootstrap.rs
@@ -72,6 +72,17 @@ pub async fn build_services(config: &HyperbytedbConfig) -> anyhow::Result<Bootst
             "cluster enabled: replicated durability is via shared WAL/metadata sync; embedded \
              chDB state is rebuilt from the WAL on each peer"
         );
+        let replicate_limit = config
+            .cluster
+            .effective_replicate_body_limit_bytes(config.server.max_body_size_bytes);
+        if config.cluster.replication_max_coalesce_body_bytes > replicate_limit {
+            tracing::warn!(
+                coalesce_bytes = config.cluster.replication_max_coalesce_body_bytes,
+                replicate_body_limit_bytes = replicate_limit,
+                "replication_max_coalesce_body_bytes exceeds replicate_body_limit_bytes; \
+                 outbound coalesced batches may be rejected by peers (HTTP 413)"
+            );
+        }
     }
 
     // -- Infrastructure adapters --
@@ -365,10 +376,14 @@ pub async fn build_services(config: &HyperbytedbConfig) -> anyhow::Result<Bootst
         auth_enabled: config.auth.enabled,
         prometheus_handle: Some(prometheus_handle.clone()),
         statement_summary,
+        statement_summary_require_auth: config.statement_summary.require_auth,
         replication_apply,
         chdb_session_data_path: config.chdb.session_data_path.clone(),
         node_id: config.cluster.node_id,
         max_body_size_bytes: config.server.max_body_size_bytes,
+        replicate_body_limit_bytes: config
+            .cluster
+            .effective_replicate_body_limit_bytes(config.server.max_body_size_bytes),
         max_points_per_request: config.server.max_points_per_request,
         request_timeout_secs: config.server.request_timeout_secs,
         rate_limiter: if config.rate_limit.enabled && config.rate_limit.max_requests_per_second > 0
diff --git a/hyperbytedb/src/config.rs b/hyperbytedb/src/config.rs
index 09c0478..0ea26e6 100644
--- a/hyperbytedb/src/config.rs
+++ b/hyperbytedb/src/config.rs
@@ -177,6 +177,11 @@ pub struct ClusterConfig {
     /// Max bytes for coalescing consecutive WAL batches (same db/rp/precision).
     #[serde(default = "default_replication_max_coalesce_body_bytes")]
     pub replication_max_coalesce_body_bytes: usize,
+    /// Max HTTP body size for `/internal/replicate` and `/internal/replicate-mutation`.
+    /// `0` resolves at startup to `max(4 × replication_max_coalesce_body_bytes,
+    /// server.max_body_size_bytes)`.
+    #[serde(default)]
+    pub replicate_body_limit_bytes: usize,
     /// Bounded apply queue on the replicate receiver.
     #[serde(default = "default_replicate_receiver_queue_depth")]
     pub replicate_receiver_queue_depth: usize,
@@ -343,6 +348,17 @@ impl ClusterConfig {
             .filter(|s| !s.is_empty())
             .collect()
     }
+
+    /// Resolved HTTP body cap for peer replication endpoints.
+    pub fn effective_replicate_body_limit_bytes(&self, max_body_size_bytes: usize) -> usize {
+        if self.replicate_body_limit_bytes > 0 {
+            self.replicate_body_limit_bytes
+        } else {
+            self.replication_max_coalesce_body_bytes
+                .saturating_mul(4)
+                .max(max_body_size_bytes)
+        }
+    }
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
@@ -355,6 +371,13 @@ pub struct LoggingConfig {
 pub struct StatementSummaryConfig {
     pub enabled: bool,
     pub max_entries: usize,
+    /// Require auth for GET/DELETE `/api/v1/statements` when `[auth] enabled = true`.
+    #[serde(default = "default_true")]
+    pub require_auth: bool,
+}
+
+fn default_true() -> bool {
+    true
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
@@ -498,6 +521,7 @@ impl HyperbytedbConfig {
                 replication_queue_depth: default_replication_queue_depth(),
                 replication_max_inflight_batches: default_replication_max_inflight_batches(),
                 replication_max_coalesce_body_bytes: default_replication_max_coalesce_body_bytes(),
+                replicate_body_limit_bytes: 0,
                 replicate_receiver_queue_depth: default_replicate_receiver_queue_depth(),
                 replicate_receiver_workers: default_replicate_receiver_workers(),
                 replication_truncate_stale_peer_multiplier:
@@ -514,6 +538,7 @@ impl HyperbytedbConfig {
             statement_summary: StatementSummaryConfig {
                 enabled: true,
                 max_entries: 1000,
+                require_auth: true,
             },
             hinted_handoff: HintedHandoffConfig {
                 enabled: true,
@@ -655,3 +680,64 @@ mod retention_config_tests {
         assert_eq!(r.interval, "12h");
     }
 }
+
+#[cfg(test)]
+mod replicate_body_limit_tests {
+    use super::ClusterConfig;
+
+    fn base_cluster() -> ClusterConfig {
+        ClusterConfig {
+            enabled: true,
+            node_id: 1,
+            cluster_addr: "127.0.0.1:8086".into(),
+            peers: String::new(),
+            heartbeat_interval_secs: 2,
+            heartbeat_miss_threshold: 5,
+            anti_entropy_enabled: false,
+            anti_entropy_interval_secs: 60,
+            replication_log_dir: "./replication_log".into(),
+            raft_dir: "./raft".into(),
+            replication_max_retries: 5,
+            replication_queue_depth: 8192,
+            replication_max_inflight_batches: 8,
+            replication_max_coalesce_body_bytes: 8 * 1024 * 1024,
+            replicate_body_limit_bytes: 0,
+            replicate_receiver_queue_depth: 1024,
+            replicate_receiver_workers: 1,
+            replication_truncate_stale_peer_multiplier: 2,
+            raft_heartbeat_interval_ms: None,
+            raft_election_timeout_ms: None,
+            raft_snapshot_threshold: None,
+            replication: super::ReplicationConfig::default(),
+        }
+    }
+
+    #[test]
+    fn auto_limit_uses_four_times_coalesce_when_larger_than_write_cap() {
+        let c = base_cluster();
+        assert_eq!(
+            c.effective_replicate_body_limit_bytes(25 * 1024 * 1024),
+            32 * 1024 * 1024
+        );
+    }
+
+    #[test]
+    fn auto_limit_uses_write_cap_when_coalesce_times_four_is_smaller() {
+        let mut c = base_cluster();
+        c.replication_max_coalesce_body_bytes = 1024;
+        assert_eq!(
+            c.effective_replicate_body_limit_bytes(25 * 1024 * 1024),
+            25 * 1024 * 1024
+        );
+    }
+
+    #[test]
+    fn explicit_limit_overrides_auto() {
+        let mut c = base_cluster();
+        c.replicate_body_limit_bytes = 64 * 1024 * 1024;
+        assert_eq!(
+            c.effective_replicate_body_limit_bytes(25 * 1024 * 1024),
+            64 * 1024 * 1024
+        );
+    }
+}
diff --git a/hyperbytedb/src/domain/chdb_naming.rs b/hyperbytedb/src/domain/chdb_naming.rs
index dfb01d1..16ea709 100644
--- a/hyperbytedb/src/domain/chdb_naming.rs
+++ b/hyperbytedb/src/domain/chdb_naming.rs
@@ -53,8 +53,8 @@ pub fn unquoted_table_name(db: &str, rp: &str, measurement: &str) -> String {
 /// generated SQL.
 #[must_use]
 pub fn quote_backticks(ident: &str) -> String {
-    let escaped = ident.replace('`', "``");
-    format!("`{}`", escaped)
+    let escaped = ident.replace('\\', "\\\\").replace('`', "``");
+    format!("`{escaped}`")
 }
 
 /// Backtick-quoted, sanitised table name suitable for splicing into
diff --git a/hyperbytedb/src/domain/database.rs b/hyperbytedb/src/domain/database.rs
index dc4c74e..7b68b37 100644
--- a/hyperbytedb/src/domain/database.rs
+++ b/hyperbytedb/src/domain/database.rs
@@ -193,9 +193,9 @@ impl Precision {
     pub fn to_nanos(&self, ts: i64) -> i64 {
         match self {
             Precision::Nanosecond => ts,
-            Precision::Microsecond => ts * 1_000,
-            Precision::Millisecond => ts * 1_000_000,
-            Precision::Second => ts * 1_000_000_000,
+            Precision::Microsecond => ts.saturating_mul(1_000),
+            Precision::Millisecond => ts.saturating_mul(1_000_000),
+            Precision::Second => ts.saturating_mul(1_000_000_000),
         }
     }
 
diff --git a/hyperbytedb/src/error.rs b/hyperbytedb/src/error.rs
index 5402be0..23d4c1e 100644
--- a/hyperbytedb/src/error.rs
+++ b/hyperbytedb/src/error.rs
@@ -70,6 +70,9 @@ pub enum HyperbytedbError {
     #[error("request exceeds maximum point count: {count} points (limit: {limit})")]
     RequestPointLimitExceeded { count: usize, limit: usize },
 
+    #[error("request payload too large: {0}")]
+    PayloadTooLarge(String),
+
     #[error("WAL backpressure: write queue full for {timeout_ms}ms")]
     WalBackpressure { timeout_ms: u64 },
 
diff --git a/hyperbytedb/src/timeseriesql/digest.rs b/hyperbytedb/src/timeseriesql/digest.rs
index 1bc67b7..c09078c 100644
--- a/hyperbytedb/src/timeseriesql/digest.rs
+++ b/hyperbytedb/src/timeseriesql/digest.rs
@@ -49,6 +49,35 @@ pub fn fingerprint(stmt: &Statement) -> (String, String) {
     (short_digest, normalized)
 }
 
+/// Redact credential literals before storing query text in the statement summary.
+pub fn redact_credentials(query: &str) -> String {
+    use regex::Regex;
+    use std::sync::OnceLock;
+
+    static PASSWORD_SINGLE: OnceLock<Regex> = OnceLock::new();
+    static PASSWORD_DOUBLE: OnceLock<Regex> = OnceLock::new();
+    static QUERY_CREDS: OnceLock<Regex> = OnceLock::new();
+
+    let mut out = query.to_string();
+    // Literal patterns; invalid regex is a programming error, not user input.
+    let re = PASSWORD_SINGLE.get_or_init(|| {
+        #[allow(clippy::expect_used)]
+        Regex::new(r"(?i)(PASSWORD\s+)'[^']*'").expect("password single-quote re")
+    });
+    out = re.replace_all(&out, "$1'****'").into_owned();
+    let re = PASSWORD_DOUBLE.get_or_init(|| {
+        #[allow(clippy::expect_used)]
+        Regex::new(r#"(?i)(PASSWORD\s+)"[^"]*""#).expect("password double-quote re")
+    });
+    out = re.replace_all(&out, r#"$1"****""#).into_owned();
+    let re = QUERY_CREDS.get_or_init(|| {
+        #[allow(clippy::expect_used)]
+        Regex::new(r"(?i)([?&](?:u|p)=)[^&\s]+").expect("query creds re")
+    });
+    out = re.replace_all(&out, "$1****").into_owned();
+    out
+}
+
 fn normalize_statement(stmt: &Statement) -> String {
     let mut out = String::new();
     match stmt {
@@ -537,4 +566,12 @@ mod tests {
         // collide on the same digest.
         assert_eq!(n1, "select mean(usage) from cpu where (host = ?)");
     }
+
+    #[test]
+    fn redact_credentials_masks_password_literals() {
+        let raw = r#"CREATE USER "x" WITH PASSWORD 's3cret'"#;
+        let redacted = redact_credentials(raw);
+        assert!(!redacted.contains("s3cret"));
+        assert!(redacted.contains("****"));
+    }
 }
diff --git a/hyperbytedb/tests/compat/http_tests.rs b/hyperbytedb/tests/compat/http_tests.rs
index f15ae6c..b3013e7 100644
--- a/hyperbytedb/tests/compat/http_tests.rs
+++ b/hyperbytedb/tests/compat/http_tests.rs
@@ -122,10 +122,12 @@ impl HttpTestContext {
             auth_enabled: false,
             prometheus_handle: None,
             statement_summary: None,
+            statement_summary_require_auth: true,
             replication_apply: None,
             chdb_session_data_path: chdb_dir.to_string_lossy().into_owned(),
             node_id: 1,
             max_body_size_bytes: 25 * 1024 * 1024,
+            replicate_body_limit_bytes: 32 * 1024 * 1024,
             max_points_per_request: 0,
             request_timeout_secs: 30,
             rate_limiter: None,
diff --git a/hyperbytedb/tests/integration.rs b/hyperbytedb/tests/integration.rs
index 313f7d5..ba1adf1 100644
--- a/hyperbytedb/tests/integration.rs
+++ b/hyperbytedb/tests/integration.rs
@@ -104,10 +104,12 @@ fn setup(dir: &tempfile::TempDir) -> (Arc<AppState>, Arc<FlushServiceImpl>) {
         auth_enabled: false,
         prometheus_handle: None,
         statement_summary: None,
+        statement_summary_require_auth: true,
         replication_apply: None,
         chdb_session_data_path: chdb_path_str,
         node_id: 1,
         max_body_size_bytes: 25 * 1024 * 1024,
+        replicate_body_limit_bytes: 32 * 1024 * 1024,
         max_points_per_request: 0,
         request_timeout_secs: 30,
         rate_limiter: None,
@@ -184,10 +186,12 @@ async fn test_auth_blocks_unauthenticated() {
         auth_enabled: true,
         prometheus_handle: None,
         statement_summary: None,
+        statement_summary_require_auth: true,
         replication_apply: None,
         chdb_session_data_path: chdb_dir.to_string_lossy().into_owned(),
         node_id: 1,
         max_body_size_bytes: 25 * 1024 * 1024,
+        replicate_body_limit_bytes: 32 * 1024 * 1024,
         max_points_per_request: 0,
         request_timeout_secs: 30,
         rate_limiter: None,
@@ -283,10 +287,12 @@ async fn test_cardinality_limit() {
         auth_enabled: false,
         prometheus_handle: None,
         statement_summary: None,
+        statement_summary_require_auth: true,
         replication_apply: None,
         chdb_session_data_path: chdb_dir.to_string_lossy().into_owned(),
         node_id: 1,
         max_body_size_bytes: 25 * 1024 * 1024,
+        replicate_body_limit_bytes: 32 * 1024 * 1024,
         max_points_per_request: 0,
         request_timeout_secs: 30,
         rate_limiter: None,
@@ -427,10 +433,12 @@ async fn test_metrics_endpoint() {
         auth_enabled: false,
         prometheus_handle: Some(prometheus_handle),
         statement_summary: None,
+        statement_summary_require_auth: true,
         replication_apply: None,
         chdb_session_data_path: chdb_dir.to_string_lossy().into_owned(),
         node_id: 1,
         max_body_size_bytes: 25 * 1024 * 1024,
+        replicate_body_limit_bytes: 32 * 1024 * 1024,
         max_points_per_request: 0,
         request_timeout_secs: 30,
         rate_limiter: None,
@@ -759,10 +767,12 @@ async fn test_rate_limiter_refills_and_denies() {
         auth_enabled: false,
         prometheus_handle: Some(prometheus_handle),
         statement_summary: None,
+        statement_summary_require_auth: true,
         replication_apply: None,
         chdb_session_data_path: chdb_dir.to_string_lossy().into_owned(),
         node_id: 1,
         max_body_size_bytes: 25 * 1024 * 1024,
+        replicate_body_limit_bytes: 32 * 1024 * 1024,
         max_points_per_request: 0,
         request_timeout_secs: 30,
         rate_limiter: Some(Arc::new(EndpointRateLimiters::new(5))),
@@ -911,10 +921,12 @@ async fn test_cross_database_on_clause_requires_authorization() {
         auth_enabled: true,
         prometheus_handle: None,
         statement_summary: None,
+        statement_summary_require_auth: true,
         replication_apply: None,
         chdb_session_data_path: chdb_dir.to_string_lossy().into_owned(),
         node_id: 1,
         max_body_size_bytes: 25 * 1024 * 1024,
+        replicate_body_limit_bytes: 32 * 1024 * 1024,
         max_points_per_request: 0,
         request_timeout_secs: 30,
         rate_limiter: None,
diff --git a/hyperbytedb/tests/raft_integration.rs b/hyperbytedb/tests/raft_integration.rs
index e1bf32a..0d9ec9f 100644
--- a/hyperbytedb/tests/raft_integration.rs
+++ b/hyperbytedb/tests/raft_integration.rs
@@ -140,10 +140,12 @@ async fn start_cluster_node_with_listener(
         auth_enabled: false,
         prometheus_handle: None,
         statement_summary: None,
+        statement_summary_require_auth: true,
         replication_apply,
         chdb_session_data_path: chdb_path,
         node_id,
         max_body_size_bytes: 25 * 1024 * 1024,
+        replicate_body_limit_bytes: 32 * 1024 * 1024,
         max_points_per_request: 0,
         request_timeout_secs: 30,
         rate_limiter: None,
@@ -337,10 +339,12 @@ async fn test_cluster_endpoints_without_peers() {
         auth_enabled: false,
         prometheus_handle: None,
         statement_summary: None,
+        statement_summary_require_auth: true,
         replication_apply: None,
         chdb_session_data_path: chdb_dir.to_string_lossy().into_owned(),
         node_id: 1,
         max_body_size_bytes: 25 * 1024 * 1024,
+        replicate_body_limit_bytes: 32 * 1024 * 1024,
         max_points_per_request: 0,
         request_timeout_secs: 30,
         rate_limiter: None,
diff --git a/hyperbytedb/tests/security_auth.rs b/hyperbytedb/tests/security_auth.rs
new file mode 100644
index 0000000..7249d83
--- /dev/null
+++ b/hyperbytedb/tests/security_auth.rs
@@ -0,0 +1,349 @@
+//! Security-focused HTTP integration tests: cluster+auth route gating,
+//! statement summary auth/redaction, and GRANT/REVOKE enforcement.
+
+use std::sync::Arc;
+
+use axum::http::StatusCode;
+use hyperbytedb::adapters::chdb::native_adapter::ChdbNativeAdapter;
+use hyperbytedb::adapters::chdb::query_adapter::ChdbQueryAdapter;
+use hyperbytedb::adapters::chdb::session::SharedSession;
+use hyperbytedb::adapters::cluster::peer_client::PeerClient;
+use hyperbytedb::adapters::cluster::replication_log::ReplicationLog;
+use hyperbytedb::adapters::http::router::{AppState, build_router};
+use hyperbytedb::adapters::metadata::rocksdb_meta::RocksDbMetadata;
+use hyperbytedb::adapters::wal::rocksdb_wal::RocksDbWal;
+use hyperbytedb::application::ingest_metadata::IngestCardinalityLimits;
+use hyperbytedb::application::materialized_view_service::MaterializedViewService;
+use hyperbytedb::application::peer_ingestion_service::PeerIngestionService;
+use hyperbytedb::application::peer_query_service::PeerQueryService;
+use hyperbytedb::application::query_service::QueryServiceImpl;
+use hyperbytedb::application::replication_apply::ReplicationApplyQueue;
+use hyperbytedb::application::statement_summary::StatementSummary;
+use hyperbytedb::domain::cluster::membership::{
+    ClusterMembership, NodeInfo, NodeState, new_shared,
+};
+use hyperbytedb::ports::metadata::MetadataPort;
+use hyperbytedb::ports::points_sink::PointsSinkPort;
+use serial_test::serial;
+
+struct AuthClusterNode {
+    url: String,
+    _handle: tokio::task::JoinHandle<()>,
+}
+
+async fn start_auth_cluster_node(dir: &std::path::Path) -> AuthClusterNode {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let listener_addr = listener.local_addr().unwrap();
+    let url = format!("http://{}", listener_addr);
+
+    let wal_dir = dir.join("wal");
+    let meta_dir = dir.join("meta");
+    let chdb_dir = dir.join("chdb");
+    std::fs::create_dir_all(&wal_dir).unwrap();
+    std::fs::create_dir_all(&meta_dir).unwrap();
+    std::fs::create_dir_all(&chdb_dir).unwrap();
+
+    let wal = Arc::new(RocksDbWal::open(&wal_dir).unwrap());
+    let metadata = Arc::new(RocksDbMetadata::open(&meta_dir).unwrap());
+    let shared = SharedSession::new_eager(chdb_dir.to_str().unwrap(), 1).unwrap();
+    let chdb_path = shared.data_path().to_string();
+    let chdb = Arc::new(ChdbQueryAdapter::from_shared(shared.clone(), 0));
+    let sink: Arc<dyn PointsSinkPort> = Arc::new(ChdbNativeAdapter::new(shared));
+
+    let admin_hash =
+        hyperbytedb::adapters::http::auth_middleware::hash_password("adminpw").unwrap();
+    metadata
+        .create_user("admin", &admin_hash, true)
+        .await
+        .unwrap();
+    let writer_hash =
+        hyperbytedb::adapters::http::auth_middleware::hash_password("writerpw").unwrap();
+    metadata
+        .create_user("writer", &writer_hash, false)
+        .await
+        .unwrap();
+
+    let repl_dir = dir.join("repl");
+    std::fs::create_dir_all(&repl_dir).unwrap();
+    let replication_log = Arc::new(ReplicationLog::open(&repl_dir).unwrap());
+
+    let mut membership = ClusterMembership::new();
+    let now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap()
+        .as_millis() as i64;
+    membership.add_node(NodeInfo {
+        node_id: 1,
+        addr: listener_addr.to_string(),
+        state: NodeState::Active,
+        joined_at: now,
+        last_heartbeat: now,
+        needs_sync: false,
+    });
+    let shared_membership = new_shared(membership);
+
+    let peer_client = Arc::new(PeerClient::new(
+        1,
+        listener_addr.to_string(),
+        shared_membership.clone(),
+        replication_log.clone(),
+        5,
+        8192,
+        8,
+        8 * 1024 * 1024,
+    ));
+
+    let replication_apply = Some(ReplicationApplyQueue::with_workers_and_sink(
+        1024,
+        8,
+        metadata.clone(),
+        wal.clone(),
+        Some(sink.clone()),
+        IngestCardinalityLimits::default(),
+        0,
+    ));
+
+    let base_query_service: Arc<dyn hyperbytedb::adapters::http::router::QueryService> =
+        Arc::new(QueryServiceImpl::new(
+            chdb.clone(),
+            metadata.clone(),
+            wal.clone(),
+            30,
+            sink.clone(),
+        ));
+
+    let ingestion_service: Arc<dyn hyperbytedb::ports::ingestion::IngestionPort> =
+        Arc::new(PeerIngestionService::new(
+            wal.clone(),
+            metadata.clone(),
+            peer_client.clone(),
+            1,
+            IngestCardinalityLimits::default(),
+        ));
+
+    let query_service: Arc<dyn hyperbytedb::adapters::http::router::QueryService> = Arc::new(
+        PeerQueryService::new(base_query_service, metadata.clone(), peer_client.clone()),
+    );
+
+    let app_state = Arc::new(AppState {
+        ingestion: ingestion_service,
+        query: query_service,
+        query_port: chdb.clone(),
+        metadata: metadata.clone(),
+        wal: wal.clone(),
+        points_sink: sink.clone(),
+        mv_service: Arc::new(MaterializedViewService::new(
+            metadata.clone(),
+            chdb.clone(),
+            sink.clone(),
+        )),
+        auth: Arc::new(hyperbytedb::adapters::auth::MetadataAuthAdapter::new(
+            metadata.clone(),
+        )),
+        peer_client: Some(peer_client),
+        membership: Some(shared_membership),
+        replication_log: Some(replication_log),
+        drain_service: None,
+        raft: None,
+        auth_enabled: true,
+        prometheus_handle: None,
+        statement_summary: Some(Arc::new(StatementSummary::new(100))),
+        statement_summary_require_auth: true,
+        replication_apply,
+        chdb_session_data_path: chdb_path,
+        node_id: 1,
+        max_body_size_bytes: 25 * 1024 * 1024,
+        replicate_body_limit_bytes: 32 * 1024 * 1024,
+        max_points_per_request: 0,
+        request_timeout_secs: 30,
+        rate_limiter: None,
+    });
+
+    let app = build_router(app_state);
+    let handle = tokio::spawn(async move {
+        axum::serve(listener, app).await.unwrap();
+    });
+    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
+
+    AuthClusterNode {
+        url,
+        _handle: handle,
+    }
+}
+
+#[tokio::test]
+#[serial(chdb)]
+async fn cluster_auth_public_health_stays_open() {
+    let dir = tempfile::tempdir().unwrap();
+    let node = start_auth_cluster_node(dir.path()).await;
+    let client = reqwest::Client::new();
+
+    let ping = client
+        .get(format!("{}/ping", node.url))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(ping.status(), StatusCode::NO_CONTENT);
+
+    let health = client
+        .get(format!("{}/health", node.url))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(health.status(), StatusCode::OK);
+
+    let write = client
+        .post(format!("{}/write", node.url))
+        .query(&[("db", "mydb")])
+        .body("cpu,host=a value=1")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(
+        write.status(),
+        StatusCode::UNAUTHORIZED,
+        "non-admin write should require user auth, not admin auth"
+    );
+}
+
+#[tokio::test]
+#[serial(chdb)]
+async fn cluster_auth_internal_routes_require_admin() {
+    let dir = tempfile::tempdir().unwrap();
+    let node = start_auth_cluster_node(dir.path()).await;
+    let client = reqwest::Client::new();
+
+    let unauth = client
+        .get(format!("{}/cluster/metrics", node.url))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(unauth.status(), StatusCode::UNAUTHORIZED);
+
+    let non_admin = client
+        .get(format!("{}/internal/membership", node.url))
+        .query(&[("u", "writer"), ("p", "writerpw")])
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(non_admin.status(), StatusCode::FORBIDDEN);
+
+    let admin = client
+        .get(format!("{}/internal/membership", node.url))
+        .query(&[("u", "admin"), ("p", "adminpw")])
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(admin.status(), StatusCode::OK);
+}
+
+#[tokio::test]
+#[serial(chdb)]
+async fn statement_summary_requires_auth_and_redacts_passwords() {
+    let dir = tempfile::tempdir().unwrap();
+    let node = start_auth_cluster_node(dir.path()).await;
+    let client = reqwest::Client::new();
+
+    let unauth = client
+        .get(format!("{}/api/v1/statements", node.url))
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(unauth.status(), StatusCode::UNAUTHORIZED);
+
+    let create_user = client
+        .get(format!("{}/query", node.url))
+        .query(&[
+            ("q", r#"CREATE USER "leak" WITH PASSWORD 's3cret'"#),
+            ("u", "admin"),
+            ("p", "adminpw"),
+        ])
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(create_user.status(), StatusCode::OK);
+
+    let list = client
+        .get(format!("{}/api/v1/statements", node.url))
+        .query(&[("u", "admin"), ("p", "adminpw")])
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(list.status(), StatusCode::OK);
+    let body: serde_json::Value = list.json().await.unwrap();
+    let entries = body["statements"].as_array().expect("statements array");
+    let sample = entries
+        .iter()
+        .find_map(|e| e["sample_query"].as_str())
+        .expect("sample_query present");
+    assert!(!sample.contains("s3cret"), "password leaked: {sample}");
+    assert!(sample.contains("****"), "password not redacted: {sample}");
+}
+
+#[tokio::test]
+#[serial(chdb)]
+async fn grant_revoke_controls_write_access() {
+    let dir = tempfile::tempdir().unwrap();
+    let node = start_auth_cluster_node(dir.path()).await;
+    let client = reqwest::Client::new();
+
+    metadata_create_db(&node.url, &client).await;
+
+    let grant = client
+        .get(format!("{}/query", node.url))
+        .query(&[
+            ("db", "mydb"),
+            ("q", r#"GRANT ALL ON "mydb" TO "writer""#),
+            ("u", "admin"),
+            ("p", "adminpw"),
+        ])
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(grant.status(), StatusCode::OK);
+
+    let allowed = client
+        .post(format!("{}/write", node.url))
+        .query(&[("db", "mydb"), ("u", "writer"), ("p", "writerpw")])
+        .body("cpu,host=a value=1")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(allowed.status(), StatusCode::NO_CONTENT);
+
+    let revoke = client
+        .get(format!("{}/query", node.url))
+        .query(&[
+            ("db", "mydb"),
+            ("q", r#"REVOKE ALL ON "mydb" FROM "writer""#),
+            ("u", "admin"),
+            ("p", "adminpw"),
+        ])
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(revoke.status(), StatusCode::OK);
+
+    let denied = client
+        .post(format!("{}/write", node.url))
+        .query(&[("db", "mydb"), ("u", "writer"), ("p", "writerpw")])
+        .body("cpu,host=a value=2")
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(denied.status(), StatusCode::FORBIDDEN);
+}
+
+async fn metadata_create_db(url: &str, client: &reqwest::Client) {
+    let resp = client
+        .get(format!("{url}/query"))
+        .query(&[
+            ("q", r#"CREATE DATABASE "mydb""#),
+            ("u", "admin"),
+            ("p", "adminpw"),
+        ])
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::OK);
+}
diff --git a/hyperbytedb/tests/sync_quorum_integration.rs b/hyperbytedb/tests/sync_quorum_integration.rs
index 9de4bfb..c0ea407 100644
--- a/hyperbytedb/tests/sync_quorum_integration.rs
+++ b/hyperbytedb/tests/sync_quorum_integration.rs
@@ -182,10 +182,12 @@ async fn start_node_on(
         auth_enabled: false,
         prometheus_handle: None,
         statement_summary: None,
+        statement_summary_require_auth: true,
         replication_apply,
         chdb_session_data_path: chdb_path_str,
         node_id,
         max_body_size_bytes: 25 * 1024 * 1024,
+        replicate_body_limit_bytes: 32 * 1024 * 1024,
         max_points_per_request: 0,
         request_timeout_secs: 30,
         rate_limiter: None,