feat: client connection security hardening (P0 + P1)

CHANGELOG.md

@@ -22,6 +22,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Fair comparison benchmark** (`tests/graph_bench_compare.rs`): Moon 2.4x FalkorDB on Cypher MATCH, 19x on native 1-hop, 23x on population.
 - **New dependencies**: `slotmap` 1.x (generational indices), `boomphf` 0.6 (MPH), `logos` 0.14 (Cypher lexer, optional).
 
+### Added — Client Connection Security Hardening (2026-04-10)
+
+- **`--maxclients` (P0):** Connection limit with atomic CAS rejection (default 10000, 0=unlimited). Returns `-ERR max number of clients reached` when exceeded.
+- **`--timeout` (P0):** Client idle timeout in seconds (default 0=disabled). Disconnects idle clients via `tokio::time::timeout` / `monoio::select!`.
+- **`--tcp-keepalive` (P0):** TCP keepalive interval (default 300s, 0=disabled). Sets `SO_KEEPALIVE` + `TCP_KEEPIDLE` on accepted sockets via `socket2`.
+- **AUTH rate limiting (P0):** Per-IP exponential backoff on AUTH failures (100ms base, 10s cap, 60s auto-reset). New module `src/auth_ratelimit.rs`.
+- **CLIENT LIST / INFO / KILL (P1):** Global client registry with Drop-guard deregister. Redis-compatible output format. Kill by ID/ADDR/USER. New module `src/client_registry.rs`.
+- **CLIENT PAUSE / UNPAUSE (P1):** Server-wide pause with ALL/WRITE modes and auto-expiry. New module `src/client_pause.rs`.
+- **CLIENT NO-EVICT / NO-TOUCH (P1):** Accepted stubs for Redis compatibility.
+- **ACL GENPASS (P1):** Cryptographically secure random password generation (1-4096 bits, hex output).
+- **CONFIG GET/SET** support for `maxclients`, `timeout`, `tcp-keepalive` (runtime-mutable).
+- **Monoio connection tracking:** Added missing `record_connection_opened` / `record_connection_closed` for accurate `connected_clients` metric.
+
 ### Fixed — Deep Review Findings (2026-04-11)
 
 - **DoS protection**: `execute_profile` and `execute_mut` Cypher paths now enforce MAX_HOPS_LIMIT=20 and MAX_RESULT_ROWS=100K (were unbounded).

deny.toml

@@ -1,17 +1,20 @@
-# cargo-deny configuration for Moon.
+# cargo-deny configuration for Moon (v2 format).
 # Run: cargo deny check
 # CI: .github/workflows/ci.yml safety-audit job
 
+[graph]
+targets = []
+all-features = false
+no-default-features = false
+
 [advisories]
+version = 2
 db-path = "~/.cargo/advisory-db"
 db-urls = ["https://github.com/rustsec/advisory-db"]
-vulnerability = "deny"
-unmaintained = "warn"
-yanked = "warn"
-notice = "warn"
+ignore = []
 
 [licenses]
