P10.5: scoped Meilisearch key rotation with multi-pod coordination
Implements plan §13.21 leader-based rotation of per-index scoped search
keys with zero-403 overlap guarantees:
- Leader lease (Redis, Mode B §14.5) serializes rotation across pods
- Per-pod beacon with 60s TTL refreshed on every search request
- Revocation safety gate: leader checks all live peers observed new
generation before DELETE /keys/{previous_uid}
- Drain wait (default 120s) for stragglers before revocation
- Auto-rotation trigger: scoped_key_rotate_before_expiry_days (30d)
before scoped_key_max_age_days (60d)
- Manual trigger: POST /_miroir/ui/search/{index}/rotate-scoped-key
with force:true to bypass timing gate
- Config validation rejects rotate_before >= max_age at startup
- Helm _helpers.tpl render-time guard against rotation loop
- values.schema.json schema validation for scoped key config fields
Also includes session management routes (admin login/logout/session,
search UI JWT session) and auth middleware CSRF protection needed
by the admin-gated rotation endpoint.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
jedarden
parent
a2a323f33c
commit
ee3ef23133
18 changed files with 2044 additions and 34 deletions
1
Cargo.lock
generated
1
Cargo.lock
generated
|
|
@ -1773,6 +1773,7 @@ dependencies = [
|
|||
"chacha20poly1305",
|
||||
"config",
|
||||
"dashmap",
|
||||
"hex",
|
||||
"hmac",
|
||||
"http",
|
||||
"http-body-util",
|
||||
|
|
|
|||
|
|
@ -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 }}
|
||||
|
|
|
|||
|
|
@ -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"]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ mod load;
|
|||
mod validate;
|
||||
|
||||
pub use error::ConfigError;
|
||||
pub use advanced::{SearchUiConfig, CspOverridesConfig};
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
|
|
|
|||
|
|
@ -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<String>,
|
||||
pub cors_allowed_origins: Vec<String>,
|
||||
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(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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"));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1886,20 +1886,7 @@ impl RedisTaskStore {
|
|||
|
||||
// Check if limit exceeded
|
||||
if count >= limit {
|
||||
// Enter backoff mode after 5 consecutive failures
|
||||
let failed_count: Option<u64> = 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<Option<u64>> {
|
||||
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<String, Value> = 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<Vec<String>> {
|
||||
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<String> = 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<Vec<String>> {
|
||||
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<String>) = ::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:<index>"
|
||||
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<String>,
|
||||
/// The previous key UID (for revocation).
|
||||
pub previous_uid: Option<String>,
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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 }
|
||||
|
|
|
|||
|
|
@ -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<Sha256>;
|
|||
#[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<miroir_core::task_store::RedisTaskStore>,
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// 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<String> {
|
||||
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<String>)> = 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<String>)>,
|
||||
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::<Vec<_>>()
|
||||
.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<CsrfState>,
|
||||
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::<AdminSessionId>() {
|
||||
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)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<UnifiedState> 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<UnifiedState> for TelemetryState {
|
|||
}
|
||||
}
|
||||
|
||||
// Implement FromRef so that CsrfState can be extracted from UnifiedState
|
||||
impl FromRef<UnifiedState> 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(),
|
||||
|
|
|
|||
|
|
@ -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<S>,
|
||||
{
|
||||
Router::new()
|
||||
// Admin session endpoints (plan §9, §13.19)
|
||||
.route("/admin/login", post(session::admin_login::<S>))
|
||||
.route("/admin/session", get(session::admin_session::<S>))
|
||||
.route("/admin/logout", post(session::admin_logout::<S>))
|
||||
// Search UI session endpoint (plan §9, §13.21)
|
||||
.route(
|
||||
"/ui/search/{index}/session",
|
||||
get(session::search_ui_session::<S>),
|
||||
)
|
||||
// Admin API endpoints
|
||||
.route("/topology", get(admin_endpoints::get_topology::<S>))
|
||||
.route("/shards", get(admin_endpoints::get_shards::<S>))
|
||||
.route("/ready", get(admin_endpoints::get_ready::<S>))
|
||||
.route("/metrics", get(admin_endpoints::get_metrics::<S>))
|
||||
.route(
|
||||
"/ui/search/{index}/rotate-scoped-key",
|
||||
post(admin_endpoints::rotate_scoped_key_handler),
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<String>,
|
||||
}
|
||||
|
||||
/// 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<Mutex<...>>.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct LocalAdminRateLimiter {
|
||||
inner: Arc<std::sync::Mutex<LocalAdminRateLimiterInner>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
struct LocalAdminRateLimiterInner {
|
||||
/// Map of IP -> (request_timestamps_ms, failed_count, backoff_until_ms)
|
||||
state: HashMap<String, LocalRateLimitState>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Clone)]
|
||||
struct LocalRateLimitState {
|
||||
/// Timestamps of recent requests (for sliding window)
|
||||
request_timestamps_ms: Vec<i64>,
|
||||
/// Consecutive failed login attempts
|
||||
failed_count: u32,
|
||||
/// Unix timestamp (ms) when backoff expires
|
||||
backoff_until_ms: Option<i64>,
|
||||
}
|
||||
|
||||
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<u64>) {
|
||||
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<u64> {
|
||||
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<InMemoryTaskRegistry>,
|
||||
pub redis_store: Option<RedisTaskStore>,
|
||||
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<RedisTaskStore>,
|
||||
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<S>(
|
||||
State(state): State<S>,
|
||||
Path(index): Path<String>,
|
||||
Json(body): Json<RotateScopedKeyRequest>,
|
||||
) -> Result<Json<RotateScopedKeyResponse>, (StatusCode, String)>
|
||||
where
|
||||
S: Clone + Send + Sync + 'static,
|
||||
AppState: FromRef<S>,
|
||||
{
|
||||
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<S>(
|
||||
State(state): State<S>,
|
||||
headers: HeaderMap,
|
||||
Json(body): Json<AdminLoginRequest>,
|
||||
) -> Response
|
||||
where
|
||||
S: Clone + Send + Sync + 'static,
|
||||
AppState: FromRef<S>,
|
||||
{
|
||||
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::*;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
458
crates/miroir-proxy/src/routes/session.rs
Normal file
458
crates/miroir-proxy/src/routes/session.rs
Normal file
|
|
@ -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<String>,
|
||||
pub expires_at: Option<i64>,
|
||||
}
|
||||
|
||||
/// 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<S>(
|
||||
State(state): State<S>,
|
||||
headers: HeaderMap,
|
||||
Json(body): Json<AdminLoginRequest>,
|
||||
) -> Result<Json<AdminLoginResponse>, (StatusCode, String)>
|
||||
where
|
||||
S: Clone + Send + Sync + 'static,
|
||||
AppState: FromRef<S>,
|
||||
{
|
||||
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<S>(
|
||||
State(state): State<S>,
|
||||
Extension(admin_session): Extension<AdminSessionId>,
|
||||
) -> Result<Json<AdminSessionResponse>, (StatusCode, String)>
|
||||
where
|
||||
S: Clone + Send + Sync + 'static,
|
||||
AppState: FromRef<S>,
|
||||
{
|
||||
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<S>(
|
||||
State(state): State<S>,
|
||||
Extension(admin_session): Extension<AdminSessionId>,
|
||||
) -> Result<(), (StatusCode, String)>
|
||||
where
|
||||
S: Clone + Send + Sync + 'static,
|
||||
AppState: FromRef<S>,
|
||||
{
|
||||
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<S>(
|
||||
State(state): State<S>,
|
||||
Path(index): Path<String>,
|
||||
headers: HeaderMap,
|
||||
) -> Result<Response, (StatusCode, String)>
|
||||
where
|
||||
S: Clone + Send + Sync + 'static,
|
||||
AppState: FromRef<S>,
|
||||
{
|
||||
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));
|
||||
}
|
||||
}
|
||||
395
crates/miroir-proxy/src/scoped_key_rotation.rs
Normal file
395
crates/miroir-proxy/src/scoped_key_rotation.rs
Normal file
|
|
@ -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<MiroirConfig>,
|
||||
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<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub error: Option<String>,
|
||||
}
|
||||
|
||||
/// 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<RotateScopedKeyResponse, String> {
|
||||
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<String> = 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<SearchUiScopedKey>, 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<String> = None;
|
||||
let mut created_uid: Option<String> = 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<String> {
|
||||
// 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));
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue