From 2c4e96e1750c378a6c218e5ba2faaec79dc070e3 Mon Sep 17 00:00:00 2001 From: kanengchik Date: Sun, 30 Aug 2026 08:43:24 +0000 Subject: [PATCH] Add profiler overhead guard, canary baseline rollback, pool leak detection, WebAuthn downgrade protection - profiler.rs: auto-throttle sample rate when recording overhead exceeds 5% of profiled operation time, expose overhead/sample rate via GET /admin/profiler/overhead - canary.rs: compare canary error rate against a continuously reported baseline cohort error rate and auto-rollback when it exceeds baseline by more than the configured margin, independent of absolute thresholds - pool_optimizer.rs: track per-connection checkout timestamps and flag connections held beyond max_checkout_secs as suspected leaks; log and page on-call via a new oncall::raise_alert helper - webauthn.rs: maintain an explicit COSE algorithm allowlist and reject registrations that downgrade a user to a weaker algorithm than one already on file Closes #384 Closes #385 Closes #386 Closes #387 --- backend/src/canary.rs | 167 +++++++++++++++++++++++++++-- backend/src/oncall.rs | 92 ++++++++++++++++ backend/src/pool_optimizer.rs | 184 ++++++++++++++++++++++++++++++++ backend/src/profiler.rs | 164 ++++++++++++++++++++++++++-- backend/src/webauthn.rs | 152 +++++++++++++++++++++++++- docs/canary-deployments.md | 41 +++++-- docs/configuration-reference.md | 24 +++++ docs/profiling-guide.md | 29 +++++ docs/webauthn-setup.md | 21 ++++ 9 files changed, 855 insertions(+), 19 deletions(-) diff --git a/backend/src/canary.rs b/backend/src/canary.rs index 1c852f3f..4d7cec3e 100644 --- a/backend/src/canary.rs +++ b/backend/src/canary.rs @@ -40,6 +40,17 @@ pub struct CanaryStage { pub struct MetricThresholds { pub max_error_rate: f64, pub max_latency_p99_ms: f64, + /// Maximum amount the canary cohort's error rate may exceed the + /// baseline (non-canary) cohort's error rate before an automatic + /// rollback is triggered, independent of `max_error_rate`. E.g. `0.01` + /// means the canary is rolled back once it's more than 1 percentage + /// point worse than baseline. + #[serde(default = "default_error_rate_margin")] + pub max_error_rate_margin: f64, +} + +fn default_error_rate_margin() -> f64 { + 0.01 } impl Default for MetricThresholds { @@ -47,6 +58,7 @@ impl Default for MetricThresholds { Self { max_error_rate: 0.02, max_latency_p99_ms: 500.0, + max_error_rate_margin: default_error_rate_margin(), } } } @@ -85,6 +97,9 @@ pub struct CanaryDeployment { pub thresholds: MetricThresholds, pub status: CanaryStatus, pub last_metrics: Option, + /// Most recently reported error rate for the baseline (non-canary) + /// cohort, used for continuous canary-vs-baseline comparison. + pub last_baseline_error_rate: Option, pub history: Vec, pub started_at: DateTime, pub stage_started_at: DateTime, @@ -122,6 +137,11 @@ pub struct StartCanaryRequest { #[derive(Debug, Deserialize)] pub struct EvaluateCanaryRequest { pub metrics: CanaryMetrics, + /// Current error rate observed in the baseline (non-canary) cohort, for + /// continuous canary-vs-baseline comparison. If omitted, the previously + /// reported baseline error rate (if any) continues to be used. + #[serde(default)] + pub baseline_error_rate: Option, } /// Request body for `POST /deployments/canary/:id/rollback`. @@ -188,6 +208,7 @@ pub async fn start_canary_deployment( thresholds: body.thresholds.unwrap_or_default(), status: CanaryStatus::InProgress, last_metrics: None, + last_baseline_error_rate: None, history: vec![], started_at: now, stage_started_at: now, @@ -229,20 +250,43 @@ pub async fn evaluate_canary( } deployment.last_metrics = Some(body.metrics.clone()); + if let Some(baseline_error_rate) = body.baseline_error_rate { + deployment.last_baseline_error_rate = Some(baseline_error_rate); + } - let breached = body.metrics.error_rate > deployment.thresholds.max_error_rate + let absolute_breach = body.metrics.error_rate > deployment.thresholds.max_error_rate || body.metrics.latency_p99_ms > deployment.thresholds.max_latency_p99_ms; - if breached { + // Continuous canary-vs-baseline comparison: roll back if the canary's + // error rate has drifted more than `max_error_rate_margin` above the + // baseline cohort's error rate, even if it hasn't breached the absolute + // `max_error_rate` threshold on its own. + let baseline_breach = deployment.last_baseline_error_rate.is_some_and(|baseline| { + body.metrics.error_rate > baseline + deployment.thresholds.max_error_rate_margin + }); + + if absolute_breach || baseline_breach { deployment.status = CanaryStatus::RolledBack; - deployment.push_event(format!( - "automated rollback: error_rate={:.4} latency_p99_ms={:.1} breached thresholds", - body.metrics.error_rate, body.metrics.latency_p99_ms - )); + let reason = if baseline_breach { + format!( + "automated rollback: canary error_rate={:.4} exceeded baseline error_rate={:.4} by more than the configured margin ({:.4})", + body.metrics.error_rate, + deployment.last_baseline_error_rate.unwrap_or(0.0), + deployment.thresholds.max_error_rate_margin + ) + } else { + format!( + "automated rollback: error_rate={:.4} latency_p99_ms={:.1} breached thresholds", + body.metrics.error_rate, body.metrics.latency_p99_ms + ) + }; + deployment.push_event(reason); tracing::error!( deployment_id = %id, error_rate = body.metrics.error_rate, latency_p99_ms = body.metrics.latency_p99_ms, + baseline_error_rate = deployment.last_baseline_error_rate, + baseline_breach, "canary rolled back automatically due to metric breach" ); return Ok(Json(deployment.clone())); @@ -285,3 +329,114 @@ pub async fn rollback_canary( Ok(Json(deployment.clone())) } + +#[cfg(test)] +mod tests { + use super::*; + + async fn start_deployment(state: &Arc) -> CanaryDeployment { + let (_, Json(deployment)) = start_canary_deployment( + State(Arc::clone(state)), + Json(StartCanaryRequest { + service: "vault-api".into(), + version: "v1.0.0".into(), + stages: Some(vec![CanaryStage { + traffic_percent: 5, + min_duration_minutes: 60, + }]), + thresholds: None, + }), + ) + .await + .unwrap(); + deployment + } + + #[tokio::test] + async fn rollback_triggers_when_canary_exceeds_baseline_margin() { + let state = Arc::new(CanaryState::new()); + let deployment = start_deployment(&state).await; + + let (_, Json(result)) = evaluate_canary( + State(Arc::clone(&state)), + Path(deployment.id.clone()), + Json(EvaluateCanaryRequest { + metrics: CanaryMetrics { + error_rate: 0.015, + latency_p99_ms: 100.0, + }, + baseline_error_rate: Some(0.001), + }), + ) + .await + .unwrap(); + + assert_eq!(result.status, CanaryStatus::RolledBack); + } + + #[tokio::test] + async fn rollback_does_not_trigger_within_baseline_margin() { + let state = Arc::new(CanaryState::new()); + let deployment = start_deployment(&state).await; + + let (_, Json(result)) = evaluate_canary( + State(Arc::clone(&state)), + Path(deployment.id.clone()), + Json(EvaluateCanaryRequest { + metrics: CanaryMetrics { + error_rate: 0.005, + latency_p99_ms: 100.0, + }, + baseline_error_rate: Some(0.001), + }), + ) + .await + .unwrap(); + + assert_eq!(result.status, CanaryStatus::InProgress); + } + + #[tokio::test] + async fn rollback_still_triggers_on_absolute_threshold_without_baseline() { + let state = Arc::new(CanaryState::new()); + let deployment = start_deployment(&state).await; + + let (_, Json(result)) = evaluate_canary( + State(Arc::clone(&state)), + Path(deployment.id.clone()), + Json(EvaluateCanaryRequest { + metrics: CanaryMetrics { + error_rate: 0.05, + latency_p99_ms: 100.0, + }, + baseline_error_rate: None, + }), + ) + .await + .unwrap(); + + assert_eq!(result.status, CanaryStatus::RolledBack); + } + + #[tokio::test] + async fn no_rollback_when_healthy_and_no_baseline_reported() { + let state = Arc::new(CanaryState::new()); + let deployment = start_deployment(&state).await; + + let (_, Json(result)) = evaluate_canary( + State(Arc::clone(&state)), + Path(deployment.id.clone()), + Json(EvaluateCanaryRequest { + metrics: CanaryMetrics { + error_rate: 0.001, + latency_p99_ms: 100.0, + }, + baseline_error_rate: None, + }), + ) + .await + .unwrap(); + + assert_eq!(result.status, CanaryStatus::InProgress); + } +} diff --git a/backend/src/oncall.rs b/backend/src/oncall.rs index ba9b7c17..5b7c2756 100644 --- a/backend/src/oncall.rs +++ b/backend/src/oncall.rs @@ -159,6 +159,51 @@ impl Default for OnCallState { } } +/// An ad-hoc alert raised against a schedule's primary escalation contacts, +/// outside the normal `trigger_escalation` HTTP flow — used by other +/// subsystems (e.g. connection-pool leak detection) that need to page +/// on-call without going through a pre-defined escalation level. +#[derive(Debug, Clone)] +pub struct AlertRecord { + pub schedule_id: String, + pub source: String, + pub message: String, + pub contacts_notified: Vec, +} + +/// Raise an alert against `schedule_id`'s primary (level-1) escalation +/// contacts and log it. Returns `None` if the schedule doesn't exist, in +/// which case the alert is only logged, not attributed to any contacts. +pub fn raise_alert( + state: &OnCallState, + schedule_id: &str, + source: &str, + message: &str, +) -> Option { + let contacts = { + let store = state.store.lock().unwrap(); + store + .get(schedule_id) + .and_then(|schedule| schedule.escalation_policy.levels.first()) + .map(|level| level.contacts.clone()) + }; + + tracing::error!( + schedule_id = %schedule_id, + source = %source, + message = %message, + contacts = ?contacts, + "alert raised" + ); + + contacts.map(|contacts_notified| AlertRecord { + schedule_id: schedule_id.to_string(), + source: source.to_string(), + message: message.to_string(), + contacts_notified, + }) +} + /// Build a round-robin rotation of `shift_count` shifts across /// `participants`, each `rotation_hours` long, starting now. fn build_rotation( @@ -309,3 +354,50 @@ pub async fn trigger_escalation( reason: body.reason, })) } + +#[cfg(test)] +mod tests { + use super::*; + + fn schedule_with_contacts(contacts: Vec) -> OnCallSchedule { + OnCallSchedule { + id: "sched-1".into(), + name: "Backend On-Call".into(), + rotation_hours: 24, + shifts: vec![], + escalation_policy: EscalationPolicy { + levels: vec![EscalationLevel { + level: 1, + delay_minutes: 5, + contacts, + }], + }, + handoffs: vec![], + created_at: Utc::now(), + } + } + + #[test] + fn raise_alert_notifies_primary_escalation_contacts() { + let state = OnCallState::new(); + let schedule = schedule_with_contacts(vec!["oncall@example.com".into()]); + state + .store + .lock() + .unwrap() + .insert(schedule.id.clone(), schedule.clone()); + + let alert = raise_alert(&state, &schedule.id, "pool_optimizer", "leak detected") + .expect("schedule exists"); + + assert_eq!(alert.contacts_notified, vec!["oncall@example.com".to_string()]); + assert_eq!(alert.source, "pool_optimizer"); + } + + #[test] + fn raise_alert_on_unknown_schedule_returns_none() { + let state = OnCallState::new(); + let alert = raise_alert(&state, "does-not-exist", "pool_optimizer", "leak detected"); + assert!(alert.is_none()); + } +} diff --git a/backend/src/pool_optimizer.rs b/backend/src/pool_optimizer.rs index 38b44150..a6d33ff9 100644 --- a/backend/src/pool_optimizer.rs +++ b/backend/src/pool_optimizer.rs @@ -30,8 +30,10 @@ //! | `DB_POOL_IDLE_TIMEOUT_SECS` | `300` | Idle threshold before a connection is culled | //! | `DB_POOL_MAX_LIFETIME_SECS` | `3600` | Maximum connection age before recycling | //! | `DB_POOL_QUEUE_TIMEOUT_MS` | `5000` | How long a caller waits for a free connection | +//! | `DB_POOL_MAX_CHECKOUT_SECS` | `60` | How long a connection may stay checked out before being flagged as a suspected leak | use rusqlite::Connection; +use std::collections::HashMap; use std::sync::{Arc, Condvar, Mutex}; use std::time::{Duration, Instant}; @@ -52,6 +54,9 @@ pub struct OptimizedPoolConfig { pub max_lifetime_secs: u64, /// How long a caller will block waiting for a free connection (ms). pub queue_timeout_ms: u64, + /// A connection checked out longer than this is flagged as a suspected + /// leak (checked out and never returned). + pub max_checkout_secs: u64, /// Path to the SQLite database (`:memory:` for tests). pub db_path: String, } @@ -65,6 +70,7 @@ impl Default for OptimizedPoolConfig { idle_timeout_secs: 300, max_lifetime_secs: 3600, queue_timeout_ms: 5000, + max_checkout_secs: 60, db_path: ":memory:".to_string(), } } @@ -97,6 +103,10 @@ impl OptimizedPoolConfig { .ok() .and_then(|v| v.parse().ok()) .unwrap_or(5000), + max_checkout_secs: std::env::var("DB_POOL_MAX_CHECKOUT_SECS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(60), db_path: std::env::var("DB_PATH").unwrap_or_else(|_| ":memory:".to_string()), } } @@ -149,6 +159,19 @@ pub struct PoolMetrics { pub min: u32, /// Current configured maximum. pub max: u32, + /// Number of currently checked-out connections held longer than + /// `max_checkout_secs` — suspected leaks. + pub suspected_leaks: u32, +} + +/// A connection checked out longer than `max_checkout_secs` without being +/// returned — a suspected connection leak. +#[derive(Debug, Clone)] +pub struct LeakFinding { + /// ID of the suspected-leaked connection. + pub connection_id: u64, + /// How long the connection has been checked out, in seconds. + pub held_for_secs: u64, } // ── Pool inner state ────────────────────────────────────────────────────────── @@ -169,6 +192,10 @@ struct Inner { idle_culled: u64, /// Simulated in-use count (updated on acquire/release). in_use_count: u32, + /// Checkout time of every currently-leased connection, by connection ID. + /// Populated on acquire, cleared on release; anything still present + /// beyond `max_checkout_secs` is a suspected leak. + checked_out: HashMap, } impl Inner { @@ -184,6 +211,7 @@ impl Inner { recycled_connections: 0, idle_culled: 0, in_use_count: 0, + checked_out: HashMap::new(), } } @@ -255,8 +283,24 @@ impl Inner { idle_culled: self.idle_culled, min: self.config.min, max: self.config.max, + suspected_leaks: self.detect_leaks().len() as u32, } } + + /// Connections currently checked out longer than `max_checkout_secs`. + fn detect_leaks(&self) -> Vec { + let max = Duration::from_secs(self.config.max_checkout_secs); + self.checked_out + .iter() + .filter_map(|(id, checked_out_at)| { + let held_for = checked_out_at.elapsed(); + (held_for >= max).then_some(LeakFinding { + connection_id: *id, + held_for_secs: held_for.as_secs(), + }) + }) + .collect() + } } // ── Pool ────────────────────────────────────────────────────────────────────── @@ -328,6 +372,7 @@ impl OptimizedConnectionPool { pc.meta.last_used_at = Instant::now(); inner.total_acquires += 1; inner.in_use_count += 1; + inner.checked_out.insert(pc.meta.id, Instant::now()); return Ok(PoolGuard { conn: Some(pc), pool: Arc::clone(&self.inner), @@ -341,6 +386,7 @@ impl OptimizedConnectionPool { pc.meta.total_uses += 1; inner.total_acquires += 1; inner.in_use_count += 1; + inner.checked_out.insert(pc.meta.id, Instant::now()); return Ok(PoolGuard { conn: Some(pc), pool: Arc::clone(&self.inner), @@ -381,6 +427,42 @@ impl OptimizedConnectionPool { lock.lock().unwrap().metrics() } + /// List connections currently checked out longer than + /// `max_checkout_secs` — suspected leaks (checked out and never + /// returned). Should be called periodically alongside [`Self::maintain`]. + pub fn detect_leaks(&self) -> Vec { + let (lock, _) = &*self.inner; + lock.lock().unwrap().detect_leaks() + } + + /// Run leak detection and, for anything found, log an error and raise an + /// on-call alert via [`crate::oncall::raise_alert`]. + /// + /// `schedule_id` identifies which on-call schedule (`OnCallSchedule::id`) + /// should be paged. + pub fn check_for_leaks_and_alert( + &self, + oncall_state: &crate::oncall::OnCallState, + schedule_id: &str, + ) { + for leak in self.detect_leaks() { + tracing::error!( + connection_id = leak.connection_id, + held_for_secs = leak.held_for_secs, + "suspected leaked database connection: checked out and not returned" + ); + crate::oncall::raise_alert( + oncall_state, + schedule_id, + "pool_optimizer", + &format!( + "connection {} has been checked out for {}s without being returned", + leak.connection_id, leak.held_for_secs + ), + ); + } + } + /// Current number of idle connections. pub fn idle_count(&self) -> usize { let (lock, _) = &*self.inner; @@ -420,6 +502,7 @@ impl Drop for PoolGuard<'_> { let (lock, cvar) = &*self.pool; let mut inner = lock.lock().unwrap(); inner.in_use_count = inner.in_use_count.saturating_sub(1); + inner.checked_out.remove(&pc.meta.id); inner.idle.push(pc); cvar.notify_one(); } @@ -496,6 +579,7 @@ mod tests { idle_timeout_secs: 300, max_lifetime_secs: 3600, queue_timeout_ms: 200, + max_checkout_secs: 60, db_path: ":memory:".to_string(), }; let pool = OptimizedConnectionPool::new(config).expect("pool"); @@ -588,6 +672,7 @@ mod tests { idle_timeout_secs: 0, // instant idle timeout max_lifetime_secs: 3600, queue_timeout_ms: 200, + max_checkout_secs: 60, db_path: ":memory:".to_string(), }; let pool = OptimizedConnectionPool::new(config).expect("pool"); @@ -640,4 +725,103 @@ mod tests { // All connections returned, idle >= min. assert!(pool.idle_count() >= 2); } + + #[test] + fn test_leaked_connection_is_detected() { + let config = OptimizedPoolConfig { + min: 1, + max: 5, + timeout_secs: 5, + idle_timeout_secs: 300, + max_lifetime_secs: 3600, + queue_timeout_ms: 200, + max_checkout_secs: 0, // anything checked out is immediately "leaked" + db_path: ":memory:".to_string(), + }; + let pool = OptimizedConnectionPool::new(config).expect("pool"); + pool.prefill().expect("prefill"); + + let guard = pool.acquire().expect("acquire"); + std::thread::sleep(Duration::from_millis(5)); + + let leaks = pool.detect_leaks(); + assert_eq!(leaks.len(), 1); + + drop(guard); + // Returned connections are no longer flagged. + assert!(pool.detect_leaks().is_empty()); + } + + #[test] + fn test_connection_within_max_checkout_not_flagged() { + let pool = make_pool(1, 5); // max_checkout_secs: 60 + let _guard = pool.acquire().expect("acquire"); + assert!(pool.detect_leaks().is_empty()); + } + + #[test] + fn test_metrics_reflect_suspected_leaks() { + let config = OptimizedPoolConfig { + min: 1, + max: 5, + timeout_secs: 5, + idle_timeout_secs: 300, + max_lifetime_secs: 3600, + queue_timeout_ms: 200, + max_checkout_secs: 0, + db_path: ":memory:".to_string(), + }; + let pool = OptimizedConnectionPool::new(config).expect("pool"); + pool.prefill().expect("prefill"); + + let _guard = pool.acquire().expect("acquire"); + std::thread::sleep(Duration::from_millis(5)); + + assert_eq!(pool.metrics().suspected_leaks, 1); + } + + #[test] + fn test_check_for_leaks_and_alert_pages_oncall() { + use crate::oncall::{EscalationLevel, EscalationPolicy, OnCallSchedule, OnCallState}; + + let config = OptimizedPoolConfig { + min: 1, + max: 5, + timeout_secs: 5, + idle_timeout_secs: 300, + max_lifetime_secs: 3600, + queue_timeout_ms: 200, + max_checkout_secs: 0, + db_path: ":memory:".to_string(), + }; + let pool = OptimizedConnectionPool::new(config).expect("pool"); + pool.prefill().expect("prefill"); + let _guard = pool.acquire().expect("acquire"); + std::thread::sleep(Duration::from_millis(5)); + + let oncall_state = OnCallState::new(); + let schedule = OnCallSchedule { + id: "db-oncall".into(), + name: "Database On-Call".into(), + rotation_hours: 24, + shifts: vec![], + escalation_policy: EscalationPolicy { + levels: vec![EscalationLevel { + level: 1, + delay_minutes: 5, + contacts: vec!["dba@example.com".into()], + }], + }, + handoffs: vec![], + created_at: chrono::Utc::now(), + }; + oncall_state + .store + .lock() + .unwrap() + .insert(schedule.id.clone(), schedule); + + // Should not panic and should log/alert for the one leaked connection. + pool.check_for_leaks_and_alert(&oncall_state, "db-oncall"); + } } diff --git a/backend/src/profiler.rs b/backend/src/profiler.rs index aff35764..a6084776 100644 --- a/backend/src/profiler.rs +++ b/backend/src/profiler.rs @@ -22,8 +22,15 @@ //! - `GET /admin/profiler/flamegraph` — folded-stack flame graph data //! - `POST /admin/profiler/baseline` — record current averages as baseline //! - `GET /admin/profiler/regressions` — operations that regressed vs baseline +//! - `GET /admin/profiler/overhead` — current profiler overhead % and sample rate +//! +//! Recording overhead is tracked continuously; if it exceeds +//! [`DEFAULT_OVERHEAD_THRESHOLD_PCT`] of profiled operation time, the sample +//! rate is automatically halved (down to a floor) so the profiler cannot +//! itself become a production bottleneck. use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Instant; @@ -38,6 +45,18 @@ const MAX_SAMPLES: usize = 5_000; /// a performance regression. const DEFAULT_REGRESSION_THRESHOLD_PCT: f64 = 20.0; +/// If the profiler's own recording overhead exceeds this percentage of the +/// time spent in profiled operations, the sample rate is automatically +/// reduced so profiling doesn't itself become a bottleneck. +const DEFAULT_OVERHEAD_THRESHOLD_PCT: f64 = 5.0; + +/// Sample rate is tracked in parts-per-thousand: 1000 = sample every call. +const INITIAL_SAMPLE_RATE_PER_MILLE: u64 = 1000; + +/// Floor below which the auto-throttle will not reduce the sample rate +/// further, so profiling never goes fully blind. +const MIN_SAMPLE_RATE_PER_MILLE: u64 = 10; + /// A single recorded profiling sample. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ProfileSample { @@ -58,11 +77,26 @@ pub struct RegressionFinding { pub sample_count: usize, } +/// Running totals used to compute the profiler's own overhead as a +/// percentage of profiled operation time. +#[derive(Default)] +struct OverheadTracker { + overhead_ms_sum: f64, + operation_ms_sum: f64, +} + /// Continuous profiling state shared across the application. pub struct ProfilerState { samples: Mutex>, /// Recorded baseline average duration (ms) per operation. baseline: Mutex>, + /// Current sample rate, in parts-per-thousand (1000 = sample every call). + /// Reduced automatically when recording overhead exceeds the configured + /// threshold. + sample_rate_per_mille: AtomicU64, + /// Monotonic counter used to decide, per call, whether to sample. + call_counter: AtomicU64, + overhead: Mutex, } impl ProfilerState { @@ -70,7 +104,59 @@ impl ProfilerState { Self { samples: Mutex::new(Vec::new()), baseline: Mutex::new(HashMap::new()), + sample_rate_per_mille: AtomicU64::new(INITIAL_SAMPLE_RATE_PER_MILLE), + call_counter: AtomicU64::new(0), + overhead: Mutex::new(OverheadTracker::default()), + } + } + + /// Decide whether the current call should be sampled, given the current + /// (possibly auto-throttled) sample rate. + pub fn should_sample(&self) -> bool { + let rate = self.sample_rate_per_mille.load(Ordering::Relaxed); + if rate >= 1000 { + return true; + } + let n = self.call_counter.fetch_add(1, Ordering::Relaxed); + (n % 1000) < rate + } + + /// Current sample rate, in parts-per-thousand. + pub fn current_sample_rate_per_mille(&self) -> u64 { + self.sample_rate_per_mille.load(Ordering::Relaxed) + } + + /// Record how long recording itself took (`overhead_ms`) relative to the + /// operation it was profiling (`operation_ms`), and auto-throttle the + /// sample rate if the accumulated overhead exceeds the threshold. + fn note_overhead(&self, overhead_ms: f64, operation_ms: f64) { + let mut tracker = self.overhead.lock().unwrap(); + tracker.overhead_ms_sum += overhead_ms; + tracker.operation_ms_sum += operation_ms; + + if tracker.operation_ms_sum > 0.0 { + let pct = (tracker.overhead_ms_sum / tracker.operation_ms_sum) * 100.0; + if pct > DEFAULT_OVERHEAD_THRESHOLD_PCT { + self.throttle(); + } + } + } + + /// Halve the sample rate, down to `MIN_SAMPLE_RATE_PER_MILLE`. + fn throttle(&self) { + let current = self.sample_rate_per_mille.load(Ordering::Relaxed); + let reduced = (current / 2).max(MIN_SAMPLE_RATE_PER_MILLE); + self.sample_rate_per_mille.store(reduced, Ordering::Relaxed); + } + + /// Current profiler recording overhead as a percentage of total profiled + /// operation time, for exposure as a metric. + pub fn current_overhead_pct(&self) -> f64 { + let tracker = self.overhead.lock().unwrap(); + if tracker.operation_ms_sum <= 0.0 { + return 0.0; } + (tracker.overhead_ms_sum / tracker.operation_ms_sum) * 100.0 } /// Record a completed profiling sample, evicting the oldest sample if @@ -193,12 +279,17 @@ where let result = f().await; let duration_ms = start.elapsed().as_secs_f64() * 1000.0; - state.record(ProfileSample { - operation: operation.to_string(), - stack: stack.iter().map(|s| s.to_string()).collect(), - duration_ms, - recorded_at: Utc::now(), - }); + if state.should_sample() { + let record_start = Instant::now(); + state.record(ProfileSample { + operation: operation.to_string(), + stack: stack.iter().map(|s| s.to_string()).collect(), + duration_ms, + recorded_at: Utc::now(), + }); + let overhead_ms = record_start.elapsed().as_secs_f64() * 1000.0; + state.note_overhead(overhead_ms, duration_ms); + } result } @@ -236,6 +327,23 @@ pub async fn get_regressions( Json(state.detect_regressions(threshold)) } +/// Current profiler recording overhead, exposed for dashboards/alerting. +#[derive(Debug, Serialize)] +pub struct OverheadReport { + /// Recording overhead as a percentage of total profiled operation time. + pub overhead_pct: f64, + /// Current sample rate, in parts-per-thousand (1000 = every call). + pub sample_rate_per_mille: u64, +} + +/// `GET /admin/profiler/overhead` — current profiler overhead and sample rate. +pub async fn get_overhead(State(state): State>) -> Json { + Json(OverheadReport { + overhead_pct: state.current_overhead_pct(), + sample_rate_per_mille: state.current_sample_rate_per_mille(), + }) +} + #[cfg(test)] mod tests { use super::*; @@ -291,4 +399,48 @@ mod tests { let folded = state.flamegraph_folded(); assert!(folded.contains("handler;vault.create 15")); } + + #[test] + fn sample_rate_starts_at_full() { + let state = ProfilerState::new(); + assert_eq!(state.current_sample_rate_per_mille(), 1000); + assert!(state.should_sample()); + } + + #[test] + fn overhead_above_threshold_throttles_sample_rate() { + let state = ProfilerState::new(); + // Recording cost is 10% of operation time, well above the 5% default + // threshold, so the sample rate should be reduced. + state.note_overhead(1.0, 10.0); + assert!(state.current_sample_rate_per_mille() < 1000); + } + + #[test] + fn overhead_below_threshold_does_not_throttle() { + let state = ProfilerState::new(); + // Recording cost is 1% of operation time, below the 5% threshold. + state.note_overhead(0.1, 10.0); + assert_eq!(state.current_sample_rate_per_mille(), 1000); + } + + #[test] + fn repeated_high_overhead_throttles_toward_floor() { + let state = ProfilerState::new(); + for _ in 0..20 { + state.note_overhead(5.0, 10.0); + } + assert_eq!( + state.current_sample_rate_per_mille(), + MIN_SAMPLE_RATE_PER_MILLE + ); + } + + #[test] + fn current_overhead_pct_reflects_accumulated_ratio() { + let state = ProfilerState::new(); + state.note_overhead(1.0, 10.0); + state.note_overhead(1.0, 10.0); + assert!((state.current_overhead_pct() - 10.0).abs() < 1e-9); + } } diff --git a/backend/src/webauthn.rs b/backend/src/webauthn.rs index 74f0b420..f24763a7 100644 --- a/backend/src/webauthn.rs +++ b/backend/src/webauthn.rs @@ -67,10 +67,36 @@ pub enum CoseAlgorithm { EdDsa = -8, } +/// Algorithms this server accepts for new credential registrations. Removing +/// a variant here (e.g. an algorithm later found to be weak or deprecated) +/// causes registrations declaring it to be rejected without touching the +/// COSE parsing / signature-verification code paths. +const ALLOWED_ALGORITHMS: &[CoseAlgorithm] = &[ + CoseAlgorithm::EdDsa, + CoseAlgorithm::ES256, + CoseAlgorithm::RS256, +]; + impl CoseAlgorithm { pub fn cose_id(self) -> i32 { self as i32 } + + /// Whether this algorithm is currently accepted for new registrations. + pub fn is_allowed(self) -> bool { + ALLOWED_ALGORITHMS.contains(&self) + } + + /// Relative cryptographic strength, higher is stronger. Used to detect + /// downgrade attempts: registering a new credential with a weaker + /// algorithm than one already registered for the same user. + fn strength(self) -> u8 { + match self { + CoseAlgorithm::EdDsa => 3, + CoseAlgorithm::ES256 => 2, + CoseAlgorithm::RS256 => 1, + } + } } // ── Data types ──────────────────────────────────────────────────────────────── @@ -723,7 +749,20 @@ pub async fn complete_registration( let (_public_key, algorithm) = parse_cose_public_key(&attested.cose_key_bytes) .map_err(|e| bad_request(format!("could not parse public key: {e}")))?; - // 6. Check for duplicate credential IDs across all users. + if !algorithm.is_allowed() { + return Err(( + StatusCode::BAD_REQUEST, + Json(serde_json::json!({ + "error": format!("algorithm {algorithm:?} is not permitted for registration") + })), + )); + } + + // 6. Check for duplicate credential IDs across all users, and reject a + // downgrade: this user must not register a credential using an + // algorithm weaker than one they've already registered, which would + // let an attacker add a weak-algorithm credential to an account that + // has moved to a stronger one. { let store = state.credentials.lock().unwrap(); for creds in store.values() { @@ -734,6 +773,19 @@ pub async fn complete_registration( )); } } + + if let Some(existing) = store.get(&pending.user_id) { + if let Some(strongest) = existing.iter().map(|c| c.algorithm.strength()).max() { + if algorithm.strength() < strongest { + return Err(( + StatusCode::CONFLICT, + Json(serde_json::json!({ + "error": "algorithm downgrade rejected: a stronger algorithm was already registered for this account" + })), + )); + } + } + } } // 7. Persist credential. @@ -1372,6 +1424,104 @@ mod tests { (client_data, auth_data, b64url_encode(&signature)) } + #[test] + fn test_allowed_algorithms_permit_all_current_variants() { + assert!(CoseAlgorithm::ES256.is_allowed()); + assert!(CoseAlgorithm::RS256.is_allowed()); + assert!(CoseAlgorithm::EdDsa.is_allowed()); + } + + /// Registering a weaker RS256 credential for a user who already has a + /// stronger ES256 credential must be rejected as a downgrade attempt. + #[tokio::test] + async fn test_algorithm_downgrade_after_stronger_registration_rejected() { + let state = make_state(); + let user_id = "user5"; + + register_es256_credential(&state, user_id, "Eve", 0xEE).await; + + let begin = begin_registration( + State(Arc::clone(&state)), + Json(BeginRegistrationRequest { + user_id: user_id.into(), + user_name: "Eve".into(), + label: None, + is_backup: false, + }), + ) + .await + .unwrap(); + let (_, Json(begin)) = begin; + + let client_data = + client_data_json("webauthn.create", &begin.challenge, "http://localhost:3000"); + let cred_id = b64url_encode(&[0xFF; 32]); + let cred_id_bytes = b64url_decode(&cred_id).unwrap(); + let reg_auth_data = registration_auth_data(&cred_id_bytes, &cose_key_rs256()); + + let result = complete_registration( + State(Arc::clone(&state)), + Json(CompleteRegistrationRequest { + session_id: begin.session_id, + credential_id: cred_id, + client_data_json: client_data, + attestation_object: attestation_object(®_auth_data), + label: None, + is_backup: false, + }), + ) + .await; + + assert!(result.is_err()); + let (code, _) = result.unwrap_err(); + assert_eq!(code, StatusCode::CONFLICT); + } + + /// Registering a stronger ES256 credential for a user who already has a + /// weaker RS256 credential must still be allowed (only downgrades are + /// blocked). + #[tokio::test] + async fn test_stronger_algorithm_allowed_after_weaker_registration() { + let state = make_state(); + let user_id = "user6"; + + let begin = begin_registration( + State(Arc::clone(&state)), + Json(BeginRegistrationRequest { + user_id: user_id.into(), + user_name: "Frank".into(), + label: None, + is_backup: false, + }), + ) + .await + .unwrap(); + let (_, Json(begin)) = begin; + + let client_data = + client_data_json("webauthn.create", &begin.challenge, "http://localhost:3000"); + let cred_id = b64url_encode(&[0x11; 32]); + let cred_id_bytes = b64url_decode(&cred_id).unwrap(); + let reg_auth_data = registration_auth_data(&cred_id_bytes, &cose_key_rs256()); + + complete_registration( + State(Arc::clone(&state)), + Json(CompleteRegistrationRequest { + session_id: begin.session_id, + credential_id: cred_id, + client_data_json: client_data, + attestation_object: attestation_object(®_auth_data), + label: None, + is_backup: false, + }), + ) + .await + .unwrap(); + + let result = register_es256_credential(&state, user_id, "Frank", 0x22).await; + let _ = result; + } + #[tokio::test] async fn test_begin_registration_empty_user_id() { let state = make_state(); diff --git a/docs/canary-deployments.md b/docs/canary-deployments.md index 23ae2e9e..60dd4099 100644 --- a/docs/canary-deployments.md +++ b/docs/canary-deployments.md @@ -48,21 +48,50 @@ status, and history of progression events. Reports the latest observed metrics for the canary's traffic slice: ```json -{ "metrics": { "error_rate": 0.004, "latency_p99_ms": 210 } } +{ + "metrics": { "error_rate": 0.004, "latency_p99_ms": 210 }, + "baseline_error_rate": 0.002 +} ``` +`baseline_error_rate` is the current error rate observed in the *baseline* +(non-canary) cohort at the same point in time — reporting it on every call is +what makes the canary-vs-baseline comparison continuous. It's optional; if +omitted, the deployment keeps using the last baseline error rate it was +told about (or skips the baseline comparison if none has ever been reported). + This is the metric-based progression and automated-rollback entry point: -- If `error_rate` or `latency_p99_ms` breaches `thresholds`, the deployment - is immediately marked `rolled_back` — this is the **automated rollback on - errors** behavior. +- If `error_rate` or `latency_p99_ms` breaches `thresholds` (`max_error_rate` / + `max_latency_p99_ms`), the deployment is immediately marked `rolled_back`. +- **Automatic rollback on error-rate spike relative to baseline**: even if + `error_rate` doesn't breach the absolute `max_error_rate` threshold, the + deployment is also rolled back once it exceeds the reported + `baseline_error_rate` by more than `thresholds.max_error_rate_margin` + (default `0.01`, i.e. 1 percentage point). This catches regressions that + are only visible relative to what the rest of the fleet is currently + experiencing (e.g. during a broader incident where both cohorts' error + rates are elevated, but the canary's is meaningfully worse). - Otherwise, once the current stage's `min_duration_minutes` has elapsed, the deployment advances to the next stage (or `completed` if it was the last stage) — this is **metric-based progression**. +Both the triggering condition and the current baseline are recorded in the +deployment's `history` and returned in `last_baseline_error_rate` / +`last_metrics` on `GET /deployments/canary/:id`. + A monitoring loop (e.g. reusing the polling pattern in `scheduler.rs`) -should call this endpoint periodically with live metrics for each active -canary. +should call this endpoint periodically with live metrics for both the canary +and baseline cohorts for each active canary. + +### Configuring the auto-rollback margin + +`thresholds.max_error_rate_margin` is set per-deployment via +`POST /deployments/canary` (`thresholds` field), alongside `max_error_rate` +and `max_latency_p99_ms`. It defaults to `0.01`. Set it tighter for +high-traffic, low-error-rate services where even a small relative regression +is significant, and looser for noisy, low-traffic services where baseline +error rate naturally fluctuates. ### `POST /deployments/canary/:id/rollback` diff --git a/docs/configuration-reference.md b/docs/configuration-reference.md index 7fcfcb94..b047e31e 100644 --- a/docs/configuration-reference.md +++ b/docs/configuration-reference.md @@ -152,6 +152,9 @@ REMINDER_SECOND_LEAD_TIME_HOURS=24 | `DB_MIN_CONNECTIONS` | `integer` | `1` | Minimum idle connections | | `DB_CONNECT_TIMEOUT_SECONDS` | `integer` | `30` | Connection timeout | | `DB_IDLE_TIMEOUT_SECONDS` | `integer` | `600` | Idle connection timeout | +| `DB_POOL_MIN` | `integer` | `2` | Minimum live connections (adaptive pool, `pool_optimizer.rs`) | +| `DB_POOL_MAX` | `integer` | `10` | Maximum live connections (adaptive pool, `pool_optimizer.rs`) | +| `DB_POOL_MAX_CHECKOUT_SECS` | `integer` | `60` | How long a connection may stay checked out before being flagged as a suspected leak | **Recommended**: @@ -162,6 +165,27 @@ DB_MAX_CONNECTIONS=10 **Validation**: `DATABASE_URL` must be a valid PostgreSQL URI. The schema must be initialized before starting the backend. +### Connection leak detection + +`backend/src/pool_optimizer.rs`'s adaptive connection pool tracks a checkout +timestamp for every connection it hands out. A connection still checked out +longer than `DB_POOL_MAX_CHECKOUT_SECS` (default `60`) is treated as a +suspected leak — code that acquired a connection and never returned it +(e.g. via an early `return` that skips dropping the guard, or a stuck +downstream call holding it indefinitely). + +- `OptimizedConnectionPool::detect_leaks()` returns the list of suspected + leaks (connection ID + how long it's been held). +- `PoolMetrics::suspected_leaks` exposes the current count for dashboards. +- `OptimizedConnectionPool::check_for_leaks_and_alert(oncall_state, + schedule_id)` logs an error for each suspected leak and pages the given + on-call schedule via `oncall::raise_alert` (`backend/src/oncall.rs`). Call + this periodically (e.g. alongside `maintain()`) from a background task. + +Set `DB_POOL_MAX_CHECKOUT_SECS` above your slowest expected query/transaction +duration to avoid false positives; lower it in environments where connection +leaks are a known risk (e.g. after a recent incident) to catch them sooner. + --- ### Authentication diff --git a/docs/profiling-guide.md b/docs/profiling-guide.md index 492262dc..70f650cb 100644 --- a/docs/profiling-guide.md +++ b/docs/profiling-guide.md @@ -45,6 +45,35 @@ Each line's weight is the cumulative milliseconds spent in that exact call stack across all recorded samples, so wider frames represent more total time spent, matching standard flame graph semantics. +## Overhead-based sample throttling + +Profiling isn't free — recording a sample (locking + pushing into the ring +buffer) has its own cost. If profiling were left always-on at 100% sampling +in a very hot, very cheap operation, that recording cost could itself become +a measurable fraction of request time. + +`ProfilerState` tracks the ratio of recording overhead to profiled operation +time on every call. If that ratio exceeds **5%**, the sample rate is +automatically halved (starting from 100%, i.e. every call), down to a floor +of **1%**, so the profiler backs off under pressure instead of adding to it. + +Check current overhead and sample rate at any time: + +``` +GET /admin/profiler/overhead +``` + +```json +{ "overhead_pct": 1.8, "sample_rate_per_mille": 1000 } +``` + +`sample_rate_per_mille` is parts-per-thousand (`1000` = sample every call, +`10` = the floor of 1%). Safe sampling rates depend on operation volume and +duration — for high-throughput, sub-millisecond operations, expect the +auto-throttle to reduce sampling; for slower operations (tens of +milliseconds or more) recording overhead is normally negligible and sampling +stays at 100%. + ## Performance regression detection 1. Establish a baseline after a known-good deploy: `POST /admin/profiler/baseline`. diff --git a/docs/webauthn-setup.md b/docs/webauthn-setup.md index 4f388243..9053ff91 100644 --- a/docs/webauthn-setup.md +++ b/docs/webauthn-setup.md @@ -320,3 +320,24 @@ await fetch('/webauthn/register/complete', { > validation for formats like `"packed"` or `"tpm"`) is not performed. The server requests > `attestation: "none"`, so only the credential's own public key is cryptographically > verified — not the authenticator manufacturer's attestation certificate. + +--- + +## Algorithm allowlist & downgrade protection + +The server maintains an explicit allowlist of accepted COSE algorithms +(`ALLOWED_ALGORITHMS` in `backend/src/webauthn.rs`), currently `ES256`, +`RS256`, and `EdDSA`. A registration whose credential uses an algorithm +outside this list is rejected with `400 Bad Request`. Removing an algorithm +from the allowlist (e.g. if it is later found to be weak or deprecated) is +enough to reject new registrations using it, without touching the COSE +parsing or signature-verification code. + +Algorithms are also ranked by relative strength (`EdDSA` > `ES256` > `RS256`). +If a user already has a credential registered with a stronger algorithm, +attempting to register a new credential with a weaker one is rejected with +`409 Conflict` — this prevents a **downgrade attack**, where an attacker who +can influence registration (e.g. via a compromised client) tries to add a +weaker-algorithm credential to an account that has already moved to a +stronger algorithm. Registering an algorithm that is the same as or stronger +than any existing credential is always allowed.