-unlicensed = "deny"
+version = 2
 allow = [
     "MIT",
     "Apache-2.0",
@@ -26,7 +29,6 @@ allow = [
     "CC0-1.0",
     "0BSD",
 ]
-copyleft = "deny"
 
 [bans]
 multiple-versions = "warn"

src/admin/metrics_setup.rs

@@ -365,6 +365,42 @@ pub fn connected_clients() -> u64 {
     CONNECTED_CLIENTS.load(Ordering::Relaxed)
 }
 
+/// Try to open a connection if under the maxclients limit.
+/// Returns true if the connection was accepted, false if at limit.
+/// When maxclients is 0, the limit is disabled (unlimited).
+#[inline]
+pub fn try_accept_connection(maxclients: usize) -> bool {
+    if maxclients == 0 {
+        record_connection_opened();
+        return true;
+    }
+    // CAS loop: only increment if under limit.
+    // AcqRel on success ensures the counter increment is visible to other cores
+    // before the connection handler runs (important on ARM/weak-memory archs).
+    let mut current = CONNECTED_CLIENTS.load(Ordering::Acquire);
+    loop {
+        if current >= maxclients as u64 {
+            return false;
+        }
+        match CONNECTED_CLIENTS.compare_exchange_weak(
+            current,
+            current + 1,
+            Ordering::AcqRel,
+            Ordering::Acquire,
+        ) {
+            Ok(_) => {
+                TOTAL_CONNECTIONS.fetch_add(1, Ordering::Relaxed);
+                if METRICS_INITIALIZED.load(Ordering::Relaxed) {
+                    counter!("moon_connections_total").increment(1);
+                    gauge!("moon_connected_clients").increment(1.0);
+                }
+                return true;
+            }
+            Err(actual) => current = actual,
+        }
+    }
+}
+
 // ── Keyspace metrics ────────────────────────────────────────────────────
 
 /// Record keyspace hit/miss.

src/auth_ratelimit.rs

@@ -0,0 +1,132 @@
+//! Per-IP AUTH failure rate limiting.
+//!
+//! Tracks failed AUTH attempts per client IP and enforces exponential backoff
+//! delays to prevent brute-force attacks. Successful AUTH has zero overhead.
+//!
+//! Design:
+//! - Global `parking_lot::Mutex<HashMap>` (not on hot path — only touched on AUTH)
+//! - Exponential backoff: 100ms * 2^(failures-1), capped at 10s
+//! - Auto-reset after 60s of inactivity per IP
+//! - Periodic cleanup of stale entries via `cleanup_stale()`
+
+use parking_lot::Mutex;
+use std::collections::HashMap;
+use std::net::IpAddr;
+use std::sync::LazyLock;
+use std::time::Instant;
+
+/// Base delay for first failure (100ms).
+const BASE_DELAY_MS: u64 = 100;
+
+/// Maximum delay cap (10 seconds).
+const MAX_DELAY_MS: u64 = 10_000;
+
+/// Entries older than this are pruned (60 seconds).
+const STALE_THRESHOLD_SECS: u64 = 60;
+
+/// Maximum entries before forced cleanup.
+const MAX_ENTRIES: usize = 10_000;
+
+struct FailureRecord {
+    count: u32,
+    last_failure: Instant,
+}
+
+static RATE_LIMITER: LazyLock<Mutex<HashMap<IpAddr, FailureRecord>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
+
+/// Record a failed AUTH attempt for the given IP.
+/// Returns the delay in milliseconds that should be applied before
+/// sending the error response.
+pub fn record_failure(ip: IpAddr) -> u64 {
+    let mut map = RATE_LIMITER.lock();
+
+    // Periodic cleanup when map grows too large
+    if map.len() >= MAX_ENTRIES {
+        let cutoff = Instant::now() - std::time::Duration::from_secs(STALE_THRESHOLD_SECS);
+        map.retain(|_, r| r.last_failure > cutoff);
+    }
+
+    let now = Instant::now();
+    let record = map.entry(ip).or_insert(FailureRecord {
+        count: 0,
+        last_failure: now,
+    });
+
+    // Reset if stale (no failures for STALE_THRESHOLD_SECS)
+    if now.duration_since(record.last_failure).as_secs() >= STALE_THRESHOLD_SECS {
+        record.count = 0;
+    }
+
+    record.count = record.count.saturating_add(1);
+    record.last_failure = now;
+
+    // Exponential backoff: 100ms * 2^(count-1), capped at 10s
+    let delay = BASE_DELAY_MS.saturating_mul(
+        1u64.checked_shl(record.count.saturating_sub(1))
+            .unwrap_or(u64::MAX),
+    );
+    delay.min(MAX_DELAY_MS)
+}
+
+/// Clear failure record on successful AUTH (reset for this IP).
+pub fn record_success(ip: IpAddr) {
+    let mut map = RATE_LIMITER.lock();
+    map.remove(&ip);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::Ipv4Addr;
+
+    #[test]
+    fn test_first_failure_returns_base_delay() {
+        let ip = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100));
+        // Clean up from any prior test state
+        record_success(ip);
+        let delay = record_failure(ip);
+        assert_eq!(delay, 100);
+        // Cleanup
+        record_success(ip);
+    }
+
+    #[test]
+    fn test_exponential_backoff() {
+        let ip = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 101));
+        record_success(ip);
+        assert_eq!(record_failure(ip), 100); // 100 * 2^0
+        assert_eq!(record_failure(ip), 200); // 100 * 2^1
+        assert_eq!(record_failure(ip), 400); // 100 * 2^2
+        assert_eq!(record_failure(ip), 800); // 100 * 2^3
+        assert_eq!(record_failure(ip), 1600); // 100 * 2^4
+        record_success(ip);
+    }
+
+    #[test]
+    fn test_max_delay_cap() {
+        let ip = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 102));
+        record_success(ip);
+        // 7 failures: 100, 200, 400, 800, 1600, 3200, 6400
+        for _ in 0..7 {
+            record_failure(ip);
+        }
+        // 8th failure should be capped at 10000
+        let delay = record_failure(ip);
+        assert!(delay <= MAX_DELAY_MS);
+        record_success(ip);
+    }
+
+    #[test]
+    fn test_success_clears_record() {
+        let ip = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 103));
+        record_success(ip);
+        record_failure(ip);
+        record_failure(ip);
+        record_success(ip);
+        // After success, next failure should be back to base
+        let delay = record_failure(ip);
+        assert_eq!(delay, 100);
+        record_success(ip);
+    }
+}

