fix(audit-2026-06-11/H-B1): login rate limit + Argon2 concurrency cap

The login handler had no admission control before
`spawn_blocking(verify_password)`. On Pi-class hardware Argon2id is ~50-
100 ms per attempt, so a few dozen concurrent anonymous POSTs to
/auth/login park every blocking worker, wedging the entire admin API
and any other path that uses `spawn_blocking`.

Adds two cheap, in-process guards modeled on EmailRateLimiter:

* `LoginRateLimiter` — two sliding-window token buckets, one per
  `(remote_ip, username)` (burst 5/60s) and one per `username`
  (burst 10/15min). Per-(ip, user) defeats a single attacker pounding
  one account; per-user defeats credential-stuffing distributed across
  many IPs. Username is case-folded so case variants share a bucket.
  Maps GC lazily when they cross a soft size cap.

* Per-process `tokio::sync::Semaphore` — caps concurrent Argon2 verifies
  during login. Default 2 permits (Pi-class), overridable via
  `PICLOUD_LOGIN_ARGON2_PARALLELISM`. Acquired AFTER the bucket check so
  attackers can't queue.

Limit check fires before the DB credentials lookup, so even an unknown
username doesn't cost a query. Denied attempts return 429 + Retry-After.
Real client IP comes from the first X-Forwarded-For entry (Caddy is the
trusted single hop); falls back to "unknown" so the per-user bucket
still gates.

4 unit tests cover bucket burst exhaustion, per-user crossing IPs, case
folding, and per-user independence.

Audit ref: security_audit/08_dos_resource.md (H-2 / H-B1).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/manager-core/src/auth_api.rs

@@ -16,7 +16,7 @@
 
 use axum::body::Body;
 use axum::extract::{Extension, Request, State};
-use axum::http::{header, StatusCode};
+use axum::http::{header, HeaderMap, HeaderValue, StatusCode};
 use axum::middleware::from_fn_with_state;
 use axum::response::{IntoResponse, Json, Response};
 use axum::routing::{get, post};
@@ -30,6 +30,7 @@ use picloud_shared::Principal;
 
 use crate::auth::{generate_session_token, hash_token, verify_password};
 use crate::auth_middleware::{require_authenticated, AuthState};
+use crate::login_rate_limit::LoginRateOutcome;
 
 pub fn auth_router(state: AuthState) -> Router {
     // /login + /logout are unguarded (login is how you get in; logout
@@ -75,7 +76,23 @@ pub struct AdminUserDto {
 // Handlers
 // ----------------------------------------------------------------------------
 
-async fn login(State(state): State<AuthState>, Json(input): Json<LoginRequest>) -> Response {
+async fn login(
+    State(state): State<AuthState>,
+    headers: HeaderMap,
+    Json(input): Json<LoginRequest>,
+) -> Response {
+    // Audit 2026-06-11 H-B1 — rate limit BEFORE any DB read or
+    // Argon2 verify so an attacker can't even queue. Keying is
+    // (X-Forwarded-For first hop, username) + (username), so a single
+    // hot user and a credential-stuffing flurry both saturate.
+    let remote_ip = extract_remote_ip(&headers);
+    if let LoginRateOutcome::Denied { retry_after } = state
+        .login_rate_limiter
+        .try_consume(&remote_ip, &input.username)
+    {
+        return rate_limited(retry_after);
+    }
+
     // Always perform a verify, even on missing/inactive users, to flatten
     // timing and prevent username enumeration. The dummy hash is the
     // shared `TIMING_FLAT_DUMMY_HASH` constant from `auth.rs` so the
@@ -99,6 +116,18 @@ async fn login(State(state): State<AuthState>, Json(input): Json<LoginRequest>)
         None => (crate::auth::TIMING_FLAT_DUMMY_HASH.to_string(), None, false),
     };
 
+    // Audit 2026-06-11 H-B1 — global Argon2 concurrency cap. Acquired
+    // AFTER the bucket check so attackers can't queue; held only across
+    // the verify, not the surrounding DB I/O. acquire() awaits, but the
+    // bucket already gated so the queue is bounded.
+    let permit = match state.argon2_login_semaphore.clone().acquire_owned().await {
+        Ok(p) => p,
+        Err(_) => {
+            // Semaphore is closed (process shutting down). Fail closed.
+            return internal_error();
+        }
+    };
+
     // F-P-002: Argon2id verify off the async worker.
     let password = input.password.clone();
     let password_ok =
@@ -106,9 +135,11 @@ async fn login(State(state): State<AuthState>, Json(input): Json<LoginRequest>)
             Ok(b) => b,
             Err(err) => {
                 tracing::error!(?err, "verify_password spawn_blocking join failed");
+                drop(permit);
                 return internal_error();
             }
         };
+    drop(permit);
     if !password_ok || user_id.is_none() || !is_active {
         return invalid_credentials();
     }
@@ -212,6 +243,44 @@ fn extract_token_for_logout(req: &Request<Body>) -> Option<String> {
     Some(token.to_string())
 }
 
+/// Pull the originating client IP from the request. Caddy is the only
+/// trusted proxy in front of picloud; X-Forwarded-For carries the
+/// client. We take the first entry (closest to the original client).
+/// Returns `"unknown"` if the header is absent/malformed so the
+/// per-user bucket still gates without an IP qualifier.
+fn extract_remote_ip(headers: &HeaderMap) -> String {
+    if let Some(xff) = headers.get("x-forwarded-for") {
+        if let Ok(s) = xff.to_str() {
+            if let Some(first) = s.split(',').next() {
+                let trimmed = first.trim();
+                if !trimmed.is_empty() {
+                    return trimmed.to_string();
+                }
+            }
+        }
+    }
+    "unknown".to_string()
+}
+
+fn rate_limited(retry_after: std::time::Duration) -> Response {
+    let mut secs = retry_after.as_secs();
+    if secs == 0 {
+        secs = 1;
+    }
+    let mut hdrs = HeaderMap::new();
+    if let Ok(v) = HeaderValue::from_str(&secs.to_string()) {
+        hdrs.insert("retry-after", v);
+    }
+    (
+        StatusCode::TOO_MANY_REQUESTS,
+        hdrs,
+        Json(json!({
+            "error": "too many login attempts; try again later",
+        })),
+    )
+        .into_response()
+}
+
 fn invalid_credentials() -> Response {
     (
         StatusCode::UNAUTHORIZED,

crates/manager-core/src/auth_middleware.rs

@@ -115,6 +115,13 @@ pub struct AuthState {
     /// at startup; cloned cheaply into every router state. `None` for
     /// tests / harnesses that don't wire it.
     pub principal_cache: Arc<PrincipalCache>,
+    /// Audit 2026-06-11 H-B1 — per-`(remote_ip, username)` + per-`username`
+    /// burst limiter for `/auth/login`. Cheap to clone (Arc).
+    pub login_rate_limiter: Arc<crate::login_rate_limit::LoginRateLimiter>,
+    /// Audit 2026-06-11 H-B1 — caps concurrent Argon2 verifies during
+    /// login so a credential-stuffing flurry can't park every blocking
+    /// worker. Sized via `PICLOUD_LOGIN_ARGON2_PARALLELISM` (default 2).
+    pub argon2_login_semaphore: Arc<tokio::sync::Semaphore>,
 }
 
 /// Legacy request-extension alias retained so the (only remaining)

crates/manager-core/src/lib.rs

@@ -49,6 +49,7 @@ pub mod invoke_service;
 pub mod kv_repo;
 pub mod kv_service;
 pub mod log_sink;
+pub mod login_rate_limit;
 pub mod migrations;
 pub mod module_source;
 pub mod outbox_event_emitter;

crates/manager-core/src/login_rate_limit.rs

@@ -0,0 +1,264 @@
+//! Login rate limiter (audit 2026-06-11 H-B1).
+//!
+//! Two sliding-window token buckets gating `POST /api/v1/admin/auth/login`:
+//!
+//! 1. **Per `(remote_ip, username)`** — burst 5 in 60 s. Defeats a single
+//!    attacker hammering one account.
+//! 2. **Per username** — burst 10 in 15 min. Defeats credential-stuffing
+//!    that distributes one username across many IPs.
+//!
+//! Buckets reset lazily when a request arrives in a stale window. Both
+//! caches sweep when they cross a fixed size threshold so an attacker
+//! enumerating usernames can't grow the table unbounded.
+//!
+//! A separate `tokio::sync::Semaphore` caps concurrent Argon2 verifies
+//! (the actual CPU cost) so even a single legitimate user can't park
+//! every blocking-thread on password hashing during a credential-
+//! stuffing flurry. The login handler acquires it AFTER the bucket
+//! check so attackers can't even queue.
+
+use std::collections::HashMap;
+use std::sync::Mutex;
+use std::time::{Duration, Instant};
+
+/// Outcome of [`LoginRateLimiter::try_consume`].
+#[derive(Debug, Clone, Copy)]
+pub enum LoginRateOutcome {
+    /// Both buckets had a slot; this attempt is allowed.
+    Allowed,
+    /// One of the buckets is full. The advisory `retry_after` is the
+    /// soonest a slot will next free up.
+    Denied { retry_after: Duration },
+}
+
+#[derive(Debug, Clone, Copy)]
+struct Bucket {
+    window_started_at: Instant,
+    count: u32,
+}
+
+impl Bucket {
+    fn fresh(now: Instant) -> Self {
+        Self {
+            window_started_at: now,
+            count: 0,
+        }
+    }
+
+    /// Reset the window if `window` has elapsed.
+    fn roll(&mut self, now: Instant, window: Duration) {
+        if now.duration_since(self.window_started_at) >= window {
+            *self = Self::fresh(now);
+        }
+    }
+
+    fn retry_after(&self, now: Instant, window: Duration) -> Duration {
+        window.saturating_sub(now.duration_since(self.window_started_at))
+    }
+}
+
+#[derive(Debug, Default)]
+struct State {
+    per_ip_user: HashMap<(String, String), Bucket>,
+    per_user: HashMap<String, Bucket>,
+}
+
+#[derive(Debug)]
+pub struct LoginRateLimiter {
+    ip_user_burst: u32,
+    ip_user_window: Duration,
+    user_burst: u32,
+    user_window: Duration,
+    state: Mutex<State>,
+}
+
+impl Default for LoginRateLimiter {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl LoginRateLimiter {
+    #[must_use]
+    pub fn new() -> Self {
+        Self {
+            ip_user_burst: 5,
+            ip_user_window: Duration::from_secs(60),
+            user_burst: 10,
+            user_window: Duration::from_secs(15 * 60),
+            state: Mutex::new(State::default()),
+        }
+    }
+
+    /// Test-only constructor with custom budgets.
+    #[doc(hidden)]
+    #[must_use]
+    pub fn with_budgets(
+        ip_user_burst: u32,
+        ip_user_window: Duration,
+        user_burst: u32,
+        user_window: Duration,
+    ) -> Self {
+        Self {
+            ip_user_burst,
+            ip_user_window,
+            user_burst,
+            user_window,
+            state: Mutex::new(State::default()),
+        }
+    }
+
+    /// Attempt to consume one slot. The username is lowercased before
+    /// keying so case-folded variants share a bucket (admin usernames
+    /// are case-insensitive elsewhere). `remote_ip` is whatever the
+    /// handler resolved (X-Forwarded-For first hop, or `"unknown"` if
+    /// it was absent).
+    pub fn try_consume(&self, remote_ip: &str, username: &str) -> LoginRateOutcome {
+        let now = Instant::now();
+        let user_key = username.trim().to_ascii_lowercase();
+        let mut state = self.state.lock().expect("login rate limiter poisoned");
+
+        // Per-username check.
+        {
+            let bucket = state
+                .per_user
+                .entry(user_key.clone())
+                .or_insert_with(|| Bucket::fresh(now));
+            bucket.roll(now, self.user_window);
+            if bucket.count >= self.user_burst {
+                return LoginRateOutcome::Denied {
+                    retry_after: bucket.retry_after(now, self.user_window),
+                };
+            }
+        }
+
+        // Per-(ip, user) check.
+        let ip_key = (remote_ip.to_string(), user_key.clone());
+        {
+            let bucket = state
+                .per_ip_user
+                .entry(ip_key.clone())
+                .or_insert_with(|| Bucket::fresh(now));
+            bucket.roll(now, self.ip_user_window);
+            if bucket.count >= self.ip_user_burst {
+                return LoginRateOutcome::Denied {
+                    retry_after: bucket.retry_after(now, self.ip_user_window),
+                };
+            }
+        }
+
+        // Both buckets have room — bump both.
+        if let Some(b) = state.per_user.get_mut(&user_key) {
+            b.count += 1;
+        }
+        if let Some(b) = state.per_ip_user.get_mut(&ip_key) {
+            b.count += 1;
+        }
+
+        // Bounded growth: sweep expired entries when the maps cross a
+        // soft cap.
+        if state.per_ip_user.len() > 4096 {
+            let win = self.ip_user_window;
+            state
+                .per_ip_user
+                .retain(|_, b| now.duration_since(b.window_started_at) < win);
+        }
+        if state.per_user.len() > 1024 {
+            let win = self.user_window;
+            state
+                .per_user
+                .retain(|_, b| now.duration_since(b.window_started_at) < win);
+        }
+
+        LoginRateOutcome::Allowed
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn allows_up_to_burst_then_denies() {
+        let l = LoginRateLimiter::with_budgets(
+            3,
+            Duration::from_secs(60),
+            100,
+            Duration::from_secs(60),
+        );
+        for _ in 0..3 {
+            assert!(matches!(
+                l.try_consume("1.2.3.4", "alice"),
+                LoginRateOutcome::Allowed
+            ));
+        }
+        assert!(matches!(
+            l.try_consume("1.2.3.4", "alice"),
+            LoginRateOutcome::Denied { .. }
+        ));
+    }
+
+    #[test]
+    fn per_user_bucket_crosses_ips() {
+        let l = LoginRateLimiter::with_budgets(
+            100,
+            Duration::from_secs(60),
+            3,
+            Duration::from_secs(60),
+        );
+        // Three different IPs each try once → the per-user bucket
+        // exhausts after 3, denying the 4th regardless of source IP.
+        assert!(matches!(
+            l.try_consume("1.1.1.1", "alice"),
+            LoginRateOutcome::Allowed
+        ));
+        assert!(matches!(
+            l.try_consume("2.2.2.2", "alice"),
+            LoginRateOutcome::Allowed
+        ));
+        assert!(matches!(
+            l.try_consume("3.3.3.3", "alice"),
+            LoginRateOutcome::Allowed
+        ));
+        assert!(matches!(
+            l.try_consume("4.4.4.4", "alice"),
+            LoginRateOutcome::Denied { .. }
+        ));
+    }
+
+    #[test]
+    fn username_is_case_folded() {
+        let l = LoginRateLimiter::with_budgets(
+            1,
+            Duration::from_secs(60),
+            100,
+            Duration::from_secs(60),
+        );
+        assert!(matches!(
+            l.try_consume("1.2.3.4", "Alice"),
+            LoginRateOutcome::Allowed
+        ));
+        assert!(matches!(
+            l.try_consume("1.2.3.4", "alice"),
+            LoginRateOutcome::Denied { .. }
+        ));
+    }
+
+    #[test]
+    fn distinct_users_have_independent_buckets() {
+        let l = LoginRateLimiter::with_budgets(
+            1,
+            Duration::from_secs(60),
+            100,
+            Duration::from_secs(60),
+        );
+        assert!(matches!(
+            l.try_consume("1.2.3.4", "alice"),
+            LoginRateOutcome::Allowed
+        ));
+        assert!(matches!(
+            l.try_consume("1.2.3.4", "bob"),
+            LoginRateOutcome::Allowed
+        ));
+    }
+}

crates/picloud/src/lib.rs

@@ -488,12 +488,23 @@ pub async fn build_app(
         authz: authz.clone(),
     };
 
+    // Audit 2026-06-11 H-B1 — login DoS defenses. PICLOUD_LOGIN_ARGON2_PARALLELISM
+    // overrides the per-process Argon2-during-login concurrency cap.
+    let argon2_login_parallelism = std::env::var("PICLOUD_LOGIN_ARGON2_PARALLELISM")
+        .ok()
+        .and_then(|s| s.parse::<usize>().ok())
+        .filter(|n| *n > 0)
+        .unwrap_or(2);
     let auth_state = AuthState {
         users: auth.users.clone(),
         sessions: auth.sessions.clone(),
         keys: auth.keys.clone(),
         ttl: auth.ttl,
         principal_cache: Arc::new(picloud_manager_core::auth_middleware::PrincipalCache::new()),
+        login_rate_limiter: Arc::new(
+            picloud_manager_core::login_rate_limit::LoginRateLimiter::new(),
+        ),
+        argon2_login_semaphore: Arc::new(tokio::sync::Semaphore::new(argon2_login_parallelism)),
     };
     let admins_state = AdminsState {
         users: auth.users.clone(),