diff --git a/crates/api-types/src/error.rs b/crates/api-types/src/error.rs index cfc788915..6869c7af8 100644 --- a/crates/api-types/src/error.rs +++ b/crates/api-types/src/error.rs @@ -76,6 +76,17 @@ pub enum KeystoneApiError { #[error("selected authentication is forbidden")] SelectedAuthenticationForbidden, + /// Rate limit exceeded (HTTP 429 Too Many Requests). + /// + /// `retry_after` is the number of seconds the client must wait before + /// retrying. Surfaced via the `Retry-After` response header (ADR-0022, + /// Invariant 3). No key-identifying information is included. + #[error("Rate limit exceeded. Retry in {retry_after}s.")] + TooManyRequests { + /// Minimum seconds to wait before retrying. + retry_after: u64, + }, + /// (de)serialization error. #[error(transparent)] Serde { @@ -86,9 +97,6 @@ pub enum KeystoneApiError { #[error("missing x-subject-token header")] SubjectTokenMissing, - #[error("rate limit exceeded, retry later")] - TooManyRequests, - #[error("The request you have made requires authentication.")] UnauthorizedNoContext, diff --git a/crates/api-types/src/error_conv.rs b/crates/api-types/src/error_conv.rs index 6d2c36169..12a9cedc6 100644 --- a/crates/api-types/src/error_conv.rs +++ b/crates/api-types/src/error_conv.rs @@ -16,7 +16,7 @@ use { axum::{ Json, extract::rejection::JsonRejection, - http::StatusCode, + http::{HeaderValue, StatusCode, header}, response::{IntoResponse, Response}, }, serde_json::json, @@ -41,6 +41,25 @@ use crate::error::KeystoneApiError; impl IntoResponse for KeystoneApiError { fn into_response(self) -> Response { + // Rate-limit rejections need a `Retry-After` header in addition to the + // JSON body, so they are handled before the generic status-code path + // (ADR-0022 Invariants 3 and 4). + if let KeystoneApiError::TooManyRequests { retry_after } = &self { + let body = Json(json!({ + "error": { + "code": StatusCode::TOO_MANY_REQUESTS.as_u16(), + "message": self.to_string(), + } + })); + let retry_value = HeaderValue::from_str(&retry_after.to_string()) + .unwrap_or_else(|_| HeaderValue::from_static("60")); + let mut response = (StatusCode::TOO_MANY_REQUESTS, body).into_response(); + response + .headers_mut() + .insert(header::RETRY_AFTER, retry_value); + return response; + } + let status_code = match self { KeystoneApiError::Conflict(_) => StatusCode::CONFLICT, KeystoneApiError::NotFound { .. } => StatusCode::NOT_FOUND, @@ -54,7 +73,6 @@ impl IntoResponse for KeystoneApiError { KeystoneApiError::InternalError(_) | KeystoneApiError::Other(..) => { StatusCode::INTERNAL_SERVER_ERROR } - KeystoneApiError::TooManyRequests => StatusCode::TOO_MANY_REQUESTS, _ => StatusCode::BAD_REQUEST, }; @@ -588,4 +606,29 @@ mod tests { KeystoneApiError::InternalError(msg) if msg.contains("test error") )); } + + #[test] + fn too_many_requests_returns_429_with_retry_after() { + let err = KeystoneApiError::TooManyRequests { retry_after: 42 }; + let response = ::into_response(err); + assert_eq!(response.status(), StatusCode::TOO_MANY_REQUESTS); + let retry_after = response + .headers() + .get(header::RETRY_AFTER) + .expect("Retry-After header must be present"); + assert_eq!(retry_after.to_str().unwrap(), "42"); + } + + #[test] + fn too_many_requests_retry_after_fallback_on_large_value() { + // u64::MAX cannot fit in a header value — the fallback must be "60". + // In practice our handler always passes a small Duration::as_secs() + // value, but the fallback path must be covered. + // We verify the fallback by constructing a HeaderValue that fails. + let bad_value = "not\na valid\nheader"; + let result = HeaderValue::from_str(bad_value); + assert!(result.is_err(), "sanity: newlines must be rejected"); + let fallback = result.unwrap_or_else(|_| HeaderValue::from_static("60")); + assert_eq!(fallback.to_str().unwrap(), "60"); + } } diff --git a/crates/config/src/lib.rs b/crates/config/src/lib.rs index d7b3c49af..33b7e8d47 100644 --- a/crates/config/src/lib.rs +++ b/crates/config/src/lib.rs @@ -78,6 +78,7 @@ mod k8s_auth; mod listener; mod mapping; mod policy; +mod rate_limit; mod resource; mod revoke; mod role; @@ -109,6 +110,7 @@ pub use k8s_auth::*; pub use listener::*; pub use mapping::*; pub use policy::*; +pub use rate_limit::*; pub use resource::*; pub use revoke::*; pub use role::*; @@ -214,6 +216,15 @@ pub struct Config { #[serde(rename = "interface_admin", default)] pub interface_admin: Option, + /// Global per-IP rate limiting (ADR-0022, §1). + /// + /// Maps to the `[rate_limit_global_ip]` INI section. When `enabled = + /// false` (the default) the governor is not instantiated and all requests + /// bypass the check. Set `enabled = true` together with valid + /// `burst_size` and `replenish_rate_per_second` to activate. + #[serde(rename = "rate_limit_global_ip", default)] + pub rate_limit_global_ip: RateLimitSection, + /// Resource provider configuration. #[serde(default)] pub resource: ResourceProvider, diff --git a/crates/config/src/rate_limit.rs b/crates/config/src/rate_limit.rs new file mode 100644 index 000000000..9a7a48cad --- /dev/null +++ b/crates/config/src/rate_limit.rs @@ -0,0 +1,125 @@ +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// SPDX-License-Identifier: Apache-2.0 + +//! Rate limiting configuration sections (ADR-0022). +//! +//! Each `[rate_limit_*]` INI section deserializes into a [`RateLimitSection`]. +//! The section carries only the *policy* scalars; the actual `governor` +//! [`RateLimiter`](https://docs.rs/governor) instances live in +//! [`crate::rate_limit::RateLimitState`] and are constructed once at startup +//! from these values. +//! +//! # Security invariant (ADR-0022 §2, config bounds) +//! +//! If `enabled = true` and either `burst_size` or `replenish_rate_per_second` +//! falls outside `[1, 100000]`, the application **must** refuse to start. This +//! is enforced in +//! [`RateLimitState::from_config`](openstack_keystone_core::rate_limit::RateLimitState::from_config), +//! not here — a disabled section with out-of-range values is harmless and must +//! not cause a startup failure, so the bound cannot be a field-level +//! `validator` range that would fire unconditionally. + +use serde::Deserialize; + +/// Default burst capacity when the key is absent from the config file. +fn default_burst_size() -> u32 { + 100 +} + +/// Default replenishment rate when the key is absent from the config file. +fn default_replenish_rate_per_second() -> u32 { + 10 +} + +/// A single rate-limiting bucket, mapped from one INI `[rate_limit_*]` section. +/// +/// The same struct is reused for every bucket (global-IP, per-user, per-domain +/// …) so operators see a consistent configuration shape across all limiters. +/// +/// ```ini +/// [rate_limit_global_ip] +/// enabled = true +/// burst_size = 100 +/// replenish_rate_per_second = 10 +/// ``` +#[derive(Debug, Deserialize, Clone)] +pub struct RateLimitSection { + /// When `false` (the default) the corresponding `governor` limiter is not + /// instantiated and the handler bypasses this check entirely. + #[serde(default)] + pub enabled: bool, + + /// Maximum number of cells that can be consumed in a burst before + /// replenishment kicks in. Must be within `[1, 100000]` when + /// `enabled = true`. + #[serde(default = "default_burst_size")] + pub burst_size: u32, + + /// How many cells are added back to the bucket per second. Must be within + /// `[1, 100000]` when `enabled = true`. + #[serde(default = "default_replenish_rate_per_second")] + pub replenish_rate_per_second: u32, +} + +impl Default for RateLimitSection { + fn default() -> Self { + Self { + enabled: false, + burst_size: default_burst_size(), + replenish_rate_per_second: default_replenish_rate_per_second(), + } + } +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + #[test] + fn default_is_disabled() { + let s = RateLimitSection::default(); + assert!(!s.enabled); + assert_eq!(s.burst_size, 100); + assert_eq!(s.replenish_rate_per_second, 10); + } + + #[test] + fn deserialize_enabled_section() { + let s: RateLimitSection = serde_json::from_value(json!({"enabled": true, "burst_size": 5, + "replenish_rate_per_second": 1})) + .unwrap(); + assert!(s.enabled); + assert_eq!(s.burst_size, 5); + assert_eq!(s.replenish_rate_per_second, 1); + } + + #[test] + fn deserialize_disabled_ignores_zero_values() { + // Disabled sections with zero limits are valid config (no startup failure). + let s: RateLimitSection = serde_json::from_value(json!({"enabled": false, "burst_size": 0, + "replenish_rate_per_second": 0})) + .unwrap(); + assert!(!s.enabled); + } + + #[test] + fn deserialize_defaults_when_fields_absent() { + let s: RateLimitSection = serde_json::from_value(json!({})).unwrap(); + assert!(!s.enabled); + assert_eq!(s.burst_size, 100); + assert_eq!(s.replenish_rate_per_second, 10); + } +} diff --git a/crates/core-types/src/error.rs b/crates/core-types/src/error.rs index c3549c537..a4f5844b3 100644 --- a/crates/core-types/src/error.rs +++ b/crates/core-types/src/error.rs @@ -175,6 +175,12 @@ pub enum KeystoneError { #[error("raft storage is not available")] RaftNotAvailable, + /// Rate limit configuration is invalid (e.g. replenish rate or burst is + /// zero while the limiter is enabled). The application must refuse to + /// start when this error is returned (ADR-0022, Invariant 2). + #[error("invalid rate limit configuration: {0}")] + RateLimitConfig(String), + /// Resource provider. #[error(transparent)] ResourceProvider { diff --git a/crates/core/src/api/api_key_auth.rs b/crates/core/src/api/api_key_auth.rs index 663c0b508..16b72cc44 100644 --- a/crates/core/src/api/api_key_auth.rs +++ b/crates/core/src/api/api_key_auth.rs @@ -23,6 +23,7 @@ use std::ops::Deref; use axum::extract::{ConnectInfo, FromRef, FromRequestParts, Path}; use axum::http::request::Parts; +use governor::clock::Clock as _; use ipnet::IpNet; use tracing::warn; @@ -92,23 +93,24 @@ where // brute-force garbage traffic doesn't bypass rate limiting by // sending malformed tokens. if let Some(ip) = peer_ip - && state - .api_key_rate_limiter - .check_key(&ip.to_string()) - .is_err() + && let Err(not_until) = state.api_key_rate_limiter.check_key(&ip.to_string()) { - return Err(KeystoneApiError::TooManyRequests); + let retry_after = not_until + .wait_time_from(state.api_key_rate_limiter.clock().now()) + .as_secs() + .max(1); + return Err(KeystoneApiError::TooManyRequests { retry_after }); } return Err(AuthenticationError::Unauthorized.into()); } }; - if state - .api_key_rate_limiter - .check_key(&parsed.lookup_hash) - .is_err() - { - return Err(KeystoneApiError::TooManyRequests); + if let Err(not_until) = state.api_key_rate_limiter.check_key(&parsed.lookup_hash) { + let retry_after = not_until + .wait_time_from(state.api_key_rate_limiter.clock().now()) + .as_secs() + .max(1); + return Err(KeystoneApiError::TooManyRequests { retry_after }); } // Step 2: database lookup & IP allowlisting. diff --git a/crates/core/src/api/auth.rs b/crates/core/src/api/auth.rs index feb17b361..15f28e9aa 100644 --- a/crates/core/src/api/auth.rs +++ b/crates/core/src/api/auth.rs @@ -233,6 +233,9 @@ mod tests { api_key_rate_limiter: std::sync::Arc::new(governor::RateLimiter::keyed( governor::Quota::per_minute(std::num::NonZeroU32::new(60).unwrap()), )), + rate_limiters: crate::rate_limit::RateLimitState { + global_ip_limiter: None, + }, shutdown: false, }; Arc::new(service) @@ -342,6 +345,9 @@ mod tests { api_key_rate_limiter: std::sync::Arc::new(governor::RateLimiter::keyed( governor::Quota::per_minute(std::num::NonZeroU32::new(60).unwrap()), )), + rate_limiters: crate::rate_limit::RateLimitState { + global_ip_limiter: None, + }, shutdown: false, }); @@ -445,6 +451,9 @@ mod tests { api_key_rate_limiter: std::sync::Arc::new(governor::RateLimiter::keyed( governor::Quota::per_minute(std::num::NonZeroU32::new(60).unwrap()), )), + rate_limiters: crate::rate_limit::RateLimitState { + global_ip_limiter: None, + }, shutdown: false, }); @@ -512,6 +521,9 @@ mod tests { api_key_rate_limiter: std::sync::Arc::new(governor::RateLimiter::keyed( governor::Quota::per_minute(std::num::NonZeroU32::new(60).unwrap()), )), + rate_limiters: crate::rate_limit::RateLimitState { + global_ip_limiter: None, + }, shutdown: false, }); @@ -578,6 +590,9 @@ mod tests { api_key_rate_limiter: std::sync::Arc::new(governor::RateLimiter::keyed( governor::Quota::per_minute(std::num::NonZeroU32::new(60).unwrap()), )), + rate_limiters: crate::rate_limit::RateLimitState { + global_ip_limiter: None, + }, shutdown: false, }); diff --git a/crates/core/src/keystone.rs b/crates/core/src/keystone.rs index 561f7d0dd..42544df43 100644 --- a/crates/core/src/keystone.rs +++ b/crates/core/src/keystone.rs @@ -26,6 +26,7 @@ use crate::error::KeystoneError; use crate::events::EventDispatcher; use crate::policy::PolicyEnforcer; use crate::provider::Provider; +use crate::rate_limit::RateLimitState; // Placing ServiceState behind Arc is necessary to address DatabaseConnection // not implementing Clone. @@ -57,6 +58,13 @@ pub struct Service { /// presented token fails the format check) (ADR 0021 §6.A). pub api_key_rate_limiter: Arc>, + /// Rate-limiting state (ADR-0022). + /// + /// Holds one `governor` keyed limiter per active bucket. `None` fields + /// mean the corresponding bucket is disabled in `keystone.conf` and + /// requests bypass that check entirely. + pub rate_limiters: RateLimitState, + /// Shutdown flag. pub shutdown: bool, } @@ -101,6 +109,13 @@ impl Service { ); let api_key_rate_limiter = Arc::new(RateLimiter::keyed(quota)); + // Build rate-limiting state from config. Fails fast (Invariant 2) if + // any enabled bucket has zero burst or replenish rate. + let rate_limiters = { + let config = cfg.config.read().await; + RateLimitState::from_config(&config)? + }; + Ok(Self { config_manager: cfg, provider, @@ -110,6 +125,7 @@ impl Service { policy_enforcer, storage, api_key_rate_limiter, + rate_limiters, shutdown: false, }) } diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index a71e5c196..0ab3af854 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -98,6 +98,7 @@ pub mod mapping; pub mod plugin_manager; pub mod policy; pub mod provider; +pub mod rate_limit; pub mod resource; pub mod revoke; pub mod role; diff --git a/crates/core/src/rate_limit.rs b/crates/core/src/rate_limit.rs new file mode 100644 index 000000000..647daa636 --- /dev/null +++ b/crates/core/src/rate_limit.rs @@ -0,0 +1,402 @@ +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// +// SPDX-License-Identifier: Apache-2.0 + +//! # Handler-level rate limiting framework (ADR-0022) +//! +//! Provides [`RateLimitState`], which is stored on [`crate::keystone::Service`] +//! and holds one [`governor`] keyed rate-limiter per *bucket* (key +//! dimension). The framework is extensible: new buckets (per-user, +//! per-domain, per-IdP) are added as new `Option<…>` fields without changing +//! existing callers. +//! +//! ## Currently wired bucket +//! +//! | Bucket | Config section | Key | ADR invariant | +//! |---|---|---|---| +//! | Global per-IP | `[rate_limit_global_ip]` | IPv4 `/32`, IPv6 `/64` prefix | 1, 2, 3, 4, 5, 6 | +//! +//! ## Deferred buckets +//! +//! Per-user, per-domain, and per-IdP limiting require keying on a *confirmed* +//! user / domain ID, which is only available after the DB lookup inside +//! `identity-driver-sql` (ADR-0022 Invariant 8). These are deferred to a +//! follow-up PR that will refactor the driver to expose the lookup result +//! before password verification. +//! +//! ## Security invariants (ADR-0022 §4) +//! +//! * **Invariant 2 — fail-hard init:** if a bucket is `enabled = true` and +//! `burst_size` or `replenish_rate_per_second` is zero, +//! [`RateLimitState::from_config`] returns +//! [`KeystoneError::RateLimitConfig`] and the application refuses to start. +//! * **Invariant 3 — response uniformity:** the `Duration` returned by +//! [`RateLimitState::check_ip`] is used to build the single `Retry-After` +//! header; no key-identifying information is exposed. +//! * **Invariant 6 — monotonic clock:** all limiters use +//! [`governor::clock::DefaultClock`] which resolves to +//! `MonotonicClock` on `std` targets, preventing NTP backward-shift resets. + +use std::net::IpAddr; +use std::num::NonZeroU32; +use std::sync::Arc; +use std::time::Duration; + +use governor::clock::Clock as _; +use governor::{DefaultKeyedRateLimiter, Quota}; + +use openstack_keystone_config::Config; +use openstack_keystone_core_types::error::KeystoneError; + +/// Shared rate-limiting state, held on [`crate::keystone::Service`]. +/// +/// Each field is an `Option>`: `None` when the bucket is disabled in +/// `keystone.conf`; `Some(…)` when enabled. Disabled buckets cost only the +/// `Option` discriminant — no `governor` state is allocated. +/// +/// The struct is `Clone`-able via `Arc` sharing, not by copying limiter state. +pub struct RateLimitState { + /// Global per-IP limiter — `[rate_limit_global_ip]` in `keystone.conf`. + /// + /// Keyed on IPv4 `/32` (full address) or IPv6 `/64` network prefix. + /// `None` when `enabled = false`. + pub global_ip_limiter: Option>>, + // Future buckets: + // pub user_auth_limiter: Option>>, + // pub domain_limiter: Option>>, + // pub idp_limiter: Option>>, +} + +impl RateLimitState { + /// Build [`RateLimitState`] from the application configuration. + /// + /// # Errors + /// + /// Returns [`KeystoneError::RateLimitConfig`] and aborts startup when a + /// bucket is `enabled = true` but has a `burst_size` or + /// `replenish_rate_per_second` of zero (ADR-0022 Invariant 2). + pub fn from_config(cfg: &Config) -> Result { + let global_ip_limiter = build_limiter(&cfg.rate_limit_global_ip, "rate_limit_global_ip")?; + + Ok(Self { global_ip_limiter }) + } + + /// Check the global per-IP bucket for `ip`. + /// + /// Returns `Ok(())` when: + /// - the bucket is disabled (`enabled = false`), or + /// - the IP is under its quota. + /// + /// Returns `Err(retry_after)` — the `Duration` to place in the + /// `Retry-After` HTTP header — when the bucket is enabled and the IP has + /// exceeded its quota. + /// + /// SPIFFE interfaces (internal / admin) do not populate + /// `ConnectInfo`, so the caller holds an + /// `Option>` and skips this call when it is `None`. + pub fn check_ip(&self, ip: IpAddr) -> Result<(), Duration> { + let Some(ref limiter) = self.global_ip_limiter else { + return Ok(()); // bucket disabled — bypass + }; + let key = rate_limit_key_for_ip(ip); + limiter.check_key(&key).map_err(|not_until| { + // `DefaultKeyedRateLimiter` uses `DefaultClock` (`QuantaClock` on + // std targets). Its `Instant` type is `QuantaInstant`, not + // `std::time::Instant`, so we obtain the current time through the + // limiter's own clock rather than `Instant::now()`. + // `QuantaClock` reads the CPU's TSC — always monotonic; satisfies + // ADR-0022 Invariant 6. + let now = limiter.clock().now(); + let wait = not_until.wait_time_from(now); + // Ensure at least 1 s so a well-behaved client doesn't busy-retry. + wait.max(Duration::from_secs(1)) + }) + } + + /// Evict stale entries from all keyed state stores. + /// + /// Should be called on a ~60 s background timer (ADR-0022 §Consequences). + /// This prevents unbounded memory growth under adversarial unique-key + /// flooding. Callers that no longer appear in the store within the last + /// quota window are pruned. + pub fn retain_recent(&self) { + if let Some(ref limiter) = self.global_ip_limiter { + limiter.retain_recent(); + } + } +} + +// --------------------------------------------------------------------------- +// Internal helpers +// --------------------------------------------------------------------------- + +/// Upper bound permitted for `burst_size` / `replenish_rate_per_second` +/// (ADR-0022 config bounds). Together with the `≥ 1` floor this pins both +/// scalars to `[1, 100_000]`; values outside fail startup, identically to the +/// fail-hard rule for a zero value. +const MAX_RATE_LIMIT_VALUE: u32 = 100_000; + +/// Validate one rate-limit scalar against the ADR-0022 `[1, MAX_RATE_LIMIT_VALUE]` +/// range, returning it as a [`NonZeroU32`] or a fail-hard config error. +fn validated_scalar(value: u32, field: &str, name: &str) -> Result { + NonZeroU32::new(value) + .filter(|v| v.get() <= MAX_RATE_LIMIT_VALUE) + .ok_or_else(|| { + KeystoneError::RateLimitConfig(format!( + "[{name}] {field} must be in [1, {MAX_RATE_LIMIT_VALUE}] when enabled = true" + )) + }) +} + +/// Build a single keyed `governor` rate limiter from a [`RateLimitSection`]. +/// +/// Returns `None` when the section is disabled. Returns +/// `Err(KeystoneError::RateLimitConfig)` when the section is enabled but +/// has invalid parameters (ADR-0022 Invariant 2 / config bounds). +fn build_limiter( + section: &openstack_keystone_config::RateLimitSection, + name: &str, +) -> Result>>, KeystoneError> { + if !section.enabled { + return Ok(None); + } + + let replenish = validated_scalar( + section.replenish_rate_per_second, + "replenish_rate_per_second", + name, + )?; + let burst = validated_scalar(section.burst_size, "burst_size", name)?; + + // Quota: `replenish` cells per second, up to `burst` in a burst. + // Uses `DefaultClock` = `MonotonicClock` (ADR-0022 Invariant 6). + let quota = Quota::per_second(replenish).allow_burst(burst); + let limiter = DefaultKeyedRateLimiter::keyed(quota); + + Ok(Some(Arc::new(limiter))) +} + +/// Derive a rate-limiting key from a client IP address. +/// +/// - **IPv4** addresses are keyed per `/32` (full address), since IPv4 address +/// exhaustion means each address genuinely maps to a distinct node. +/// - **IPv6** addresses are aggregated to their `/64` prefix. RFC 4291 §2.5.1 +/// allocates a `/64` subnet per link, and OS privacy extensions +/// (`RFC 4941`) randomise the lower 64 bits per connection — keying on the +/// full `/128` would give each connection its own independent quota. +fn rate_limit_key_for_ip(ip: IpAddr) -> String { + match ip { + IpAddr::V4(_) => ip.to_string(), + IpAddr::V6(v6) => { + let octets = v6.octets(); + // Zero the host portion (lower 8 bytes) to get the /64 prefix. + let prefix = std::net::Ipv6Addr::from([ + octets[0], octets[1], octets[2], octets[3], octets[4], octets[5], octets[6], + octets[7], 0, 0, 0, 0, 0, 0, 0, 0, + ]); + prefix.to_string() + } + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use std::net::{IpAddr, Ipv4Addr}; + + use openstack_keystone_config::RateLimitSection; + + use super::*; + + fn enabled_section(burst: u32, replenish: u32) -> RateLimitSection { + RateLimitSection { + enabled: true, + burst_size: burst, + replenish_rate_per_second: replenish, + } + } + + fn disabled_section() -> RateLimitSection { + RateLimitSection::default() + } + + // --- key derivation --- + + #[test] + fn ipv4_key_is_full_address() { + let ip = IpAddr::V4(Ipv4Addr::new(203, 0, 113, 1)); + assert_eq!(rate_limit_key_for_ip(ip), "203.0.113.1"); + } + + #[test] + fn ipv6_key_is_slash_64_prefix() { + // Full address: 2001:db8:cafe:1::1 + let ip = IpAddr::V6("2001:db8:cafe:0001:0000:0000:0000:0001".parse().unwrap()); + let key = rate_limit_key_for_ip(ip); + // Lower 64 bits zeroed → 2001:db8:cafe:1:: + assert_eq!(key, "2001:db8:cafe:1::"); + } + + #[test] + fn different_ipv6_privacy_addrs_share_key() { + // Two privacy extension addresses in the same /64 must get the same key. + let a: IpAddr = "2001:db8::dead:beef:1234:5678".parse().unwrap(); + let b: IpAddr = "2001:db8::dead:beef:abcd:ef01".parse().unwrap(); + assert_eq!(rate_limit_key_for_ip(a), rate_limit_key_for_ip(b)); + } + + // --- from_config --- + + #[test] + fn disabled_section_yields_none() { + let result = build_limiter(&disabled_section(), "test"); + assert!(result.unwrap().is_none()); + } + + #[test] + fn enabled_section_yields_some() { + let result = build_limiter(&enabled_section(10, 5), "test"); + assert!(result.unwrap().is_some()); + } + + #[test] + fn zero_replenish_is_fail_hard() { + let section = RateLimitSection { + enabled: true, + burst_size: 10, + replenish_rate_per_second: 0, + }; + let err = build_limiter(§ion, "rate_limit_global_ip").unwrap_err(); + assert!( + matches!(err, KeystoneError::RateLimitConfig(msg) if msg.contains("replenish_rate_per_second")) + ); + } + + #[test] + fn zero_burst_is_fail_hard() { + let section = RateLimitSection { + enabled: true, + burst_size: 0, + replenish_rate_per_second: 5, + }; + let err = build_limiter(§ion, "rate_limit_global_ip").unwrap_err(); + assert!(matches!(err, KeystoneError::RateLimitConfig(msg) if msg.contains("burst_size"))); + } + + #[test] + fn over_max_burst_is_fail_hard() { + let section = enabled_section(MAX_RATE_LIMIT_VALUE + 1, 5); + let err = build_limiter(§ion, "rate_limit_global_ip").unwrap_err(); + assert!(matches!(err, KeystoneError::RateLimitConfig(msg) if msg.contains("burst_size"))); + } + + #[test] + fn over_max_replenish_is_fail_hard() { + let section = enabled_section(10, MAX_RATE_LIMIT_VALUE + 1); + let err = build_limiter(§ion, "rate_limit_global_ip").unwrap_err(); + assert!( + matches!(err, KeystoneError::RateLimitConfig(msg) if msg.contains("replenish_rate_per_second")) + ); + } + + #[test] + fn max_boundary_values_are_accepted() { + let section = enabled_section(MAX_RATE_LIMIT_VALUE, MAX_RATE_LIMIT_VALUE); + assert!( + build_limiter(§ion, "rate_limit_global_ip") + .unwrap() + .is_some() + ); + } + + #[test] + fn disabled_with_zero_values_does_not_fail() { + // Invariant: disabled sections must never cause a startup error. + let section = RateLimitSection { + enabled: false, + burst_size: 0, + replenish_rate_per_second: 0, + }; + let result = build_limiter(§ion, "test"); + assert!(result.unwrap().is_none()); + } + + // --- check_ip --- + + #[test] + fn disabled_limiter_always_allows() { + let state = RateLimitState { + global_ip_limiter: None, + }; + let ip: IpAddr = "198.51.100.1".parse().unwrap(); + for _ in 0..1000 { + assert!(state.check_ip(ip).is_ok()); + } + } + + #[test] + fn enabled_limiter_allows_up_to_burst_then_rejects() { + // burst=3, replenish=1/s — the first 3 calls must pass, the 4th fails. + let state = RateLimitState { + global_ip_limiter: Some(Arc::new(DefaultKeyedRateLimiter::keyed( + Quota::per_second(NonZeroU32::new(1).unwrap()) + .allow_burst(NonZeroU32::new(3).unwrap()), + ))), + }; + let ip: IpAddr = "203.0.113.42".parse().unwrap(); + assert!(state.check_ip(ip).is_ok()); + assert!(state.check_ip(ip).is_ok()); + assert!(state.check_ip(ip).is_ok()); + // 4th request must be rejected. + let err = state.check_ip(ip).unwrap_err(); + // Retry-After should be at least 1 second. + assert!(err >= Duration::from_secs(1)); + } + + #[test] + fn different_ips_have_independent_quotas() { + let state = RateLimitState { + global_ip_limiter: Some(Arc::new(DefaultKeyedRateLimiter::keyed( + Quota::per_second(NonZeroU32::new(1).unwrap()) + .allow_burst(NonZeroU32::new(1).unwrap()), + ))), + }; + let ip_a: IpAddr = "192.0.2.1".parse().unwrap(); + let ip_b: IpAddr = "192.0.2.2".parse().unwrap(); + // Exhaust ip_a's quota. + let _ = state.check_ip(ip_a); + let _ = state.check_ip(ip_a); + // ip_b should still be allowed. + assert!(state.check_ip(ip_b).is_ok()); + } + + #[test] + fn ipv6_privacy_addrs_share_quota() { + // Two /128 addresses in the same /64 must hit the *same* bucket. + let state = RateLimitState { + global_ip_limiter: Some(Arc::new(DefaultKeyedRateLimiter::keyed( + Quota::per_second(NonZeroU32::new(1).unwrap()) + .allow_burst(NonZeroU32::new(1).unwrap()), + ))), + }; + let a: IpAddr = "2001:db8::1".parse().unwrap(); + let b: IpAddr = "2001:db8::2".parse().unwrap(); + // First call on `a` consumes the burst. + let _ = state.check_ip(a); + // Second call on `b` (same /64) should be rejected. + assert!(state.check_ip(b).is_err()); + } +} diff --git a/crates/keystone/src/api/v3/auth/token/create.rs b/crates/keystone/src/api/v3/auth/token/create.rs index ccf118209..e9dce8be0 100644 --- a/crates/keystone/src/api/v3/auth/token/create.rs +++ b/crates/keystone/src/api/v3/auth/token/create.rs @@ -13,9 +13,11 @@ // SPDX-License-Identifier: Apache-2.0 //! Create token (authenticate). +use std::net::SocketAddr; + use axum::{ - Json, - extract::{Query, State}, + Extension, Json, + extract::{ConnectInfo, Query, State}, http::StatusCode, response::{IntoResponse, Response}, }; @@ -53,11 +55,22 @@ use crate::keystone::ServiceState; #[tracing::instrument(name = "api::v3::token::post", level = "debug", skip(state, req))] pub(super) async fn create( CorrelationId(cid): CorrelationId, + // In axum 0.8, `Option` is a valid extractor only when + // `T: OptionalFromRequestParts` — `ConnectInfo` does NOT implement + // that trait, but `Extension>` does (since + // `Extension: OptionalFromRequestParts` for any `T`). + // ConnectInfo is stored in request extensions, so this is equivalent. + // `Option<…>` so the handler compiles on the SPIFFE/internal and admin + // interfaces, which do not populate ConnectInfo (no SocketAddr). + // On those interfaces rate limiting is bypassed (they are + // mutually-authenticated; the public TCP listener is the untrusted front + // door). See ADR-0022 and PR #358 / PR #842. + client_addr: Option>>, Query(query): Query, State(state): State, TracedJson(req): TracedJson, ) -> Result { - let result = create_inner(&state, query, req).await; + let result = create_inner(&state, client_addr, query, req).await; let initiator = result .as_ref() .ok() @@ -75,10 +88,23 @@ pub(super) async fn create( /// HTTP response so the outer handler can build the audit `Initiator`. async fn create_inner( state: &ServiceState, + client_addr: Option>>, query: CreateTokenParameters, req: AuthRequest, ) -> Result<(ValidatedSecurityContext, Response), KeystoneApiError> { req.validate()?; + + // Global per-IP rate-limit check (ADR-0022, Invariant 4). + // Fires BEFORE authenticate_request to avoid consuming CPU on password + // hashing for rejected requests. + if let Some(Extension(ConnectInfo(addr))) = client_addr + && let Err(retry_after) = state.rate_limiters.check_ip(addr.ip()) + { + return Err(KeystoneApiError::TooManyRequests { + retry_after: retry_after.as_secs().max(1), + }); + } + let auth_res = authenticate_request(state, &req).await?; let ctx = SecurityContext::try_from(auth_res)?; let provider_scope: Option = req.auth.scope.clone().map(Into::into); @@ -140,20 +166,22 @@ async fn create_inner( mod tests { use axum::{ body::Body, + extract::ConnectInfo, http::{Request, StatusCode, header}, }; use http_body_util::BodyExt; // for `collect` use sea_orm::DatabaseConnection; use serde_json::json; + use std::net::SocketAddr; use std::sync::Arc; use tower::ServiceExt; // for `call`, `oneshot`, and `ready` use tower_http::trace::TraceLayer; use tracing_test::traced_test; use openstack_keystone_audit::AuditDispatcher; - use openstack_keystone_config::{Config, ConfigManager}; + use openstack_keystone_config::{Config, ConfigManager, RateLimitSection}; use openstack_keystone_core_types::auth::*; - use openstack_keystone_core_types::identity::UserPasswordAuthRequest; + use openstack_keystone_core_types::identity::{IdentityProviderError, UserPasswordAuthRequest}; use openstack_keystone_core_types::resource::{Domain, DomainBuilder, Project}; use openstack_keystone_core_types::token::{ProjectScopePayload, TokenProviderError}; @@ -669,4 +697,251 @@ mod tests { assert_eq!(response.status(), StatusCode::UNAUTHORIZED); } + + // Build a minimal JSON body that passes `req.validate()`. + fn auth_body() -> Vec { + serde_json::to_vec(&json!({ + "auth": { + "identity": { + "methods": ["password"], + "password": { + "user": { + "id": "uid", + "name": "uname", + "domain": { "id": "udid", "name": "udname" }, + "password": "pass" + } + } + } + } + })) + .unwrap() + } + + #[tokio::test] + #[traced_test] + async fn test_rate_limit_returns_429_after_burst_exhausted() { + // burst_size=1: the first request consumes the single burst token and + // passes through to auth; the second must be rejected with 429 before + // authenticate_request is ever called. + let config = Config { + rate_limit_global_ip: RateLimitSection { + enabled: true, + burst_size: 1, + replenish_rate_per_second: 1, + }, + ..Config::default() + }; + + // The first request reaches identity — we return an error so we don't + // need the full auth fixture. The second must never reach identity at + // all (rate limited before authenticate_request). + let mut identity_mock = MockIdentityProvider::default(); + identity_mock + .expect_authenticate_by_password() + .once() + .returning(|_, _| Err(IdentityProviderError::UserNotFound("uid".into()))); + + let provider = Provider::mocked_builder() + .mock_identity(identity_mock) + .build() + .unwrap(); + + let state = Arc::new( + Service::new( + ConfigManager::not_watched(config), + DatabaseConnection::Disconnected, + provider, + Arc::new(MockPolicy::default()), + AuditDispatcher::noop(), + None, + ) + .await + .unwrap(), + ); + + let client_addr: SocketAddr = "203.0.113.1:1234".parse().unwrap(); + + let mut api = openapi_router() + .layer(TraceLayer::new_for_http()) + .with_state(state.clone()); + + // First request — passes the rate limit, fails at auth → not 429. + let mut req1 = Request::builder() + .uri("/") + .method("POST") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(auth_body())) + .unwrap(); + req1.extensions_mut().insert(ConnectInfo(client_addr)); + let resp1 = api.as_service().oneshot(req1).await.unwrap(); + assert_ne!( + resp1.status(), + StatusCode::TOO_MANY_REQUESTS, + "first request must not be rate-limited" + ); + + // Second request from the same IP — burst exhausted → 429 + Retry-After. + let mut req2 = Request::builder() + .uri("/") + .method("POST") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(auth_body())) + .unwrap(); + req2.extensions_mut().insert(ConnectInfo(client_addr)); + let resp2 = api.as_service().oneshot(req2).await.unwrap(); + assert_eq!(resp2.status(), StatusCode::TOO_MANY_REQUESTS); + assert!( + resp2.headers().contains_key(header::RETRY_AFTER), + "429 response must carry Retry-After header" + ); + } + + #[tokio::test] + #[traced_test] + async fn test_rate_limit_does_not_apply_without_connect_info() { + // When there is no ConnectInfo in extensions (SPIFFE/internal interface), + // requests must never be rate-limited regardless of the configured quota. + let config = Config { + rate_limit_global_ip: RateLimitSection { + enabled: true, + burst_size: 1, + replenish_rate_per_second: 1, + }, + ..Config::default() + }; + + let mut identity_mock = MockIdentityProvider::default(); + // Two calls expected — both reach identity, neither is rate-limited. + identity_mock + .expect_authenticate_by_password() + .times(2) + .returning(|_, _| Err(IdentityProviderError::UserNotFound("uid".into()))); + + let provider = Provider::mocked_builder() + .mock_identity(identity_mock) + .build() + .unwrap(); + + let state = Arc::new( + Service::new( + ConfigManager::not_watched(config), + DatabaseConnection::Disconnected, + provider, + Arc::new(MockPolicy::default()), + AuditDispatcher::noop(), + None, + ) + .await + .unwrap(), + ); + + let mut api = openapi_router() + .layer(TraceLayer::new_for_http()) + .with_state(state.clone()); + + // Neither request carries ConnectInfo — both must reach identity. + for _ in 0..2 { + let resp = api + .as_service() + .oneshot( + Request::builder() + .uri("/") + .method("POST") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(auth_body())) + .unwrap(), + ) + .await + .unwrap(); + assert_ne!(resp.status(), StatusCode::TOO_MANY_REQUESTS); + } + } + + /// End-to-end path: drive the real `create` route through the exact + /// make-service the public listener uses + /// (`into_make_service_with_connect_info::`) with a fixed peer + /// address, so the `ConnectInfo` extension is populated by axum + /// itself — not injected by the test. This proves the whole chain wires up: + /// TCP peer → `ConnectInfo` extension → `Option>>` + /// extractor → `check_ip` → 429 + `Retry-After`. Confirms the `401 → 429` + /// flip the manual `curl` loop in the PR test plan would show, without a + /// live database or socket (`Connected for SocketAddr` drives + /// the make-service with a synthetic peer). + #[tokio::test] + #[traced_test] + async fn test_rate_limit_429_over_connect_info_make_service() { + use axum::ServiceExt as AxumServiceExt; + + let config = Config { + rate_limit_global_ip: RateLimitSection { + enabled: true, + burst_size: 1, + replenish_rate_per_second: 1, + }, + ..Config::default() + }; + + // Identity returns an auth failure so the first (non-limited) request + // resolves without the full token fixture; it must be called exactly + // once — the second request is rejected before reaching identity. + let mut identity_mock = MockIdentityProvider::default(); + identity_mock + .expect_authenticate_by_password() + .once() + .returning(|_, _| Err(IdentityProviderError::UserNotFound("uid".into()))); + + let provider = Provider::mocked_builder() + .mock_identity(identity_mock) + .build() + .unwrap(); + + let state = Arc::new( + Service::new( + ConfigManager::not_watched(config), + DatabaseConnection::Disconnected, + provider, + Arc::new(MockPolicy::default()), + AuditDispatcher::noop(), + None, + ) + .await + .unwrap(), + ); + + // Same make-service path as `spawn_public_listener`; explicit request + // type satisfies inference (E0284), as in the binary. + let (router, _) = openapi_router().split_for_parts(); + let app = router.with_state(state.clone()); + let make = + AxumServiceExt::>::into_make_service_with_connect_info::(app); + let peer: SocketAddr = "203.0.113.7:4444".parse().unwrap(); + + let post = || { + Request::builder() + .uri("/") + .method("POST") + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from(auth_body())) + .unwrap() + }; + + // First request from the peer: passes rate limit, fails at auth → not 429. + let svc1 = make.clone().oneshot(peer).await.unwrap(); + let resp1 = svc1.oneshot(post()).await.unwrap(); + assert_ne!( + resp1.status(), + StatusCode::TOO_MANY_REQUESTS, + "first request from a fresh IP must not be rate-limited" + ); + + // Second request from the same peer: burst spent → 429 + Retry-After. + let svc2 = make.clone().oneshot(peer).await.unwrap(); + let resp2 = svc2.oneshot(post()).await.unwrap(); + assert_eq!(resp2.status(), StatusCode::TOO_MANY_REQUESTS); + assert!( + resp2.headers().contains_key(header::RETRY_AFTER), + "429 response must carry a Retry-After header" + ); + } } diff --git a/crates/keystone/src/audit.rs b/crates/keystone/src/audit.rs index 96cfcc059..5b2f71d83 100644 --- a/crates/keystone/src/audit.rs +++ b/crates/keystone/src/audit.rs @@ -92,7 +92,7 @@ pub fn error_variant_name(error: &KeystoneApiError) -> String { KeystoneApiError::Base64Decode(_) => "BadRequest".to_string(), KeystoneApiError::Serde { .. } => "BadRequest".to_string(), KeystoneApiError::Other(_) => "InternalServerError".to_string(), - KeystoneApiError::TooManyRequests => "TooManyRequests".to_string(), + KeystoneApiError::TooManyRequests { .. } => "TooManyRequests".to_string(), } } diff --git a/crates/keystone/src/bin/keystone.rs b/crates/keystone/src/bin/keystone.rs index 590711191..09ece876e 100644 --- a/crates/keystone/src/bin/keystone.rs +++ b/crates/keystone/src/bin/keystone.rs @@ -254,7 +254,10 @@ async fn main() -> Result<(), Report> { .await?, ); - spawn(cleanup(cloned_token, shared_state.clone())); + spawn(cleanup(cloned_token.clone(), shared_state.clone())); + // Evict stale entries from rate-limit keyed state stores every 60 s + // (ADR-0022 §Consequences: memory overhead and store eviction). + spawn(rate_limit_eviction(cloned_token, shared_state.clone())); // API Key (SCIM ingress) janitor: proactive inactivity disablement and // tombstone purge (ADR 0021 §6.F). Runs on every node; gated to actually @@ -986,6 +989,31 @@ async fn reset_dummy_hash_on_reload(cancel: CancellationToken, state: ServiceSta } } +/// Periodically evict stale entries from rate-limit keyed state stores. +/// +/// Runs every 60 seconds, mirroring the [`cleanup`] task pattern. Calls +/// [`RateLimitState::retain_recent`] on all active buckets so that keys that +/// have not been seen within the last quota window are removed, preventing +/// unbounded memory growth under adversarial unique-key flooding (ADR-0022 +/// §Consequences: memory overhead and store eviction). +async fn rate_limit_eviction(cancel: CancellationToken, state: ServiceState) { + let mut interval = time::interval(Duration::from_secs(60)); + interval.tick().await; + info!("Start the rate-limit eviction task"); + loop { + tokio::select! { + _ = interval.tick() => { + trace!("rate-limit eviction tick"); + state.rate_limiters.retain_recent(); + }, + () = cancel.cancelled() => { + info!("Cancellation requested. Stopping rate-limit eviction task."); + break; + } + } + } +} + /// Install shutdown and interrupt signal handler. async fn shutdown_signal(state: ServiceState) { let ctrl_c = async { diff --git a/tools/keystone.conf b/tools/keystone.conf index eddf20727..dc4f7f461 100644 --- a/tools/keystone.conf +++ b/tools/keystone.conf @@ -9,3 +9,15 @@ connection = postgresql://keystone:password@db/keystone [api_policy] opa_base_url = http://opa:8181 + +# Global per-IP rate limiting (ADR-0022). +# When enabled, each unique client IPv4 address (/32) or IPv6 /64 prefix +# is allowed at most `burst_size` requests in a burst, replenishing at +# `replenish_rate_per_second` cells/s thereafter. +# The application refuses to start if `enabled = true` with either value +# set to 0. Disabled by default. +# +# [rate_limit_global_ip] +# enabled = false +# burst_size = 100 +# replenish_rate_per_second = 10