src/client_pause.rs

@@ -0,0 +1,104 @@
+//! CLIENT PAUSE / UNPAUSE global state.
+//!
+//! When paused, command processing is delayed for all clients until the pause
+//! expires or CLIENT UNPAUSE is called. Supports two modes:
+//! - ALL: pause all commands
+//! - WRITE: pause only write commands (reads still served)
+
+use parking_lot::RwLock;
+use std::sync::LazyLock;
+use std::time::Instant;
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+pub enum PauseMode {
+    All,
+    Write,
+}
+
+struct PauseState {
+    active: bool,
+    mode: PauseMode,
+    until: Instant,
+}
+
+static PAUSE: LazyLock<RwLock<PauseState>> = LazyLock::new(|| {
+    RwLock::new(PauseState {
+        active: false,
+        mode: PauseMode::All,
+        until: Instant::now(),
+    })
+});
+
+/// Activate pause for the given duration and mode.
+pub fn pause(duration_ms: u64, mode: PauseMode) {
+    let mut state = PAUSE.write();
+    state.active = true;
+    state.mode = mode;
+    state.until = Instant::now() + std::time::Duration::from_millis(duration_ms);
+}
+
+/// Deactivate pause immediately.
+pub fn unpause() {
+    let mut state = PAUSE.write();
+    state.active = false;
+}
+
+/// Check if the server is currently paused. Returns the remaining duration
+/// if paused, or None if not paused. Auto-expires when the deadline passes.
+pub fn check_pause(is_write: bool) -> Option<std::time::Duration> {
+    let state = PAUSE.read();
+    if !state.active {
+        return None;
+    }
+    let now = Instant::now();
+    if now >= state.until {
+        // Expired — will be cleaned up on next write access
+        return None;
+    }
+    // In WRITE mode, only pause write commands
+    if state.mode == PauseMode::Write && !is_write {
+        return None;
+    }
+    Some(state.until - now)
+}
+
+/// Clean up expired pause state (called periodically from handlers).
+/// Takes a write lock to avoid TOCTOU between expiry check and clear.
+pub fn expire_if_needed() {
+    let mut state = PAUSE.write();
+    if state.active && Instant::now() >= state.until {
+        state.active = false;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // Tests share global state — run sequentially via a mutex
+    static TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+    #[test]
+    fn test_pause_and_check() {
+        let _lock = TEST_LOCK.lock();
+        unpause();
+        assert!(check_pause(true).is_none());
+
+        pause(5000, PauseMode::All);
+        assert!(check_pause(true).is_some());
+        assert!(check_pause(false).is_some());
+
+        unpause();
+        assert!(check_pause(true).is_none());
+    }
+
+    #[test]
+    fn test_write_mode_allows_reads() {
+        let _lock = TEST_LOCK.lock();
+        unpause();
+        pause(5000, PauseMode::Write);
+        assert!(check_pause(true).is_some()); // write blocked
+        assert!(check_pause(false).is_none()); // read allowed
+        unpause();
+    }
+}