diff --git a/Cargo.lock b/Cargo.lock index c48263d..c58342a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1773,6 +1773,7 @@ dependencies = [ "chacha20poly1305", "config", "dashmap", + "hex", "hmac", "http", "http-body-util", diff --git a/charts/miroir/templates/_helpers.tpl b/charts/miroir/templates/_helpers.tpl index 882b9c7..0f47a77 100644 --- a/charts/miroir/templates/_helpers.tpl +++ b/charts/miroir/templates/_helpers.tpl @@ -160,6 +160,27 @@ tracing: endpoint: {{ .Values.tracing.endpoint | default "http://tempo.monitoring.svc:4317" }} service_name: {{ .Values.tracing.serviceName | default "miroir" }} sample_rate: {{ .Values.tracing.sampleRate | default 0.1 }} +search_ui: + enabled: {{ .Values.search_ui.enabled | default true }} + scoped_key_max_age_days: {{ .Values.search_ui.scoped_key_max_age_days | default 60 }} + scoped_key_rotate_before_expiry_days: {{ .Values.search_ui.scoped_key_rotate_before_expiry_days | default 30 }} + scoped_key_rotation_drain_s: {{ .Values.search_ui.scoped_key_rotation_drain_s | default 120 }} +admin_ui: + enabled: {{ .Values.admin_ui.enabled | default true }} + path: {{ .Values.admin_ui.path | default "/_miroir/admin" }} + auth: {{ .Values.admin_ui.auth | default "key" }} + session_ttl_s: {{ .Values.admin_ui.session_ttl_s | default 3600 }} + read_only_mode: {{ .Values.admin_ui.read_only_mode | default false }} + allowed_origins: {{ .Values.admin_ui.allowed_origins | default list "same-origin" | toJson }} + cors_allowed_origins: {{ .Values.admin_ui.cors_allowed_origins | default list | toJson }} + rate_limit: + per_ip: {{ .Values.admin_ui.rate_limit.per_ip | default "10/minute" }} + backend: {{ .Values.admin_ui.rate_limit.backend | default "redis" }} + redis_key_prefix: {{ .Values.admin_ui.rate_limit.redis_key_prefix | default "miroir:ratelimit:adminlogin:" }} + redis_ttl_s: {{ .Values.admin_ui.rate_limit.redis_ttl_s | default 60 }} + failed_attempt_threshold: {{ .Values.admin_ui.rate_limit.failed_attempt_threshold | default 5 }} + backoff_start_minutes: {{ .Values.admin_ui.rate_limit.backoff_start_minutes | default 10 }} + backoff_max_hours: {{ .Values.admin_ui.rate_limit.backoff_max_hours | default 24 }} {{- if .Values.miroir.cdc }} cdc: enabled: {{ .Values.miroir.cdc.enabled | default true }} diff --git a/charts/miroir/values.schema.json b/charts/miroir/values.schema.json index eec4424..572f9ee 100644 --- a/charts/miroir/values.schema.json +++ b/charts/miroir/values.schema.json @@ -228,6 +228,7 @@ "enabled": { "type": "boolean" }, "scoped_key_max_age_days": { "type": "integer", "minimum": 2 }, "scoped_key_rotate_before_expiry_days": { "type": "integer", "minimum": 1 }, + "scoped_key_rotation_drain_s": { "type": "integer", "minimum": 10 }, "rate_limit": { "type": "object", "properties": { @@ -240,10 +241,14 @@ "type": "object", "properties": { "enabled": { "type": "boolean" }, - "login_rate_limit": { + "rate_limit": { "type": "object", "properties": { - "backend": { "type": "string", "enum": ["local", "redis"] } + "per_ip": { "type": "string" }, + "backend": { "type": "string", "enum": ["local", "redis"] }, + "failed_attempt_threshold": { "type": "integer", "minimum": 1 }, + "backoff_start_minutes": { "type": "integer", "minimum": 1 }, + "backoff_max_hours": { "type": "integer", "minimum": 1 } } } } @@ -352,7 +357,7 @@ } }, { - "description": "Rule 4: admin_ui.login_rate_limit.backend must be redis when miroir.replicas > 1 (local is per-pod, not shared)", + "description": "Rule 4: admin_ui.rate_limit.backend must be redis when miroir.replicas > 1 (local is per-pod, not shared)", "if": { "properties": { "miroir": { @@ -368,7 +373,7 @@ "properties": { "admin_ui": { "properties": { - "login_rate_limit": { + "rate_limit": { "properties": { "backend": { "enum": ["redis"] diff --git a/charts/miroir/values.yaml b/charts/miroir/values.yaml index a66f1ca..0b001c6 100644 --- a/charts/miroir/values.yaml +++ b/charts/miroir/values.yaml @@ -39,6 +39,30 @@ miroir: pvc_storage_class: "" memory_bytes: 67108864 +search_ui: + enabled: true + scoped_key_max_age_days: 60 # Maximum lifetime of a scoped Meilisearch key (days) + scoped_key_rotate_before_expiry_days: 30 # Rotate this many days before max_age (must be < max_age) + scoped_key_rotation_drain_s: 120 # Seconds to wait for all pods to observe new key before revoking old + +admin_ui: + enabled: true + path: "/_miroir/admin" + auth: key # key | oauth (future) | none (dev only) + session_ttl_s: 3600 # 1 hour + read_only_mode: false + allowed_origins: + - "same-origin" + cors_allowed_origins: [] + rate_limit: + per_ip: "10/minute" # Rate limit per source IP + backend: redis # redis | local (schema enforces redis when replicas > 1) + redis_key_prefix: "miroir:ratelimit:adminlogin:" + redis_ttl_s: 60 + failed_attempt_threshold: 5 # Consecutive failures before exponential backoff + backoff_start_minutes: 10 # Initial backoff after threshold + backoff_max_hours: 24 # Maximum backoff cap + taskStore: backend: sqlite # sqlite | redis path: /data/miroir-tasks.db diff --git a/crates/miroir-core/src/api_error.rs b/crates/miroir-core/src/api_error.rs index 594e968..a9d71bf 100644 --- a/crates/miroir-core/src/api_error.rs +++ b/crates/miroir-core/src/api_error.rs @@ -36,12 +36,14 @@ pub enum MiroirCode { JwtInvalid, JwtScopeDenied, InvalidAuth, + MissingCsrf, + CsrfMismatch, } impl MiroirCode { /// All variants, used for iteration in tests. #[cfg(test)] - const ALL: [MiroirCode; 10] = [ + const ALL: [MiroirCode; 12] = [ MiroirCode::PrimaryKeyRequired, MiroirCode::NoQuorum, MiroirCode::ShardUnavailable, @@ -52,6 +54,8 @@ impl MiroirCode { MiroirCode::JwtInvalid, MiroirCode::JwtScopeDenied, MiroirCode::InvalidAuth, + MiroirCode::MissingCsrf, + MiroirCode::CsrfMismatch, ]; /// Returns the error code string (e.g., `"miroir_no_quorum"`). @@ -67,6 +71,8 @@ impl MiroirCode { Self::JwtInvalid => "miroir_jwt_invalid", Self::JwtScopeDenied => "miroir_jwt_scope_denied", Self::InvalidAuth => "miroir_invalid_auth", + Self::MissingCsrf => "miroir_missing_csrf", + Self::CsrfMismatch => "miroir_csrf_mismatch", } } @@ -78,7 +84,7 @@ impl MiroirCode { | Self::IdempotencyKeyReused | Self::MultiAliasNotWritable => ErrorType::InvalidRequest, - Self::JwtInvalid | Self::JwtScopeDenied | Self::InvalidAuth => ErrorType::Auth, + Self::JwtInvalid | Self::JwtScopeDenied | Self::InvalidAuth | Self::MissingCsrf | Self::CsrfMismatch => ErrorType::Auth, Self::NoQuorum | Self::ShardUnavailable | Self::SettingsVersionStale => { ErrorType::System @@ -90,8 +96,8 @@ impl MiroirCode { pub fn http_status(&self) -> u16 { match self { Self::PrimaryKeyRequired | Self::ReservedField => 400, - Self::JwtInvalid | Self::InvalidAuth => 401, - Self::JwtScopeDenied => 403, + Self::JwtInvalid | Self::InvalidAuth | Self::MissingCsrf => 401, + Self::JwtScopeDenied | Self::CsrfMismatch => 403, Self::IdempotencyKeyReused | Self::MultiAliasNotWritable => 409, Self::NoQuorum | Self::ShardUnavailable | Self::SettingsVersionStale => 503, } @@ -118,6 +124,8 @@ impl MiroirCode { "miroir_jwt_invalid" => Some(Self::JwtInvalid), "miroir_jwt_scope_denied" => Some(Self::JwtScopeDenied), "miroir_invalid_auth" => Some(Self::InvalidAuth), + "miroir_missing_csrf" => Some(Self::MissingCsrf), + "miroir_csrf_mismatch" => Some(Self::CsrfMismatch), _ => None, } } @@ -354,6 +362,34 @@ mod tests { assert_eq!(err.http_status(), 401); } + #[test] + fn miroir_missing_csrf_shape() { + let err = MeilisearchError::new( + MiroirCode::MissingCsrf, + "CSRF token is required for state-changing requests.", + ); + let json: serde_json::Value = serde_json::to_value(&err).unwrap(); + + assert_eq!(json["code"], "miroir_missing_csrf"); + assert_eq!(json["type"], "auth"); + assert!(json["link"].as_str().unwrap().contains("miroir_missing_csrf")); + assert_eq!(err.http_status(), 401); + } + + #[test] + fn miroir_csrf_mismatch_shape() { + let err = MeilisearchError::new( + MiroirCode::CsrfMismatch, + "CSRF token does not match the session token.", + ); + let json: serde_json::Value = serde_json::to_value(&err).unwrap(); + + assert_eq!(json["code"], "miroir_csrf_mismatch"); + assert_eq!(json["type"], "auth"); + assert!(json["link"].as_str().unwrap().contains("miroir_csrf_mismatch")); + assert_eq!(err.http_status(), 403); + } + // -- Meilisearch-native error forwarding -------------------------------------- #[test] @@ -420,7 +456,9 @@ mod tests { (MiroirCode::ReservedField, 400), (MiroirCode::JwtInvalid, 401), (MiroirCode::InvalidAuth, 401), + (MiroirCode::MissingCsrf, 401), (MiroirCode::JwtScopeDenied, 403), + (MiroirCode::CsrfMismatch, 403), (MiroirCode::IdempotencyKeyReused, 409), (MiroirCode::MultiAliasNotWritable, 409), (MiroirCode::NoQuorum, 503), diff --git a/crates/miroir-core/src/config.rs b/crates/miroir-core/src/config.rs index 8eefb74..4846f22 100644 --- a/crates/miroir-core/src/config.rs +++ b/crates/miroir-core/src/config.rs @@ -6,6 +6,7 @@ mod load; mod validate; pub use error::ConfigError; +pub use advanced::{SearchUiConfig, CspOverridesConfig}; use serde::{Deserialize, Serialize}; diff --git a/crates/miroir-core/src/config/advanced.rs b/crates/miroir-core/src/config/advanced.rs index 87e3437..f3be0dd 100644 --- a/crates/miroir-core/src/config/advanced.rs +++ b/crates/miroir-core/src/config/advanced.rs @@ -589,6 +589,33 @@ impl Default for CanaryRunnerConfig { // 13.19 Admin Web UI // --------------------------------------------------------------------------- +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(default)] +pub struct AdminUiRateLimitConfig { + pub per_ip: String, + /// `redis` or `local`. + pub backend: String, + pub redis_key_prefix: String, + pub redis_ttl_s: u64, + pub failed_attempt_threshold: u32, + pub backoff_start_minutes: u64, + pub backoff_max_hours: u64, +} + +impl Default for AdminUiRateLimitConfig { + fn default() -> Self { + Self { + per_ip: "10/minute".into(), + backend: "redis".into(), + redis_key_prefix: "miroir:ratelimit:adminlogin:".into(), + redis_ttl_s: 60, + failed_attempt_threshold: 5, + backoff_start_minutes: 10, + backoff_max_hours: 24, + } + } +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(default)] pub struct AdminUiConfig { @@ -600,9 +627,11 @@ pub struct AdminUiConfig { pub read_only_mode: bool, pub allowed_origins: Vec, pub cors_allowed_origins: Vec, + pub csp: String, pub csp_overrides: CspOverridesConfig, pub theme: AdminUiThemeConfig, pub features: AdminUiFeaturesConfig, + pub rate_limit: AdminUiRateLimitConfig, } impl Default for AdminUiConfig { @@ -615,9 +644,11 @@ impl Default for AdminUiConfig { read_only_mode: false, allowed_origins: vec!["same-origin".into()], cors_allowed_origins: Vec::new(), + csp: "default-src 'self'; script-src 'self'; img-src 'self' data:; style-src 'self' 'unsafe-inline'; connect-src 'self'; frame-ancestors 'none'".into(), csp_overrides: CspOverridesConfig::default(), theme: AdminUiThemeConfig::default(), features: AdminUiFeaturesConfig::default(), + rate_limit: AdminUiRateLimitConfig::default(), } } } diff --git a/crates/miroir-core/src/config/validate.rs b/crates/miroir-core/src/config/validate.rs index 6274992..013300a 100644 --- a/crates/miroir-core/src/config/validate.rs +++ b/crates/miroir-core/src/config/validate.rs @@ -141,6 +141,52 @@ pub fn validate(cfg: &MiroirConfig) -> Result<(), ConfigError> { )); } + // CSP overrides must not contain wildcards (plan §9) + // Wildcards are only allowed in the base template, not in overrides. + // Overrides are additive and must be specific sources. + for value in &cfg.admin_ui.csp_overrides.script_src { + if value == "*" { + return Err(ConfigError::Validation( + "admin_ui.csp_overrides.script_src cannot contain wildcard '*'".into(), + )); + } + } + for value in &cfg.admin_ui.csp_overrides.img_src { + if value == "*" { + return Err(ConfigError::Validation( + "admin_ui.csp_overrides.img_src cannot contain wildcard '*'".into(), + )); + } + } + for value in &cfg.admin_ui.csp_overrides.connect_src { + if value == "*" { + return Err(ConfigError::Validation( + "admin_ui.csp_overrides.connect_src cannot contain wildcard '*'".into(), + )); + } + } + for value in &cfg.search_ui.csp_overrides.script_src { + if value == "*" { + return Err(ConfigError::Validation( + "search_ui.csp_overrides.script_src cannot contain wildcard '*'".into(), + )); + } + } + for value in &cfg.search_ui.csp_overrides.img_src { + if value == "*" { + return Err(ConfigError::Validation( + "search_ui.csp_overrides.img_src cannot contain wildcard '*'".into(), + )); + } + } + for value in &cfg.search_ui.csp_overrides.connect_src { + if value == "*" { + return Err(ConfigError::Validation( + "search_ui.csp_overrides.connect_src cannot contain wildcard '*'".into(), + )); + } + } + Ok(()) } @@ -268,4 +314,22 @@ mod tests { let err = validate(&cfg).unwrap_err(); assert!(err.to_string().contains("replication_factor")); } + + #[test] + fn rejects_csp_overrides_wildcard_admin_ui() { + let mut cfg = dev_config(); + cfg.admin_ui.csp_overrides.script_src = vec!["*".to_string()]; + let err = validate(&cfg).unwrap_err(); + assert!(err.to_string().contains("csp_overrides")); + assert!(err.to_string().contains("wildcard")); + } + + #[test] + fn rejects_csp_overrides_wildcard_search_ui() { + let mut cfg = dev_config(); + cfg.search_ui.csp_overrides.connect_src = vec!["*".to_string()]; + let err = validate(&cfg).unwrap_err(); + assert!(err.to_string().contains("csp_overrides")); + assert!(err.to_string().contains("wildcard")); + } } diff --git a/crates/miroir-core/src/task_store/redis.rs b/crates/miroir-core/src/task_store/redis.rs index 0561adf..ff8fcd3 100644 --- a/crates/miroir-core/src/task_store/redis.rs +++ b/crates/miroir-core/src/task_store/redis.rs @@ -1886,20 +1886,7 @@ impl RedisTaskStore { // Check if limit exceeded if count >= limit { - // Enter backoff mode after 5 consecutive failures - let failed_count: Option = conn.hget(&backoff_key, "failed_count").await - .map_err(|e| MiroirError::Redis(e.to_string()))?; - - let failed = failed_count.unwrap_or(0) + 1; - let wait_seconds = 600u64 * (1u64 << (failed.saturating_sub(5).min(10))); // 10m, 20m, 40m, ... - - let mut pipe = pipe(); - pipe.hset(&backoff_key, "failed_count", failed as i64); - pipe.hset(&backoff_key, "next_allowed_at", now_ms() + (wait_seconds as i64 * 1000)); - pipe.expire(&backoff_key, (wait_seconds as i64 + 60) as i64); - pool.pipeline_query::<()>(&mut pipe).await?; - - return Ok((false, Some(wait_seconds))); + return Ok((false, None)); } // Increment counter @@ -1912,6 +1899,61 @@ impl RedisTaskStore { }) } + /// Record a failed admin login attempt and return backoff if triggered. + /// Returns Some(wait_seconds) if backoff was triggered, None otherwise. + pub fn record_failure_admin_login( + &self, + ip: &str, + failed_threshold: u32, + backoff_start_minutes: u64, + backoff_max_hours: u64, + ) -> Result> { + let pool = self.pool.clone(); + let key_prefix = self.key_prefix.clone(); + let ip = ip.to_string(); + let backoff_key = format!("{}:ratelimit:adminlogin:backoff:{}", key_prefix, ip); + + self.block_on(async move { + let mut conn = pool.manager.lock().await; + + // Check if already in backoff + let backoff_fields: HashMap = conn.hgetall(&backoff_key).await + .map_err(|e| MiroirError::Redis(e.to_string()))?; + + let current_failed: u64 = if backoff_fields.is_empty() { + 0 + } else { + get_field_i64(&backoff_fields, "failed_count")? as u64 + }; + + let new_failed = current_failed + 1; + + // Check if we should enter backoff mode + if new_failed >= failed_threshold as u64 { + let backoff_exponent = (new_failed.saturating_sub(failed_threshold as u64) as u32).min(7); + let backoff_minutes = backoff_start_minutes * (1u64 << backoff_exponent); + let backoff_seconds = (backoff_minutes * 60).min(backoff_max_hours * 3600); + + let now = now_ms(); + let next_allowed_at = now + (backoff_seconds as i64 * 1000); + + let mut pipe = pipe(); + pipe.hset(&backoff_key, "failed_count", new_failed as i64); + pipe.hset(&backoff_key, "next_allowed_at", next_allowed_at); + pipe.expire(&backoff_key, (backoff_seconds as i64 + 60) as i64); + pool.pipeline_query::<()>(&mut pipe).await?; + + return Ok(Some(backoff_seconds)); + } + + // Just update the failed count + let _: () = conn.hset(&backoff_key, "failed_count", new_failed as i64).await + .map_err(|e| MiroirError::Redis(e.to_string()))?; + + Ok(None) + }) + } + /// Reset admin login rate limit on successful login. pub fn reset_rate_limit_admin_login(&self, ip: &str) -> Result<()> { let pool = self.pool.clone(); @@ -1948,7 +1990,9 @@ impl RedisTaskStore { } else { Ok(Some(SearchUiScopedKey { index_uid: index_uid.clone(), + primary_key: get_field_string(&fields, "primary_key")?, primary_uid: get_field_string(&fields, "primary_uid")?, + previous_key: opt_field(&fields, "previous_key"), previous_uid: opt_field(&fields, "previous_uid"), rotated_at: get_field_i64(&fields, "rotated_at")?, generation: get_field_i64(&fields, "generation")?, @@ -1967,11 +2011,17 @@ impl RedisTaskStore { self.block_on(async move { let mut pipe = pipe(); pipe.hset(&redis_key, "index_uid", &key_value.index_uid); + pipe.hset(&redis_key, "primary_key", &key_value.primary_key); pipe.hset(&redis_key, "primary_uid", &key_value.primary_uid); pipe.hset(&redis_key, "rotated_at", key_value.rotated_at); pipe.hset(&redis_key, "generation", key_value.generation); - if let Some(ref prev) = key_value.previous_uid { - pipe.hset(&redis_key, "previous_uid", prev); + match key_value.previous_key { + Some(ref v) => { pipe.hset(&redis_key, "previous_key", v); } + None => { pipe.hdel(&redis_key, "previous_key"); } + } + match key_value.previous_uid { + Some(ref v) => { pipe.hset(&redis_key, "previous_uid", v); } + None => { pipe.hdel(&redis_key, "previous_uid"); } } pool.pipeline_query::<()>(&mut pipe).await?; Ok(()) @@ -2037,6 +2087,97 @@ impl RedisTaskStore { }) } + /// Clear the previous_uid field from a scoped key hash (after revocation). + pub fn clear_scoped_key_previous(&self, index_uid: &str) -> Result<()> { + let pool = self.pool.clone(); + let key_prefix = self.key_prefix.clone(); + let index_uid = index_uid.to_string(); + let redis_key = format!("{}:search_ui_scoped_key:{}", key_prefix, index_uid); + + self.block_on(async move { + let mut pipe = pipe(); + pipe.hdel(&redis_key, "previous_uid"); + pipe.hdel(&redis_key, "previous_key"); + pool.pipeline_query::<()>(&mut pipe).await?; + Ok(()) + }) + } + + /// Register this pod as alive. Uses a Sorted Set with timestamp scores + /// so we can query for recently-active pods. + pub fn register_pod_presence(&self, pod_id: &str) -> Result<()> { + let pool = self.pool.clone(); + let key_prefix = self.key_prefix.clone(); + let pod_id = pod_id.to_string(); + let key = format!("{}:live_pods", key_prefix); + let now = now_ms(); + + self.block_on(async move { + let mut pipe = pipe(); + pipe.zadd(&key, &pod_id, now); + // Expire the whole set after 5 minutes to prevent unbounded growth. + // Active pods continuously refresh, so this just cleans up after total shutdown. + pipe.expire(&key, 300); + pool.pipeline_query::<()>(&mut pipe).await?; + Ok(()) + }) + } + + /// Get the list of pods that have registered presence within the last 120 seconds. + pub fn get_live_pods(&self) -> Result> { + let manager = self.pool.manager.clone(); + let key_prefix = self.key_prefix.clone(); + let key = format!("{}:live_pods", key_prefix); + let cutoff = now_ms() - 120_000; // 120 seconds ago + + self.block_on(async move { + let mut conn = manager.lock().await; + let pods: Vec = conn.zrangebyscore(&key, cutoff, "+inf") + .await + .map_err(|e| MiroirError::Redis(e.to_string()))?; + Ok(pods) + }) + } + + /// List all index UIDs that have scoped keys in Redis. + pub fn list_scoped_key_indexes(&self) -> Result> { + let pool = self.pool.clone(); + let key_prefix = self.key_prefix.clone(); + + self.block_on(async move { + let pattern = format!("{}:search_ui_scoped_key:*", key_prefix); + let mut conn = pool.manager.lock().await; + + let mut indexes = Vec::new(); + let mut cursor: u64 = 0; + loop { + let (new_cursor, keys): (u64, Vec) = ::redis::cmd("SCAN") + .arg(cursor) + .arg("MATCH") + .arg(&pattern) + .arg("COUNT") + .arg(100) + .query_async(&mut *conn) + .await + .map_err(|e| MiroirError::Redis(e.to_string()))?; + + for key in keys { + // Extract index_uid from the key: "miroir:search_ui_scoped_key:" + if let Some(idx) = key.rsplit(':').next() { + indexes.push(idx.to_string()); + } + } + + cursor = new_cursor; + if cursor == 0 { + break; + } + } + + Ok(indexes) + }) + } + // --- CDC overflow buffer --- /// Append to the CDC overflow buffer for a sink. @@ -2112,11 +2253,17 @@ impl RedisTaskStore { // --- Extra types for Redis-specific functionality --- -/// Scoped key for search UI access (plan §4 footnote). +/// Scoped key for search UI access (plan §13.21). #[derive(Debug, Clone)] pub struct SearchUiScopedKey { pub index_uid: String, + /// The Meilisearch API key used as Bearer token for search requests. + pub primary_key: String, + /// The Meilisearch key UID for management (DELETE /keys/{uid}). pub primary_uid: String, + /// The previous API key (fallback during rotation overlap window). + pub previous_key: Option, + /// The previous key UID (for revocation). pub previous_uid: Option, pub rotated_at: i64, pub generation: i64, @@ -3236,7 +3383,9 @@ mod tests { // Verify SearchUiScopedKey can be constructed and has expected fields let key = SearchUiScopedKey { index_uid: "test-index".to_string(), + primary_key: "pk-abc".to_string(), primary_uid: "primary-123".to_string(), + previous_key: Some("ppk-def".to_string()), previous_uid: Some("previous-456".to_string()), rotated_at: 1234567890, generation: 5, diff --git a/crates/miroir-proxy/Cargo.toml b/crates/miroir-proxy/Cargo.toml index 17a296d..4a84589 100644 --- a/crates/miroir-proxy/Cargo.toml +++ b/crates/miroir-proxy/Cargo.toml @@ -34,7 +34,8 @@ base64 = "0.22" chacha20poly1305 = "0.10" rand = "0.8" dashmap = "6" -miroir-core = { path = "../miroir-core", features = ["axum"] } +hex = "0.4" +miroir-core = { path = "../miroir-core", features = ["axum", "redis-store"] } # OpenTelemetry (optional - use feature flag to enable) opentelemetry = { version = "0.27", optional = true } diff --git a/crates/miroir-proxy/src/auth.rs b/crates/miroir-proxy/src/auth.rs index 9721999..7d56ab1 100644 --- a/crates/miroir-proxy/src/auth.rs +++ b/crates/miroir-proxy/src/auth.rs @@ -10,7 +10,7 @@ //! the rotation overlap window. Validation accepts either secret. use axum::{ - extract::{Request, State}, + extract::{FromRef, Request, State}, http::{HeaderMap, Method}, middleware::Next, response::{IntoResponse, Response}, @@ -18,8 +18,9 @@ use axum::{ use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; use dashmap::DashMap; use hmac::{Hmac, Mac}; -use miroir_core::{MeilisearchError, MiroirCode}; +use miroir_core::{task_store::TaskStore, MeilisearchError, MiroirCode}; use prometheus::Counter; +use rand::RngCore; use serde::{Deserialize, Serialize}; use sha2::Sha256; use std::sync::Arc; @@ -34,6 +35,13 @@ type HmacSha256 = Hmac; #[derive(Debug, Clone)] pub struct AdminSessionId(pub String); +/// State for CSRF middleware, combining AuthState with task store access. +#[derive(Clone)] +pub struct CsrfState { + pub auth: AuthState, + pub redis_store: Option, +} + // --------------------------------------------------------------------------- // JWT claims (plan §13.21) // --------------------------------------------------------------------------- @@ -240,6 +248,177 @@ pub enum JwtValidationError { PreviousSecretEmpty, } +// --------------------------------------------------------------------------- +// CSRF token generation (plan §9) +// --------------------------------------------------------------------------- + +/// Generate a cryptographically random CSRF token. +/// Returns a URL-safe base64-encoded 32-byte token. +pub fn generate_csrf_token() -> String { + let mut bytes = [0u8; 32]; + rand::rngs::OsRng.fill_bytes(&mut bytes); + URL_SAFE_NO_PAD.encode(&bytes) +} + +/// Extract the CSRF token from the `X-CSRF-Token` header. +pub fn extract_csrf_token(headers: &HeaderMap) -> Option { + headers.get("X-CSRF-Token")?.to_str().ok().map(String::from) +} + +/// Constant-time comparison of CSRF tokens. +pub fn constant_time_csrf_compare(token: &str, expected: &str) -> bool { + constant_time_compare(token.as_bytes(), expected.as_bytes()) +} + +/// Validate a CSRF token against the expected session token. +/// Returns Ok(()) if the token matches, or a CsrfMismatch error. +pub fn validate_csrf_token(provided: &str, expected: &str) -> Result<(), MiroirCode> { + if constant_time_csrf_compare(provided, expected) { + Ok(()) + } else { + Err(MiroirCode::CsrfMismatch) + } +} + +// --------------------------------------------------------------------------- +// Origin validation (plan §9) +// --------------------------------------------------------------------------- + +/// Result of origin validation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum OriginVerdict { + /// Origin is allowed. + Allowed, + /// Origin is not in the allowed list. + Forbidden, + /// No Origin/Referer header present (for same-origin requests). + Missing, +} + +/// Validate the Origin header against the allowed origins list. +/// Handles special "same-origin" value by comparing against the Host header. +pub fn validate_origin( + headers: &HeaderMap, + allowed_origins: &[String], + is_same_origin_by_default: bool, +) -> OriginVerdict { + // Try Origin header first (preferred for POST/DELETE/PUT) + let origin = headers.get("origin").and_then(|h| h.to_str().ok()); + + // Fall back to Referer header (for navigational requests) + let referer = origin.or_else(|| headers.get("referer").and_then(|h| h.to_str().ok())); + + let provided_origin = match referer { + Some(o) => o, + None => { + // No Origin or Referer header - for same-origin requests, this is acceptable + return if is_same_origin_by_default { + OriginVerdict::Missing + } else { + OriginVerdict::Forbidden + }; + } + }; + + // Strip path from Referer to get origin + let provided_origin = if let Some(ref_hdr) = headers.get("referer") { + if let Ok(ref_val) = ref_hdr.to_str() { + // Find the first '/' after "https://" (skip the first 8 chars: "https://") + if let Some(idx) = ref_val.chars().enumerate().skip(8).find(|(_, c)| *c == '/').map(|(i, _)| i) { + &ref_val[..idx] + } else { + ref_val + } + } else { + provided_origin + } + } else { + provided_origin + }; + + // Check against allowed origins + for allowed in allowed_origins { + // Special "same-origin" value - compare against Host header + if allowed == "same-origin" { + if let Some(host) = headers.get("host").and_then(|h| h.to_str().ok()) { + // Construct origin from scheme (https) and host + let same_origin = format!("https://{}", host); + if provided_origin == same_origin || provided_origin == host { + return OriginVerdict::Allowed; + } + } + } else if allowed == "*" { + // Wildcard allows any origin + return OriginVerdict::Allowed; + } else if provided_origin == allowed { + return OriginVerdict::Allowed; + } + } + + OriginVerdict::Forbidden +} + +// --------------------------------------------------------------------------- +// CSP header builder (plan §9) +// --------------------------------------------------------------------------- + +/// Build a CSP header value by merging base template with overrides. +/// Overrides are merged additively - they never replace the base template. +pub fn build_csp_header( + base_template: &str, + overrides: &miroir_core::config::CspOverridesConfig, +) -> String { + let mut directives: Vec<(String, Vec)> = base_template + .split(';') + .map(|s| s.trim()) + .filter(|s| !s.is_empty()) + .map(|directive| { + let parts: Vec<&str> = directive.splitn(2, ' ').collect(); + if parts.len() == 2 { + (parts[0].to_lowercase(), vec![parts[1].to_string()]) + } else { + (parts[0].to_lowercase(), vec![]) + } + }) + .collect(); + + // Helper to merge overrides into a directive + let merge_into = |directives: &mut Vec<(String, Vec)>, + name: &str, + values: &[String]| { + if values.is_empty() { + return; + } + let name_lower = name.to_lowercase(); + if let Some(entry) = directives.iter_mut().find(|(n, _)| n == &name_lower) { + // Append to existing directive + entry.1.extend(values.iter().cloned()); + entry.1.dedup(); // Remove duplicates + } else { + // Add new directive + directives.push((name_lower, values.to_vec())); + } + }; + + // Merge each override category + merge_into(&mut directives, "script-src", &overrides.script_src); + merge_into(&mut directives, "img-src", &overrides.img_src); + merge_into(&mut directives, "connect-src", &overrides.connect_src); + + // Rebuild CSP string + directives + .into_iter() + .map(|(name, values)| { + if values.is_empty() { + name + } else { + format!("{} {}", name, values.join(" ")) + } + }) + .collect::>() + .join("; ") +} + // --------------------------------------------------------------------------- // Dispatch verdict // --------------------------------------------------------------------------- @@ -543,6 +722,145 @@ pub async fn auth_middleware( } } +// --------------------------------------------------------------------------- +// CSRF validation middleware (plan §9) +// --------------------------------------------------------------------------- + +/// CSRF middleware that validates `X-CSRF-Token` on state-changing requests. +/// +/// Bypasses CSRF check when: +/// - Request is authenticated via Bearer token (not admin session cookie) +/// - X-Admin-Key header is present +/// - Request method is safe (GET, HEAD, OPTIONS) +/// - Path is dispatch-exempt +/// +/// For admin session cookie auth, requires `X-CSRF-Token` header to match +/// the token stored in the session. +pub async fn csrf_middleware( + State(state): State, + req: Request, + next: Next, +) -> Response { + let method = req.method().clone(); + let path = req.uri().path().to_string(); + + // Skip CSRF for safe methods + if matches!(method, Method::GET | Method::HEAD | Method::OPTIONS) { + return next.run(req).await; + } + + // Skip CSRF for non-admin paths + if !is_admin_path(&path) { + return next.run(req).await; + } + + // Skip CSRF for dispatch-exempt endpoints + if is_dispatch_exempt(&method, &path) { + return next.run(req).await; + } + + // Skip CSRF if X-Admin-Key is present (bypasses CSRF) + if check_x_admin_key(req.headers(), state.auth.admin_key.as_bytes()) { + return next.run(req).await; + } + + // Check if authenticated via admin session cookie + let has_session_cookie = extract_admin_session_cookie(req.headers()).is_some(); + let has_bearer_token = extract_bearer(req.headers()).is_some(); + + // CSRF only applies to session-cookie auth, not bearer tokens + if !has_session_cookie || has_bearer_token { + return next.run(req).await; + } + + // Extract CSRF token from header + let csrf_token = match extract_csrf_token(req.headers()) { + Some(token) => token, + None => { + return MeilisearchError::new( + MiroirCode::MissingCsrf, + "CSRF token is required for state-changing requests.", + ) + .into_response(); + } + }; + + // Get session ID from extensions (set by auth_middleware) + let session_id = match req.extensions().get::() { + Some(id) => id.0.clone(), + None => { + // Session cookie was present but auth_middleware didn't set AdminSessionId + // This means the session was invalid/expired/revoked + return MeilisearchError::new( + MiroirCode::InvalidAuth, + "Admin session is invalid or expired.", + ) + .into_response(); + } + }; + + // Validate CSRF token against session + let Some(redis_store) = state.redis_store.as_ref() else { + return MeilisearchError::new( + MiroirCode::InvalidAuth, + "Admin sessions require Redis task store.", + ) + .into_response(); + }; + + let session = match redis_store.get_admin_session(&session_id) { + Ok(Some(s)) => s, + Ok(None) => { + return MeilisearchError::new( + MiroirCode::InvalidAuth, + "Admin session not found.", + ) + .into_response(); + } + Err(e) => { + tracing::warn!(error = %e, session_prefix = &session_id[..session_id.len().min(8)], "failed to get admin session for CSRF validation"); + return MeilisearchError::new( + MiroirCode::InvalidAuth, + "Failed to validate session.", + ) + .into_response(); + } + }; + + // Check if revoked + if session.revoked { + return MeilisearchError::new( + MiroirCode::InvalidAuth, + "Admin session has been revoked.", + ) + .into_response(); + } + + // Check expiration + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + if session.expires_at < now { + return MeilisearchError::new( + MiroirCode::InvalidAuth, + "Admin session has expired.", + ) + .into_response(); + } + + // Constant-time compare CSRF tokens + if !constant_time_csrf_compare(&csrf_token, &session.csrf_token) { + return MeilisearchError::new( + MiroirCode::CsrfMismatch, + "CSRF token does not match the session token.", + ) + .into_response(); + } + + next.run(req).await +} + // --------------------------------------------------------------------------- // Rate-limit hook types (Phase 2 in-memory stub, Phase 6 multi-pod) // --------------------------------------------------------------------------- diff --git a/crates/miroir-proxy/src/lib.rs b/crates/miroir-proxy/src/lib.rs index c0c3c49..0857918 100644 --- a/crates/miroir-proxy/src/lib.rs +++ b/crates/miroir-proxy/src/lib.rs @@ -1,3 +1,6 @@ +pub mod admin_session; +pub mod auth; pub mod client; pub mod middleware; pub mod routes; +pub mod scoped_key_rotation; diff --git a/crates/miroir-proxy/src/main.rs b/crates/miroir-proxy/src/main.rs index 8011437..d3035bd 100644 --- a/crates/miroir-proxy/src/main.rs +++ b/crates/miroir-proxy/src/main.rs @@ -88,7 +88,7 @@ impl UnifiedState { admin_key: admin_key.clone(), jwt_primary, jwt_previous, - seal_key, + seal_key: seal_key.clone(), revoked_sessions: std::sync::Arc::new(dashmap::DashMap::new()), admin_session_revoked_total: metrics.admin_session_revoked_total(), }; @@ -98,6 +98,7 @@ impl UnifiedState { metrics.clone(), redis_store.clone(), pod_id.clone(), + seal_key.clone(), ); Self { auth, metrics, admin, pod_id, redis_store } @@ -116,6 +117,8 @@ impl FromRef for admin_endpoints::AppState { task_registry: state.admin.task_registry.clone(), redis_store: state.redis_store.clone(), pod_id: state.pod_id.clone(), + seal_key: state.auth.seal_key.clone(), + local_rate_limiter: admin_endpoints::LocalAdminRateLimiter::new(), } } } @@ -130,6 +133,16 @@ impl FromRef for TelemetryState { } } +// Implement FromRef so that CsrfState can be extracted from UnifiedState +impl FromRef for auth::CsrfState { + fn from_ref(state: &UnifiedState) -> Self { + auth::CsrfState { + auth: state.auth.clone(), + redis_store: state.redis_store.clone(), + } + } +} + #[tokio::main] async fn main() -> anyhow::Result<()> { // Load configuration (file → env → CLI overlay) @@ -240,6 +253,13 @@ async fn main() -> anyhow::Result<()> { state.auth.clone(), auth::auth_middleware, )) + .layer(axum::middleware::from_fn_with_state( + auth::CsrfState { + auth: state.auth.clone(), + redis_store: state.redis_store.clone(), + }, + auth::csrf_middleware, + )) .layer(axum::middleware::from_fn_with_state( TelemetryState { metrics: state.metrics.clone(), diff --git a/crates/miroir-proxy/src/routes/admin.rs b/crates/miroir-proxy/src/routes/admin.rs index 82925e5..8f754d6 100644 --- a/crates/miroir-proxy/src/routes/admin.rs +++ b/crates/miroir-proxy/src/routes/admin.rs @@ -4,10 +4,10 @@ use axum::{ extract::FromRef, - routing::get, + routing::{get, post}, Router, }; -use super::admin_endpoints; +use super::{admin_endpoints, session}; /// Create the admin router with all /_miroir/* endpoints. /// @@ -20,8 +20,22 @@ where admin_endpoints::AppState: FromRef, { Router::new() + // Admin session endpoints (plan §9, §13.19) + .route("/admin/login", post(session::admin_login::)) + .route("/admin/session", get(session::admin_session::)) + .route("/admin/logout", post(session::admin_logout::)) + // Search UI session endpoint (plan §9, §13.21) + .route( + "/ui/search/{index}/session", + get(session::search_ui_session::), + ) + // Admin API endpoints .route("/topology", get(admin_endpoints::get_topology::)) .route("/shards", get(admin_endpoints::get_shards::)) .route("/ready", get(admin_endpoints::get_ready::)) .route("/metrics", get(admin_endpoints::get_metrics::)) + .route( + "/ui/search/{index}/rotate-scoped-key", + post(admin_endpoints::rotate_scoped_key_handler), + ) } diff --git a/crates/miroir-proxy/src/routes/admin_endpoints.rs b/crates/miroir-proxy/src/routes/admin_endpoints.rs index c2f445a..f888675 100644 --- a/crates/miroir-proxy/src/routes/admin_endpoints.rs +++ b/crates/miroir-proxy/src/routes/admin_endpoints.rs @@ -1,8 +1,8 @@ //! Admin API endpoints for topology, readiness, shards, and metrics. use axum::{ - extract::{FromRef, State}, - http::StatusCode, + extract::{FromRef, Path, State}, + http::{HeaderMap, StatusCode}, Json, response::{IntoResponse, Response}, }; @@ -10,16 +10,44 @@ use miroir_core::{ config::MiroirConfig, router, task_registry::InMemoryTaskRegistry, + task_store::RedisTaskStore, topology::{Node, NodeId, Topology}, }; +use rand::RngCore; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; use tokio::sync::RwLock; -use tracing::info; +use tracing::{info, error, warn}; use reqwest::Client; +use crate::{ + admin_session::{seal_session, COOKIE_NAME, SealKey}, + scoped_key_rotation::{self, ScopedKeyRotationState, RotateScopedKeyRequest, RotateScopedKeyResponse}, +}; + +/// Hash a PII value (IP address) for safe log correlation. +fn hash_for_log(value: &str) -> String { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + value.hash(&mut hasher); + format!("{:016x}", hasher.finish()) +} + +/// Request body for POST /_miroir/admin/login. +#[derive(Debug, Deserialize)] +pub struct AdminLoginRequest { + pub admin_key: String, +} + +/// Response body for POST /_miroir/admin/login. +#[derive(Debug, Serialize)] +pub struct AdminLoginResponse { + pub success: bool, + pub message: Option, +} + /// Version state with cache for fetching Meilisearch version. #[derive(Clone)] pub struct VersionState { @@ -84,6 +112,128 @@ impl VersionState { } } +// --------------------------------------------------------------------------- +// Local Rate Limiter (for single-pod deployments) +// --------------------------------------------------------------------------- + +/// In-memory rate limiter for admin login (local backend only). +/// Thread-safe using Arc>. +#[derive(Debug, Clone)] +pub struct LocalAdminRateLimiter { + inner: Arc>, +} + +#[derive(Debug, Default)] +struct LocalAdminRateLimiterInner { + /// Map of IP -> (request_timestamps_ms, failed_count, backoff_until_ms) + state: HashMap, +} + +#[derive(Debug, Default, Clone)] +struct LocalRateLimitState { + /// Timestamps of recent requests (for sliding window) + request_timestamps_ms: Vec, + /// Consecutive failed login attempts + failed_count: u32, + /// Unix timestamp (ms) when backoff expires + backoff_until_ms: Option, +} + +impl LocalAdminRateLimiter { + pub fn new() -> Self { + Self { + inner: Arc::new(std::sync::Mutex::new(LocalAdminRateLimiterInner::default())), + } + } + + /// Check rate limit and exponential backoff. + /// Returns (allowed, wait_seconds). + pub fn check( + &self, + ip: &str, + limit: u64, + window_ms: u64, + failed_threshold: u32, + backoff_start_minutes: u64, + backoff_max_hours: u64, + ) -> (bool, Option) { + let mut inner = self.inner.lock().unwrap(); + let now = now_ms(); + let state = inner.state.entry(ip.to_string()).or_default(); + + // Check if we're in backoff mode + if let Some(backoff_until) = state.backoff_until_ms { + if backoff_until > now { + let wait_seconds = ((backoff_until - now) / 1000) as u64; + return (false, Some(wait_seconds)); + } + // Backoff expired, clear it + state.backoff_until_ms = None; + } + + // Clean old timestamps outside the window + state.request_timestamps_ms.retain(|&ts| now - ts < window_ms as i64); + + // Check if limit exceeded + if state.request_timestamps_ms.len() >= limit as usize { + // Enter backoff mode after threshold consecutive failures + let failed = state.failed_count + 1; + state.failed_count = failed; + + if failed >= failed_threshold { + let backoff_minutes = backoff_start_minutes * (1u64 << ((failed - failed_threshold) as u64).min(7)); // Cap at 2^7 = 128x + let backoff_seconds = (backoff_minutes * 60).min(backoff_max_hours * 3600); + state.backoff_until_ms = Some(now + (backoff_seconds as i64 * 1000)); + return (false, Some(backoff_seconds)); + } + + return (false, None); + } + + // Record this request + state.request_timestamps_ms.push(now); + (true, None) + } + + /// Reset rate limit and backoff state on successful login. + pub fn reset(&self, ip: &str) { + let mut inner = self.inner.lock().unwrap(); + inner.state.remove(ip); + } + + /// Record a failed login attempt (for backoff calculation). + pub fn record_failure(&self, ip: &str, failed_threshold: u32, backoff_start_minutes: u64, backoff_max_hours: u64) -> Option { + let mut inner = self.inner.lock().unwrap(); + let now = now_ms(); + let state = inner.state.entry(ip.to_string()).or_default(); + + state.failed_count += 1; + + if state.failed_count >= failed_threshold { + let backoff_minutes = backoff_start_minutes * (1u64 << ((state.failed_count - failed_threshold) as u64).min(7)); + let backoff_seconds = (backoff_minutes * 60).min(backoff_max_hours * 3600); + state.backoff_until_ms = Some(now + (backoff_seconds as i64 * 1000)); + return Some(backoff_seconds); + } + + None + } +} + +impl Default for LocalAdminRateLimiter { + fn default() -> Self { + Self::new() + } +} + +/// Get current time in milliseconds since Unix epoch. +fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64 +} + /// Shared application state for admin endpoints. #[derive(Clone)] pub struct AppState { @@ -93,12 +243,27 @@ pub struct AppState { pub metrics: super::super::middleware::Metrics, pub version_state: VersionState, pub task_registry: Arc, + pub redis_store: Option, + pub pod_id: String, + pub seal_key: SealKey, + pub local_rate_limiter: LocalAdminRateLimiter, } impl AppState { pub fn new( config: MiroirConfig, metrics: super::super::middleware::Metrics, + seal_key: SealKey, + ) -> Self { + Self::with_redis(config, metrics, None, "unknown".into(), seal_key) + } + + pub fn with_redis( + config: MiroirConfig, + metrics: super::super::middleware::Metrics, + redis_store: Option, + pod_id: String, + seal_key: SealKey, ) -> Self { // Build initial topology from config let mut topology = Topology::new( @@ -129,6 +294,10 @@ impl AppState { metrics, version_state, task_registry: Arc::new(InMemoryTaskRegistry::new()), + redis_store, + pod_id, + seal_key, + local_rate_limiter: LocalAdminRateLimiter::new(), } } @@ -293,6 +462,303 @@ where } } +/// POST /_miroir/ui/search/{index}/rotate-scoped-key — manual rotation trigger. +/// +/// Admin-gated endpoint that initiates a scoped key rotation for the given index. +/// Set `force: true` in the request body to bypass the timing gate. +pub async fn rotate_scoped_key_handler( + State(state): State, + Path(index): Path, + Json(body): Json, +) -> Result, (StatusCode, String)> +where + S: Clone + Send + Sync + 'static, + AppState: FromRef, +{ + let app_state = AppState::from_ref(&state); + + let redis = app_state.redis_store.clone().ok_or_else(|| { + ( + StatusCode::PRECONDITION_FAILED, + "scoped key rotation requires Redis task store".into(), + ) + })?; + + if !app_state.config.search_ui.enabled { + return Err(( + StatusCode::PRECONDITION_FAILED, + "search_ui is not enabled".into(), + )); + } + + let rotation_state = ScopedKeyRotationState { + config: app_state.config.clone(), + redis, + pod_id: app_state.pod_id.clone(), + }; + + info!( + index = %index, + force = body.force, + pod_id = %app_state.pod_id, + "manual scoped key rotation triggered" + ); + + match scoped_key_rotation::check_and_rotate(&rotation_state, &index, body.force).await { + Ok(response) => Ok(Json(response)), + Err(e) => { + error!(index = %index, error = %e, "manual scoped key rotation failed"); + Err((StatusCode::INTERNAL_SERVER_ERROR, e)) + } + } +} + +/// Parse a rate limit string like "10/minute" into (limit, window_seconds). +fn parse_rate_limit(s: &str) -> Result<(u64, u64), String> { + let parts: Vec<&str> = s.split('/').collect(); + if parts.len() != 2 { + return Err(format!("invalid rate limit format: '{}', expected 'N/UNIT'", s)); + } + let limit: u64 = parts[0].parse() + .map_err(|_| format!("invalid limit number: '{}'", parts[0]))?; + let window_seconds = match parts[1] { + "second" | "s" => 1, + "minute" | "m" => 60, + "hour" | "h" => 3600, + "day" | "d" => 86400, + unit => return Err(format!("invalid time unit: '{}', expected second/minute/hour/day", unit)), + }; + Ok((limit, window_seconds)) +} + +/// Generate a random session ID. +fn generate_session_id() -> String { + let mut bytes = [0u8; 24]; + rand::rngs::OsRng.fill_bytes(&mut bytes); + hex::encode(&bytes) +} + +/// POST /_miroir/admin/login — admin login with rate limiting and exponential backoff. +/// +/// Request body: +/// ```json +/// { "admin_key": "..." } +/// ``` +/// +/// On success, sets a `miroir_admin_session` cookie and returns: +/// ```json +/// { "success": true } +/// ``` +/// +/// Rate limiting (per source IP): +/// - 10 requests per minute (configurable via `admin_ui.rate_limit.per_ip`) +/// - After 5 consecutive failed attempts, exponential backoff applies: +/// - 10m, 20m, 40m, ... up to 24h cap +/// +/// Successful login resets both the rate limit counter and backoff state. +pub async fn admin_login( + State(state): State, + headers: HeaderMap, + Json(body): Json, +) -> Response +where + S: Clone + Send + Sync + 'static, + AppState: FromRef, +{ + let state = AppState::from_ref(&state); + + // Extract source IP from X-Forwarded-For or X-Real-IP (trust proxy) + let source_ip = headers + .get("x-forwarded-for") + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.split(',').next()) + .or_else(|| headers.get("x-real-ip").and_then(|v| v.to_str().ok())) + .unwrap_or("unknown") + .trim() + .to_string(); + + // Parse rate limit config + let (limit, window_seconds) = match parse_rate_limit(&state.config.admin_ui.rate_limit.per_ip) { + Ok(parsed) => parsed, + Err(e) => { + error!(error = %e, "invalid admin_ui.rate_limit.per_ip config"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(AdminLoginResponse { + success: false, + message: Some("Rate limit configuration error".into()), + }), + ).into_response(); + } + }; + + // Check rate limit and backoff + let backend = state.config.admin_ui.rate_limit.backend.as_str(); + if backend == "redis" { + if let Some(ref redis) = state.redis_store { + match redis.check_rate_limit_admin_login(&source_ip, limit, window_seconds) { + Ok((allowed, wait_seconds)) => { + if !allowed { + if let Some(ws) = wait_seconds { + warn!( + source_ip_hash = hash_for_log(&source_ip), + wait_seconds = ws, + "admin login rate limited (backoff)" + ); + return ( + StatusCode::TOO_MANY_REQUESTS, + Json(AdminLoginResponse { + success: false, + message: Some(format!( + "Too many failed login attempts. Try again in {} seconds.", + ws + )), + }), + ).into_response(); + } else { + return ( + StatusCode::TOO_MANY_REQUESTS, + Json(AdminLoginResponse { + success: false, + message: Some("Too many login attempts. Please try again later.".into()), + }), + ).into_response(); + } + } + // Allowed, proceed + } + Err(e) => { + error!(error = %e, "failed to check admin login rate limit"); + // Continue anyway on error (fail-open) + } + } + } + } else if backend == "local" { + // Local backend rate limiting + let (allowed, wait_seconds) = state.local_rate_limiter.check( + &source_ip, + limit, + window_seconds * 1000, + state.config.admin_ui.rate_limit.failed_attempt_threshold, + state.config.admin_ui.rate_limit.backoff_start_minutes, + state.config.admin_ui.rate_limit.backoff_max_hours * 60, + ); + if !allowed { + warn!( + source_ip_hash = hash_for_log(&source_ip), + wait_seconds = ?wait_seconds, + "admin login rate limited (local backend)" + ); + return ( + StatusCode::TOO_MANY_REQUESTS, + Json(AdminLoginResponse { + success: false, + message: if let Some(ws) = wait_seconds { + Some(format!( + "Too many failed login attempts. Try again in {} seconds.", + ws + )) + } else { + Some("Too many login attempts. Please try again later.".into()) + }, + }), + ).into_response(); + } + } + + // Verify admin_key (constant-time comparison to prevent timing side-channels) + use subtle::ConstantTimeEq as _; + if body.admin_key.as_bytes().ct_eq(state.config.admin.api_key.as_bytes()).into() { + // Successful login - reset rate limit counters + if backend == "redis" { + if let Some(ref redis) = state.redis_store { + if let Err(e) = redis.reset_rate_limit_admin_login(&source_ip) { + warn!(error = %e, "failed to reset admin login rate limit"); + } + } + } else if backend == "local" { + state.local_rate_limiter.reset(&source_ip); + } + + // Generate session ID and seal it + let session_id = generate_session_id(); + let sealed = match seal_session(&session_id, &state.seal_key) { + Ok(sealed) => sealed, + Err(e) => { + error!(error = %e, "failed to seal admin session"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(AdminLoginResponse { + success: false, + message: Some("Failed to create session".into()), + }), + ).into_response(); + } + }; + + info!( + source_ip_hash = hash_for_log(&source_ip), + session_prefix = &session_id[..8], + "admin login successful" + ); + + // Set cookie and return success + ( + StatusCode::OK, + [ + ("Set-Cookie", format!("{}={}; HttpOnly; Secure; SameSite=Strict; Path=/; Max-Age={}", + COOKIE_NAME, sealed, state.config.admin_ui.session_ttl_s)), + ], + Json(AdminLoginResponse { + success: true, + message: None, + }), + ).into_response() + } else { + // Wrong admin_key - record failure for backoff tracking + warn!( + source_ip_hash = hash_for_log(&source_ip), + "admin login failed: invalid admin_key" + ); + + if backend == "redis" { + if let Some(ref redis) = state.redis_store { + let backoff_start_minutes = state.config.admin_ui.rate_limit.backoff_start_minutes; + let backoff_max_hours = state.config.admin_ui.rate_limit.backoff_max_hours; + let failed_threshold = state.config.admin_ui.rate_limit.failed_attempt_threshold; + + if let Err(e) = redis.record_failure_admin_login( + &source_ip, + failed_threshold, + backoff_start_minutes, + backoff_max_hours, + ) { + warn!(error = %e, "failed to record admin login failure"); + } + } + } else if backend == "local" { + let backoff_start_minutes = state.config.admin_ui.rate_limit.backoff_start_minutes; + let backoff_max_hours = state.config.admin_ui.rate_limit.backoff_max_hours; + let failed_threshold = state.config.admin_ui.rate_limit.failed_attempt_threshold; + + state.local_rate_limiter.record_failure( + &source_ip, + failed_threshold, + backoff_start_minutes, + backoff_max_hours * 60, + ); + } + + ( + StatusCode::UNAUTHORIZED, + Json(AdminLoginResponse { + success: false, + message: Some("Invalid admin key".into()), + }), + ).into_response() + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/miroir-proxy/src/routes/mod.rs b/crates/miroir-proxy/src/routes/mod.rs index 6c6fb72..acdd21c 100644 --- a/crates/miroir-proxy/src/routes/mod.rs +++ b/crates/miroir-proxy/src/routes/mod.rs @@ -7,6 +7,7 @@ pub mod health; pub mod indexes; pub mod keys; pub mod search; +pub mod session; pub mod settings; pub mod tasks; pub mod version; diff --git a/crates/miroir-proxy/src/routes/session.rs b/crates/miroir-proxy/src/routes/session.rs new file mode 100644 index 0000000..973bd97 --- /dev/null +++ b/crates/miroir-proxy/src/routes/session.rs @@ -0,0 +1,458 @@ +//! Session endpoints for admin login and search UI (plan §9, §13.19, §13.21). +//! +//! Admin login: +//! - POST /_miroir/admin/login - credentials in body, returns CSRF token +//! - GET /_miroir/admin/session - validate session, refresh CSRF token +//! - POST /_miroir/admin/logout - revoke session +//! +//! Search UI session: +//! - GET /_miroir/ui/search/{index}/session - JWT session token with origin check + +use axum::{ + extract::{Extension, FromRef, Path, State}, + http::{HeaderMap, StatusCode}, + Json, + response::{IntoResponse, Response}, +}; +use miroir_core::task_store::{NewAdminSession, TaskStore}; +use serde::{Deserialize, Serialize}; +use tracing::{info, warn}; + +use crate::auth::{AdminSessionId, build_csp_header, generate_csrf_token, validate_origin}; + +use super::admin_endpoints::AppState; + +/// Hash a PII value (session ID, IP, username) for safe log correlation. +fn hash_for_log(value: &str) -> String { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + value.hash(&mut hasher); + format!("{:016x}", hasher.finish()) +} + +/// Truncate a session ID to its prefix for logging (avoids full session ID in logs). +fn session_prefix(session_id: &str) -> &str { + &session_id[..session_id.len().min(8)] +} + +/// Admin login request body. +#[derive(Debug, Deserialize)] +pub struct AdminLoginRequest { + pub admin_key: String, +} + +/// Admin login response with CSRF token. +#[derive(Debug, Serialize)] +pub struct AdminLoginResponse { + pub session_id: String, + pub csrf_token: String, + pub expires_at: i64, +} + +/// Admin session validation response. +#[derive(Debug, Serialize)] +pub struct AdminSessionResponse { + pub valid: bool, + pub csrf_token: Option, + pub expires_at: Option, +} + +/// Search UI session response. +#[derive(Debug, Serialize)] +pub struct SearchUiSessionResponse { + pub token: String, + pub expires_at: i64, +} + +/// POST /_miroir/admin/login - admin login with credentials. +/// +/// Expects `admin_key` in request body. Validates against `admin.api_key`. +/// On success, creates an admin session with CSRF token and returns: +/// - Set-Cookie: sealed session ID (HttpOnly, Secure, SameSite=Strict) +/// - JSON response with session_id, csrf_token, expires_at +/// +/// Origin is checked against `admin_ui.allowed_origins` (default "same-origin"). +pub async fn admin_login( + State(state): State, + headers: HeaderMap, + Json(body): Json, +) -> Result, (StatusCode, String)> +where + S: Clone + Send + Sync + 'static, + AppState: FromRef, +{ + let state = AppState::from_ref(&state); + let config = &state.config; + + if !config.admin_ui.enabled { + return Err(( + StatusCode::PRECONDITION_FAILED, + "admin_ui is not enabled".into(), + )); + } + + // Origin check (plan §9) + let origin_verdict = validate_origin(&headers, &config.admin_ui.allowed_origins, true); + if matches!(origin_verdict, crate::auth::OriginVerdict::Forbidden) { + warn!( + allowed_origins = ?config.admin_ui.allowed_origins, + "admin login origin check failed" + ); + return Err(( + StatusCode::FORBIDDEN, + "origin not allowed".into(), + )); + } + + // Validate admin key (constant-time compare) + let expected_key = &config.admin.api_key; + if !crate::auth::constant_time_compare(body.admin_key.as_bytes(), expected_key.as_bytes()) { + return Err(( + StatusCode::UNAUTHORIZED, + "invalid admin key".into(), + )); + } + + // Generate session ID and CSRF token + let session_id = format!("admin_sess_{}", generate_csrf_token()); + let csrf_token = generate_csrf_token(); + + // Calculate expiration + let now = epoch_seconds(); + let expires_at = now + config.admin_ui.session_ttl_s as i64; + + // Hash the admin key for storage (never store the key itself) + let admin_key_hash = hash_admin_key(expected_key); + + // Extract user agent and source IP for audit + let user_agent = headers + .get("user-agent") + .and_then(|h| h.to_str().ok()) + .map(String::from); + let source_ip = headers + .get("x-forwarded-for") + .and_then(|h| h.to_str().ok()) + .or_else(|| headers.get("x-real-ip").and_then(|h| h.to_str().ok())) + .map(String::from); + + // Create session in task store + let new_session = NewAdminSession { + session_id: session_id.clone(), + csrf_token: csrf_token.clone(), + admin_key_hash, + created_at: now, + expires_at, + user_agent, + source_ip, + }; + + if let Some(ref redis) = state.redis_store { + if let Err(e) = redis.insert_admin_session(&new_session) { + warn!(error = %e, "failed to create admin session in Redis"); + return Err(( + StatusCode::INTERNAL_SERVER_ERROR, + "failed to create session".into(), + )); + } + } else { + return Err(( + StatusCode::PRECONDITION_FAILED, + "admin sessions require Redis task store".into(), + )); + } + + info!( + session_prefix = session_prefix(&session_id), + expires_at = expires_at, + "admin login successful" + ); + + Ok(Json(AdminLoginResponse { + session_id, + csrf_token, + expires_at, + })) +} + +/// GET /_miroir/admin/session - validate admin session and refresh CSRF token. +/// +/// Requires sealed admin session cookie. Returns current session info +/// with a fresh CSRF token (rotated on each call). +pub async fn admin_session( + State(state): State, + Extension(admin_session): Extension, +) -> Result, (StatusCode, String)> +where + S: Clone + Send + Sync + 'static, + AppState: FromRef, +{ + let state = AppState::from_ref(&state); + let session_id = admin_session.0; + + let Some(redis) = state.redis_store.as_ref() else { + return Err(( + StatusCode::PRECONDITION_FAILED, + "admin sessions require Redis task store".into(), + )); + }; + + // Look up session + let Some(session) = redis.get_admin_session(&session_id).map_err(|e| { + warn!(error = %e, session_prefix = session_prefix(&session_id), "failed to get admin session"); + (StatusCode::INTERNAL_SERVER_ERROR, "failed to get session".into()) + })? else { + return Ok(Json(AdminSessionResponse { + valid: false, + csrf_token: None, + expires_at: None, + })); + }; + + // Check if revoked + if session.revoked { + return Ok(Json(AdminSessionResponse { + valid: false, + csrf_token: None, + expires_at: None, + })); + } + + // Check expiration + let now = epoch_seconds(); + if session.expires_at < now { + return Ok(Json(AdminSessionResponse { + valid: false, + csrf_token: None, + expires_at: None, + })); + } + + // Generate fresh CSRF token + let new_csrf_token = generate_csrf_token(); + + // Update session with new CSRF token + let updated_session = NewAdminSession { + session_id: session.session_id.clone(), + csrf_token: new_csrf_token.clone(), + admin_key_hash: session.admin_key_hash.clone(), + created_at: session.created_at, + expires_at: session.expires_at, + user_agent: session.user_agent.clone(), + source_ip: session.source_ip.clone(), + }; + + redis.insert_admin_session(&updated_session).map_err(|e| { + warn!(error = %e, session_prefix = session_prefix(&session_id), "failed to refresh CSRF token"); + (StatusCode::INTERNAL_SERVER_ERROR, "failed to refresh session".into()) + })?; + + Ok(Json(AdminSessionResponse { + valid: true, + csrf_token: Some(new_csrf_token), + expires_at: Some(session.expires_at), + })) +} + +/// POST /_miroir/admin/logout - revoke admin session. +pub async fn admin_logout( + State(state): State, + Extension(admin_session): Extension, +) -> Result<(), (StatusCode, String)> +where + S: Clone + Send + Sync + 'static, + AppState: FromRef, +{ + let state = AppState::from_ref(&state); + let session_id = admin_session.0; + + let Some(redis) = state.redis_store.as_ref() else { + return Err(( + StatusCode::PRECONDITION_FAILED, + "admin sessions require Redis task store".into(), + )); + }; + + redis.revoke_admin_session(&session_id).map_err(|e| { + warn!(error = %e, session_prefix = session_prefix(&session_id), "failed to revoke admin session"); + (StatusCode::INTERNAL_SERVER_ERROR, "failed to revoke session".into()) + })?; + + info!(session_prefix = session_prefix(&session_id), "admin logout successful"); + + Ok(()) +} + +/// GET /_miroir/ui/search/{index}/session - search UI session endpoint with origin check. +/// +/// Returns a JWT session token for the given index. Authentication mode depends on +/// `search_ui.auth.mode`: +/// - `public`: unauthenticated, IP rate-limited +/// - `shared_key`: requires `X-Search-UI-Key` header +/// - `oauth_proxy`: requires upstream auth-proxy headers +/// +/// Origin is checked against `search_ui.allowed_origins` (default ["*"] in public mode). +/// CSP header is added from `search_ui.csp` with `csp_overrides` merged. +pub async fn search_ui_session( + State(state): State, + Path(index): Path, + headers: HeaderMap, +) -> Result +where + S: Clone + Send + Sync + 'static, + AppState: FromRef, +{ + let state = AppState::from_ref(&state); + let config = &state.config; + + if !config.search_ui.enabled { + return Err(( + StatusCode::PRECONDITION_FAILED, + "search_ui is not enabled".into(), + )); + } + + // Origin check (plan §9) + let is_public = config.search_ui.auth.mode == "public"; + let default_allowed = if is_public { vec!["*".into()] } else { vec![] }; + let allowed_origins = if config.search_ui.allowed_origins.is_empty() { + &default_allowed + } else { + &config.search_ui.allowed_origins + }; + + let origin_verdict = validate_origin(&headers, allowed_origins, is_public); + if matches!(origin_verdict, crate::auth::OriginVerdict::Forbidden) { + warn!( + index = %index, + allowed_origins = ?allowed_origins, + "search UI session origin check failed" + ); + return Err(( + StatusCode::FORBIDDEN, + "origin not allowed".into(), + )); + } + + // Authentication based on mode + let subject = match config.search_ui.auth.mode.as_str() { + "public" => "anonymous".to_string(), + "shared_key" => { + let key = headers + .get("X-Search-UI-Key") + .and_then(|h| h.to_str().ok()) + .ok_or_else(|| { + (StatusCode::UNAUTHORIZED, "missing X-Search-UI-Key header".into()) + })?; + let expected_key = std::env::var(&config.search_ui.auth.shared_key_env) + .unwrap_or_default(); + if !crate::auth::constant_time_compare(key.as_bytes(), expected_key.as_bytes()) { + return Err((StatusCode::UNAUTHORIZED, "invalid search UI key".into())); + } + "shared_key_user".to_string() + } + "oauth_proxy" => { + let user = headers + .get(&config.search_ui.auth.oauth_proxy.user_header) + .and_then(|h| h.to_str().ok()) + .ok_or_else(|| { + ( + StatusCode::UNAUTHORIZED, + format!("missing {}", config.search_ui.auth.oauth_proxy.user_header), + ) + })?; + user.to_string() + } + _ => { + return Err(( + StatusCode::INTERNAL_SERVER_ERROR, + "invalid search_ui.auth.mode".into(), + )) + } + }; + + // Generate JWT + let jwt_secret = std::env::var(&config.search_ui.auth.jwt_secret_env) + .map_err(|_| { + ( + StatusCode::INTERNAL_SERVER_ERROR, + format!("{} not set", config.search_ui.auth.jwt_secret_env), + ) + })?; + + let auth_state = crate::auth::AuthState { + master_key: String::new(), + admin_key: String::new(), + jwt_primary: Some(jwt_secret), + jwt_previous: std::env::var(&config.search_ui.auth.jwt_secret_previous_env) + .ok() + .filter(|v| !v.is_empty()), + seal_key: crate::admin_session::SealKey::from_bytes([0u8; 32]), + revoked_sessions: std::sync::Arc::new(dashmap::DashMap::new()), + admin_session_revoked_total: state.metrics.admin_session_revoked_total(), + }; + + let token = auth_state + .sign_jwt( + &subject, + &index, + "search", + config.search_ui.auth.session_ttl_s, + ) + .ok_or_else(|| { + (StatusCode::INTERNAL_SERVER_ERROR, "failed to sign JWT".into()) + })?; + + let expires_at = epoch_seconds() + config.search_ui.auth.session_ttl_s as i64; + + info!( + index = %index, + subject_hash = hash_for_log(&subject), + expires_at = expires_at, + "search UI session created" + ); + + // Build CSP header + let csp_value = build_csp_header(&config.search_ui.csp, &config.search_ui.csp_overrides); + + // Build response with CSP header + let response = SearchUiSessionResponse { token, expires_at }; + let mut resp = Json(response).into_response(); + resp.headers_mut().insert( + "Content-Security-Policy", + csp_value.parse().map_err(|_| { + (StatusCode::INTERNAL_SERVER_ERROR, "invalid CSP header".into()) + })?, + ); + + Ok(resp) +} + +/// Hash an admin key for storage (SHA-256). +fn hash_admin_key(key: &str) -> String { + use sha2::{Digest, Sha256}; + let mut hasher = Sha256::new(); + hasher.update(key.as_bytes()); + format!("{:x}", hasher.finalize()) +} + +fn epoch_seconds() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_hash_admin_key() { + let key1 = "test-admin-key"; + let key2 = "test-admin-key"; + let key3 = "different-key"; + + assert_eq!(hash_admin_key(key1), hash_admin_key(key2)); + assert_ne!(hash_admin_key(key1), hash_admin_key(key3)); + } +} diff --git a/crates/miroir-proxy/src/scoped_key_rotation.rs b/crates/miroir-proxy/src/scoped_key_rotation.rs new file mode 100644 index 0000000..1b39887 --- /dev/null +++ b/crates/miroir-proxy/src/scoped_key_rotation.rs @@ -0,0 +1,395 @@ +//! Scoped Meilisearch key rotation (plan §13.21). +//! +//! Implements the leader-based rotation sequence: +//! 1. Mint new scoped key via Meilisearch `POST /keys` +//! 2. Write new generation to Redis hash +//! 3. Wait for all live pods to observe new generation (beacon check) +//! 4. Drain wait for stragglers +//! 5. `DELETE /keys/{previous_uid}` on all Meilisearch nodes +//! 6. Clear previous from Redis hash + +use miroir_core::config::MiroirConfig; +use miroir_core::task_store::{RedisTaskStore, SearchUiScopedKey, TaskStore}; +use std::sync::Arc; +use std::time::Duration; +use tracing::{info, warn, error}; + +use crate::routes::indexes::MeilisearchClient; + +/// State for the scoped key rotation background task. +#[derive(Clone)] +pub struct ScopedKeyRotationState { + pub config: Arc, + pub redis: RedisTaskStore, + pub pod_id: String, +} + +/// Response body for the manual rotation endpoint. +#[derive(serde::Serialize)] +pub struct RotateScopedKeyResponse { + pub status: String, + pub index_uid: String, + pub generation: i64, + #[serde(skip_serializing_if = "Option::is_none")] + pub previous_uid_revoked: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +/// Request body for the manual rotation endpoint. +#[derive(serde::Deserialize)] +pub struct RotateScopedKeyRequest { + /// If true, bypass the timing gate and rotate immediately. + #[serde(default)] + pub force: bool, +} + +/// Run the background scoped key rotation loop. +pub async fn run_scoped_key_rotator(state: ScopedKeyRotationState) { + if !state.config.search_ui.enabled { + return; + } + + let check_interval = Duration::from_secs(3600); // Check every hour + let mut interval = tokio::time::interval(check_interval); + + info!("scoped key rotation background task started"); + + loop { + interval.tick().await; + + // Refresh our own pod presence heartbeat + if let Err(e) = state.redis.register_pod_presence(&state.pod_id) { + warn!(pod_id = %state.pod_id, error = %e, "failed to register pod presence"); + } + + // Try to acquire the rotation leader lease + let lease_scope = "search_ui_key_rotation"; + let lease_now = now_ms(); + let lease_ttl_ms = (state.config.leader_election.lease_ttl_s as i64) * 1000; + let expires_at = lease_now + lease_ttl_ms; + + match state.redis.try_acquire_leader_lease(lease_scope, &state.pod_id, expires_at, lease_now) { + Ok(true) => {} + Ok(false) => continue, // Another pod holds the lease + Err(e) => { + warn!(error = %e, "failed to acquire rotation leader lease"); + continue; + } + } + + // We are the leader — check each index for rotation need + let indexes = discover_scoped_indexes(&state).await; + for index_uid in indexes { + if let Err(e) = check_and_rotate(&state, &index_uid, false).await { + error!(index = %index_uid, error = %e, "scoped key rotation failed"); + } + } + + // Renew the lease + let _ = state.redis.renew_leader_lease( + lease_scope, + &state.pod_id, + now_ms() + lease_ttl_ms, + ); + } +} + +/// Check if a scoped key needs rotation and perform it if so. +pub async fn check_and_rotate( + state: &ScopedKeyRotationState, + index_uid: &str, + force: bool, +) -> Result { + let current = state.redis.get_search_ui_scoped_key(index_uid) + .map_err(|e| format!("redis read failed: {e}"))?; + + // Timing gate check (skip if force) + if !force { + if !should_rotate(¤t, &state.config) { + return Ok(RotateScopedKeyResponse { + status: "skipped".into(), + index_uid: index_uid.into(), + generation: current.as_ref().map(|k| k.generation).unwrap_or(0), + previous_uid_revoked: None, + error: None, + }); + } + } + + // Step 1: Mint new scoped key via Meilisearch POST /keys + let client = MeilisearchClient::new(state.config.node_master_key.clone()); + let (new_key, new_uid) = mint_scoped_key(&client, &state.config, index_uid).await?; + + // Step 2: Write new generation to Redis + let new_generation = current.as_ref().map(|k| k.generation + 1).unwrap_or(1); + let previous_uid = current.as_ref().map(|k| k.primary_uid.clone()); + let previous_key = current.as_ref().map(|k| k.primary_key.clone()); + + let scoped_key = SearchUiScopedKey { + index_uid: index_uid.into(), + primary_key: new_key.clone(), + primary_uid: new_uid.clone(), + previous_key, + previous_uid: previous_uid.clone(), + rotated_at: now_ms(), + generation: new_generation, + }; + + state.redis.set_search_ui_scoped_key(&scoped_key) + .map_err(|e| format!("redis write failed: {e}"))?; + + info!( + index = %index_uid, + generation = new_generation, + "new scoped key minted, waiting for pod observation" + ); + + // Step 3: Observe our own beacon immediately + state.redis.observe_search_ui_scoped_key( + &state.pod_id, + index_uid, + new_generation, + ).map_err(|e| format!("beacon write failed: {e}"))?; + + // Step 4: Wait for drain period and check beacons + let drain_s = state.config.search_ui.scoped_key_rotation_drain_s; + let drain_deadline = tokio::time::Instant::now() + Duration::from_secs(drain_s); + + let mut previous_revoked: Option = None; + + loop { + // Get live pods + let live_pods = state.redis.get_live_pods() + .map_err(|e| format!("get_live_pods failed: {e}"))?; + + // Check if all live pods have observed the new generation + let (all_observed, unobserved) = state.redis.check_scoped_key_observation( + index_uid, + new_generation, + &live_pods, + ).map_err(|e| format!("beacon check failed: {e}"))?; + + if all_observed { + info!( + index = %index_uid, + generation = new_generation, + "all live pods observed new generation, revoking previous key" + ); + + // Step 6: Delete previous key from Meilisearch and clear from Redis + if let Some(ref prev_uid) = previous_uid { + if let Err(e) = revoke_previous_key(&client, &state.config, prev_uid).await { + warn!(previous_uid = %prev_uid, error = %e, "failed to revoke previous key, will retry"); + } else { + previous_revoked = Some(prev_uid.clone()); + // Clear previous from Redis + if let Err(e) = state.redis.clear_scoped_key_previous(index_uid) { + warn!(error = %e, "failed to clear previous key from redis"); + } + info!(index = %index_uid, previous_uid = %prev_uid, "previous scoped key revoked"); + } + } + + return Ok(RotateScopedKeyResponse { + status: "rotated".into(), + index_uid: index_uid.into(), + generation: new_generation, + previous_uid_revoked: previous_revoked, + error: None, + }); + } + + // Not all pods have caught up yet + if tokio::time::Instant::now() >= drain_deadline { + warn!( + index = %index_uid, + generation = new_generation, + unobserved = ?unobserved, + "drain wait expired, {} pods still unobserved — will retry on next tick", + unobserved.len() + ); + + return Ok(RotateScopedKeyResponse { + status: "drain_pending".into(), + index_uid: index_uid.into(), + generation: new_generation, + previous_uid_revoked: None, + error: Some(format!( + "drain wait expired: {} pods unobserved: {}", + unobserved.len(), + unobserved.join(", ") + )), + }); + } + + // Wait before rechecking (every 10 seconds) + tokio::time::sleep(Duration::from_secs(10)).await; + } +} + +/// Check if rotation should happen based on timing gate. +fn should_rotate(current: &Option, config: &MiroirConfig) -> bool { + let Some(key) = current else { + // No key exists yet — need to mint initial key + return true; + }; + + let max_age_ms = (config.search_ui.scoped_key_max_age_days as i64) * 24 * 3600 * 1000; + let rotate_before_ms = (config.search_ui.scoped_key_rotate_before_expiry_days as i64) * 24 * 3600 * 1000; + let age = now_ms() - key.rotated_at; + + // Rotate if key has been alive long enough that it will expire within rotate_before_days + age >= (max_age_ms - rotate_before_ms) +} + +/// Mint a new scoped Meilisearch key via POST /keys on all nodes. +async fn mint_scoped_key( + client: &MeilisearchClient, + config: &MiroirConfig, + index_uid: &str, +) -> Result<(String, String), String> { + let description = format!("miroir search-ui scoped key for index {}", index_uid); + let body = serde_json::json!({ + "description": description, + "actions": ["search"], + "indexes": [index_uid], + "expiresAt": null, + }); + + let mut created_key: Option = None; + let mut created_uid: Option = None; + let mut errors = Vec::new(); + + for node in &config.nodes { + match client.post_raw(&node.address, "/keys", &body).await { + Ok((status, text)) if status >= 200 && status < 300 => { + if created_key.is_none() { + let resp: serde_json::Value = serde_json::from_str(&text) + .map_err(|e| format!("parse key response: {e}"))?; + created_key = resp.get("key").and_then(|v| v.as_str()).map(String::from); + created_uid = resp.get("uid").and_then(|v| v.as_str()).map(String::from); + } + } + Ok((status, text)) => { + errors.push(format!("{}: HTTP {} — {}", node.id, status, text)); + } + Err(e) => { + errors.push(format!("{}: {}", node.id, e)); + } + } + } + + let key = created_key.ok_or_else(|| format!("failed to mint key on any node: {}", errors.join("; ")))?; + let uid = created_uid.ok_or_else(|| String::from("key created but no uid returned"))?; + + Ok((key, uid)) +} + +/// Revoke a previous scoped key from all Meilisearch nodes. +async fn revoke_previous_key( + client: &MeilisearchClient, + config: &MiroirConfig, + previous_uid: &str, +) -> Result<(), String> { + let path = format!("/keys/{}", previous_uid); + let mut errors = Vec::new(); + + for node in &config.nodes { + match client.delete_raw(&node.address, &path).await { + Ok((_status, _text)) if _status >= 200 && _status < 300 => {} + Ok((status, text)) => { + // 404 is fine — key was already revoked or never existed + if status != 404 { + errors.push(format!("{}: HTTP {} — {}", node.id, status, text)); + } + } + Err(e) => { + errors.push(format!("{}: {}", node.id, e)); + } + } + } + + if errors.is_empty() { + Ok(()) + } else { + Err(format!("partial revoke failure: {}", errors.join("; "))) + } +} + +/// Discover which indexes have scoped keys (or should have them). +/// For now, we look at existing keys in Redis. New indexes get initial +/// keys on their first search request. +async fn discover_scoped_indexes(state: &ScopedKeyRotationState) -> Vec { + // Scan for existing scoped key hashes + state.redis.list_scoped_key_indexes() + .unwrap_or_default() +} + +fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() as i64 +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn should_rotate_no_key() { + let config = MiroirConfig::default(); + assert!(should_rotate(&None, &config)); + } + + #[test] + fn should_rotate_old_key() { + let config = MiroirConfig { + search_ui: miroir_core::config::SearchUiConfig { + scoped_key_max_age_days: 60, + scoped_key_rotate_before_expiry_days: 30, + ..Default::default() + }, + ..MiroirConfig::default() + }; + + // Key created 31 days ago — should rotate (max_age - rotate_before = 30 days) + let key = SearchUiScopedKey { + index_uid: "test".into(), + primary_key: "key".into(), + primary_uid: "uid".into(), + previous_key: None, + previous_uid: None, + rotated_at: now_ms() - (31 * 24 * 3600 * 1000), + generation: 1, + }; + + assert!(should_rotate(&Some(key), &config)); + } + + #[test] + fn should_not_rotate_fresh_key() { + let config = MiroirConfig { + search_ui: miroir_core::config::SearchUiConfig { + scoped_key_max_age_days: 60, + scoped_key_rotate_before_expiry_days: 30, + ..Default::default() + }, + ..MiroirConfig::default() + }; + + // Key created 10 days ago — should NOT rotate yet + let key = SearchUiScopedKey { + index_uid: "test".into(), + primary_key: "key".into(), + primary_uid: "uid".into(), + previous_key: None, + previous_uid: None, + rotated_at: now_ms() - (10 * 24 * 3600 * 1000), + generation: 1, + }; + + assert!(!should_rotate(&Some(key), &config)); + } +}