From 97edb2e5cf7ac792517f48c9cc1240412ecf3706 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 17 Jul 2026 19:01:18 +0800 Subject: [PATCH] feat(api): add opt-in per-bucket dimension to the S3 API rate limiter (#4949) Follow-up to #4895 (backlog#1191 deferred sub-item). The client-IP dimension gives per-client fairness; this adds a collective per-bucket budget so one hot bucket cannot monopolize the server regardless of how many client IPs the traffic is spread across. - Generalize RateLimiter over its key type with a Borrow-based check() so &str lookups against String bucket keys stay allocation-free on the hit path; client-IP call sites are unchanged in behavior. - RateLimitLayer now carries optional client and bucket limiters; a request must pass every configured dimension, and rejections report which one tripped via a new 'dimension' metric label. - Bucket extraction mirrors s3s host routing: virtual-hosted-style resolves the Host/authority prefix against the same expanded domain set (with port variants) the s3s router uses; otherwise the first path segment. Admin and table-catalog namespaces are never buckets. - New env vars RUSTFS_API_RATE_LIMIT_BUCKET_RPM/_BURST (default 0 = dimension off) under the existing enable switch; bucket-only configurations (client RPM 0) are supported. - Bucket names are attacker-chosen, so the bounded-shards design (100k keys, lossless idle sweeps, most-idle eviction) is the memory defense; a test floods 10k random names and asserts the cap holds. - Unit tests for extraction, shared bucket budgets across IPs, both- dimensions interaction, and the extended env matrix; e2e test proves bucket-only throttling on the real server with an unrelated bucket unaffected. --- crates/config/src/constants/api.rs | 23 + crates/e2e_test/src/api_rate_limit_test.rs | 46 ++ rustfs/src/admin/console.rs | 12 +- rustfs/src/server/http.rs | 57 ++- rustfs/src/server/rate_limit.rs | 542 ++++++++++++++++----- 5 files changed, 536 insertions(+), 144 deletions(-) diff --git a/crates/config/src/constants/api.rs b/crates/config/src/constants/api.rs index 86b687c9c..7ba69efdd 100644 --- a/crates/config/src/constants/api.rs +++ b/crates/config/src/constants/api.rs @@ -50,3 +50,26 @@ pub const ENV_API_RATE_LIMIT_BURST: &str = "RUSTFS_API_RATE_LIMIT_BURST"; /// Default for `RUSTFS_API_RATE_LIMIT_BURST` (`0` = same as RPM). pub const DEFAULT_API_RATE_LIMIT_BURST: u32 = 0; + +/// Sustained S3 API request budget per addressed bucket, in requests per +/// minute — a collective ceiling shared by all clients of that bucket. +/// +/// Complements the per-client-IP dimension: it protects the server from one +/// hot bucket regardless of how many client IPs the traffic comes from. `0` +/// disables the bucket dimension. Requires `RUSTFS_API_RATE_LIMIT_ENABLE`. +/// Environment variable: RUSTFS_API_RATE_LIMIT_BUCKET_RPM +/// Example: RUSTFS_API_RATE_LIMIT_BUCKET_RPM=60000 +pub const ENV_API_RATE_LIMIT_BUCKET_RPM: &str = "RUSTFS_API_RATE_LIMIT_BUCKET_RPM"; + +/// Default for `RUSTFS_API_RATE_LIMIT_BUCKET_RPM` (`0` = dimension disabled). +pub const DEFAULT_API_RATE_LIMIT_BUCKET_RPM: u32 = 0; + +/// Burst capacity per bucket (maximum tokens in the bucket-dimension bucket). +/// +/// `0` means "same as `RUSTFS_API_RATE_LIMIT_BUCKET_RPM`". +/// Environment variable: RUSTFS_API_RATE_LIMIT_BUCKET_BURST +/// Example: RUSTFS_API_RATE_LIMIT_BUCKET_BURST=2000 +pub const ENV_API_RATE_LIMIT_BUCKET_BURST: &str = "RUSTFS_API_RATE_LIMIT_BUCKET_BURST"; + +/// Default for `RUSTFS_API_RATE_LIMIT_BUCKET_BURST` (`0` = same as bucket RPM). +pub const DEFAULT_API_RATE_LIMIT_BUCKET_BURST: u32 = 0; diff --git a/crates/e2e_test/src/api_rate_limit_test.rs b/crates/e2e_test/src/api_rate_limit_test.rs index 7da385209..28f80517d 100644 --- a/crates/e2e_test/src/api_rate_limit_test.rs +++ b/crates/e2e_test/src/api_rate_limit_test.rs @@ -86,6 +86,52 @@ async fn api_rate_limit_enforces_429_with_retry_after_when_enabled() -> TestResu Ok(()) } +#[tokio::test] +#[serial] +async fn api_rate_limit_bucket_dimension_throttles_per_bucket() -> TestResult { + init_logging(); + let mut env = RustFSTestEnvironment::new().await?; + // Bucket dimension only: the readiness-poll ListBuckets calls hit "/" + // (no bucket) and therefore do not consume any budget. + env.start_rustfs_server_with_env( + vec![], + &[ + ("RUSTFS_API_RATE_LIMIT_ENABLE", "true"), + ("RUSTFS_API_RATE_LIMIT_BUCKET_RPM", "60"), + ("RUSTFS_API_RATE_LIMIT_BUCKET_BURST", "5"), + ], + ) + .await?; + + let client = local_http_client(); + + // Unauthenticated GETs are still counted arrivals (403, not 429, while + // within budget); the sixth rapid hit on the same bucket must throttle. + let mut throttled = false; + for i in 0..6 { + let response = client.get(format!("{}/hot-bucket/object-{i}", env.url)).send().await?; + if response.status() == reqwest::StatusCode::TOO_MANY_REQUESTS { + throttled = true; + assert!( + response.headers().contains_key(reqwest::header::RETRY_AFTER), + "bucket-dimension 429 must carry Retry-After" + ); + break; + } + } + assert!(throttled, "6 rapid requests against burst 5 must trip the bucket dimension"); + + // A different bucket has its own budget. + let other = client.get(format!("{}/cold-bucket/object", env.url)).send().await?; + assert_ne!( + other.status(), + reqwest::StatusCode::TOO_MANY_REQUESTS, + "an unrelated bucket must not be throttled" + ); + + Ok(()) +} + #[tokio::test] #[serial] async fn api_rate_limit_stays_inert_by_default() -> TestResult { diff --git a/rustfs/src/admin/console.rs b/rustfs/src/admin/console.rs index 5a7a4a1d2..b86c61431 100644 --- a/rustfs/src/admin/console.rs +++ b/rustfs/src/admin/console.rs @@ -17,8 +17,9 @@ use crate::admin::storage_api::access::RequestContext; use crate::license::has_valid_license; use crate::server::has_path_prefix; use crate::server::rate_limit::{ - LABEL_RATE_LIMIT_SCOPE, METRIC_HTTP_SERVER_REQUESTS_RATE_LIMITED_TOTAL, RATE_LIMIT_SCOPE_CONSOLE, RateLimitDecision, - RateLimitQuota, RateLimiter, apply_throttle_headers, client_ip, + LABEL_RATE_LIMIT_DIMENSION, LABEL_RATE_LIMIT_SCOPE, METRIC_HTTP_SERVER_REQUESTS_RATE_LIMITED_TOTAL, + RATE_LIMIT_DIMENSION_CLIENT_IP, RATE_LIMIT_SCOPE_CONSOLE, RateLimitDecision, RateLimitQuota, RateLimiter, + apply_throttle_headers, client_ip, }; use crate::server::{ CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, HeaderMapCarrier, HealthProbe, LICENSE, RUSTFS_ADMIN_PREFIX, @@ -581,15 +582,16 @@ fn setup_console_middleware_stack( // without one fail open rather than trusting spoofable headers. if rate_limit_enable { if let Some(quota) = RateLimitQuota::per_minute(rate_limit_rpm, 0) { - let limiter = Arc::new(RateLimiter::new(quota)); + let limiter: Arc = Arc::new(RateLimiter::new(quota)); app = app.layer(middleware::from_fn(move |req: Request, next: middleware::Next| { let limiter = limiter.clone(); async move { - match client_ip(&req).map(|ip| limiter.check(ip)) { + match client_ip(&req).map(|ip| limiter.check(&ip)) { Some(RateLimitDecision::Limited(throttle)) => { metrics::counter!( METRIC_HTTP_SERVER_REQUESTS_RATE_LIMITED_TOTAL, - LABEL_RATE_LIMIT_SCOPE => RATE_LIMIT_SCOPE_CONSOLE + LABEL_RATE_LIMIT_SCOPE => RATE_LIMIT_SCOPE_CONSOLE, + LABEL_RATE_LIMIT_DIMENSION => RATE_LIMIT_DIMENSION_CLIENT_IP ) .increment(1); debug!(retry_after_secs = throttle.retry_after_secs, "Console request rejected by rate limit"); diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 7f30ade17..2b873fd51 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -609,6 +609,28 @@ pub async fn start_http_server( ); } + // Expanded virtual-hosted-style domain set (with port variants); shared by + // the s3s host router below and the rate limit layer's bucket extraction. + let host_domain_sets = if !config.server_domains.is_empty() && !config.console_enable { + MultiDomain::new(&config.server_domains).map_err(Error::other)?; // validate domains + + // add the default port number to the given server domains + let mut domain_sets = std::collections::HashSet::new(); + for domain in &config.server_domains { + domain_sets.insert(domain.to_string()); + if let Some((host, _)) = domain.split_once(':') { + domain_sets.insert(format!("{host}:{local_port}")); + } else { + domain_sets.insert(format!("{domain}:{local_port}")); + } + } + + Some(domain_sets) + } else { + None + }; + let rate_limit_vh_domains: Vec = host_domain_sets.iter().flatten().cloned().collect(); + // Setup S3 service // This project uses the S3S library to implement S3 services let s3_service = { @@ -620,24 +642,6 @@ pub async fn start_http_server( let access_key = config.access_key.clone(); let secret_key = config.secret_key.clone(); - let host_domain_sets = if !config.server_domains.is_empty() && !config.console_enable { - MultiDomain::new(&config.server_domains).map_err(Error::other)?; // validate domains - - // add the default port number to the given server domains - let mut domain_sets = std::collections::HashSet::new(); - for domain in &config.server_domains { - domain_sets.insert(domain.to_string()); - if let Some((host, _)) = domain.split_once(':') { - domain_sets.insert(format!("{host}:{local_port}")); - } else { - domain_sets.insert(format!("{domain}:{local_port}")); - } - } - - Some(domain_sets) - } else { - None - }; let metadata_route_host = host_domain_sets .as_ref() .map(MultiDomain::new) @@ -702,20 +706,23 @@ pub async fn start_http_server( ); } - // Per-client S3 API rate limiting (backlog#1191). Built once so every - // connection's stack shares the same limiter state; `None` (the default) - // leaves the request path unchanged. - let api_rate_limit_layer = api_rate_limit_layer_from_env(); + // Per-client / per-bucket S3 API rate limiting (backlog#1191). Built once + // so every connection's stack shares the same limiter state; `None` (the + // default) leaves the request path unchanged. + let api_rate_limit_layer = api_rate_limit_layer_from_env(rate_limit_vh_domains); match &api_rate_limit_layer { Some(layer) => { - let quota = layer.quota(); + let client_quota = layer.client_quota(); + let bucket_quota = layer.bucket_quota(); info!( event = EVENT_API_RATE_LIMIT_STATE, component = LOG_COMPONENT_SERVER, subsystem = LOG_SUBSYSTEM_HTTP, state = "enabled", - requests_per_minute = quota.requests_per_minute, - burst = quota.burst, + client_rpm = client_quota.map(|q| q.requests_per_minute).unwrap_or(0), + client_burst = client_quota.map(|q| q.burst).unwrap_or(0), + bucket_rpm = bucket_quota.map(|q| q.requests_per_minute).unwrap_or(0), + bucket_burst = bucket_quota.map(|q| q.burst).unwrap_or(0), "API rate limit state changed" ); } diff --git a/rustfs/src/server/rate_limit.rs b/rustfs/src/server/rate_limit.rs index 4790e24af..bdd11e9f2 100644 --- a/rustfs/src/server/rate_limit.rs +++ b/rustfs/src/server/rate_limit.rs @@ -14,35 +14,48 @@ //! Per-client request rate limiting for the S3 API (backlog#1191). //! -//! A token-bucket limiter keyed by client IP. Over-limit requests are rejected -//! with `429 Too Many Requests`, a `Retry-After` header, and the -//! `x-ratelimit-*` headers already used by the Swift protocol error mapping. +//! A token-bucket limiter with two optional dimensions: +//! - **client IP** (`RUSTFS_API_RATE_LIMIT_RPM`/`_BURST`): per-client fairness +//! and abuse protection; +//! - **bucket** (`RUSTFS_API_RATE_LIMIT_BUCKET_RPM`/`_BURST`): a collective +//! budget per addressed bucket, protecting the server from one hot bucket +//! regardless of how many client IPs the traffic comes from. +//! +//! Over-limit requests are rejected with `429 Too Many Requests`, a +//! `Retry-After` header, and the `x-ratelimit-*` headers already used by the +//! Swift protocol error mapping. //! //! Scope and invariants: //! - **Opt-in, default off.** [`api_rate_limit_layer_from_env`] returns `None` -//! unless `RUSTFS_API_RATE_LIMIT_ENABLE=true` and `RUSTFS_API_RATE_LIMIT_RPM > 0`; -//! the layer is then absent from the service stack and the request path is -//! unchanged. +//! unless `RUSTFS_API_RATE_LIMIT_ENABLE=true` and at least one dimension has +//! a non-zero RPM; the layer is then absent from the service stack and the +//! request path is unchanged. //! - **Client identity is never taken from request headers.** The key is the //! validated [`ClientInfo::real_ip`] inserted by the trusted-proxy layer //! (which only honors `X-Forwarded-For` from configured proxies) or, absent //! that, the socket peer address ([`RemoteAddr`]). A raw `X-Forwarded-For` //! header can therefore not be used to escape into an attacker-chosen bucket. +//! - **Bucket extraction mirrors s3s host routing.** Virtual-hosted-style +//! requests resolve the bucket from the Host/authority against the same +//! expanded server-domain set the s3s router uses; everything else takes the +//! first path segment. Admin and table-catalog namespaces are not buckets. //! - **Infra traffic is exempt** ([`is_rate_limit_exempt_path`]): health and //! profiling probes, internode RPC/gRPC, and the console (which has its own //! limiter, sharing this module's [`RateLimiter`] core). //! - This is client-facing abuse protection, not internal backpressure — it is //! unrelated to `workload_admission`, which schedules already-admitted work. //! -//! Memory is bounded: buckets live in [`SHARD_COUNT`] independently locked -//! shards, each capped and periodically swept. A bucket that has been idle -//! long enough to have refilled completely is indistinguishable from a fresh -//! one, so eviction never makes the limiter more permissive than a restart. +//! Memory is bounded per dimension: buckets live in [`SHARD_COUNT`] +//! independently locked shards, each capped and periodically swept. A bucket +//! that has been idle long enough to have refilled completely is +//! indistinguishable from a fresh one, so eviction never makes the limiter +//! more permissive than a restart. The cap matters doubly for the bucket +//! dimension, whose key space (bucket names) is attacker-chosen. use crate::server::{ CONSOLE_PREFIX, FAVICON_PATH, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_HEALTH_CLUSTER_PATH, MINIO_HEALTH_CLUSTER_READ_PATH, MINIO_HEALTH_LIVE_PATH, MINIO_HEALTH_READY_PATH, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, - RPC_PREFIX, RemoteAddr, TONIC_PREFIX, has_path_prefix, + RPC_PREFIX, RemoteAddr, TONIC_PREFIX, has_path_prefix, is_admin_path, is_table_catalog_path, }; use bytes::Bytes; use futures::future::{Either, Ready, ready}; @@ -50,8 +63,9 @@ use http::{HeaderMap, HeaderValue, Request, Response, StatusCode}; use http_body_util::{BodyExt, Full}; use metrics::counter; use rustfs_trusted_proxies::ClientInfo; +use std::borrow::Borrow; use std::collections::HashMap; -use std::hash::{BuildHasher, RandomState}; +use std::hash::{BuildHasher, Hash, RandomState}; use std::net::IpAddr; use std::sync::{Arc, Mutex, PoisonError}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -66,6 +80,9 @@ pub(crate) const METRIC_HTTP_SERVER_REQUESTS_RATE_LIMITED_TOTAL: &str = "rustfs_ pub(crate) const LABEL_RATE_LIMIT_SCOPE: &str = "scope"; pub(crate) const RATE_LIMIT_SCOPE_S3_API: &str = "s3_api"; pub(crate) const RATE_LIMIT_SCOPE_CONSOLE: &str = "console"; +pub(crate) const LABEL_RATE_LIMIT_DIMENSION: &str = "dimension"; +pub(crate) const RATE_LIMIT_DIMENSION_CLIENT_IP: &str = "client_ip"; +const RATE_LIMIT_DIMENSION_BUCKET: &str = "bucket"; // Header names shared with the Swift error mapping (crates/protocols swift/errors.rs). const X_RATE_LIMIT_LIMIT: &str = "x-ratelimit-limit"; @@ -75,14 +92,15 @@ const X_REQUEST_ID: &str = "x-request-id"; /// Shards bound lock contention: each request locks 1/32 of the key space. const SHARD_COUNT: usize = 32; -/// Upper bound on tracked client IPs across all shards (~100 bytes each, so -/// worst case a few MiB). Real distinct-IP cardinality is bounded by actual -/// TCP peers (or validated proxy clients), never by spoofable headers. +/// Upper bound on tracked keys per limiter instance (~100 bytes each, so +/// worst case a few MiB). For the client dimension real cardinality is +/// bounded by actual TCP peers (or validated proxy clients); for the bucket +/// dimension the cap is the defense, since names are attacker-chosen. const MAX_TRACKED_CLIENTS: usize = 100_000; /// Per-shard sweep cadence for dropping refilled-idle buckets. const CLEANUP_INTERVAL: Duration = Duration::from_secs(30); -/// Sustained-plus-burst request budget for one client. +/// Sustained-plus-burst request budget for one key. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct RateLimitQuota { /// Sustained budget in requests per minute. @@ -120,8 +138,8 @@ pub enum RateLimitDecision { Limited(ThrottleInfo), } -/// One client's bucket. Capacity and refill rate are limiter-wide (single -/// quota for all clients), so they are not duplicated per entry. +/// One key's bucket. Capacity and refill rate are limiter-wide (single quota +/// for all keys), so they are not duplicated per entry. #[derive(Debug)] struct TokenBucket { tokens: f64, @@ -163,16 +181,17 @@ impl TokenBucket { } #[derive(Debug)] -struct Shard { - clients: HashMap, +struct Shard { + clients: HashMap, next_cleanup: Instant, } -/// Sharded per-IP token-bucket limiter. Shared by the S3 API tower layer and -/// the console middleware (each with its own instance and quota). +/// Sharded token-bucket limiter over an arbitrary key dimension (client IP, +/// bucket name, ...). Shared by the S3 API tower layer and the console +/// middleware, each instance with its own quota. #[derive(Debug)] -pub struct RateLimiter { - shards: Vec>, +pub struct RateLimiter { + shards: Vec>>, shard_hasher: RandomState, quota: RateLimitQuota, capacity: f64, @@ -180,7 +199,7 @@ pub struct RateLimiter { max_clients_per_shard: usize, } -impl RateLimiter { +impl RateLimiter { pub fn new(quota: RateLimitQuota) -> Self { Self::with_limits(quota, SHARD_COUNT, MAX_TRACKED_CLIENTS) } @@ -208,13 +227,23 @@ impl RateLimiter { self.quota } - /// Consume one token for `ip`, creating its bucket on first sight. - pub fn check(&self, ip: IpAddr) -> RateLimitDecision { - self.check_at(ip, Instant::now()) + /// Consume one token for `key`, creating its bucket on first sight. The + /// borrowed-key form keeps the hit path allocation-free for owned key + /// types (e.g. `&str` lookups against `String` keys). + pub fn check(&self, key: &Q) -> RateLimitDecision + where + K: Borrow, + Q: Hash + Eq + ToOwned + ?Sized, + { + self.check_at(key, Instant::now()) } - fn check_at(&self, ip: IpAddr, now: Instant) -> RateLimitDecision { - let shard = &self.shards[self.shard_index(ip)]; + fn check_at(&self, key: &Q, now: Instant) -> RateLimitDecision + where + K: Borrow, + Q: Hash + Eq + ToOwned + ?Sized, + { + let shard = &self.shards[self.shard_index(key)]; let mut shard = shard.lock().unwrap_or_else(PoisonError::into_inner); if now >= shard.next_cleanup { @@ -225,33 +254,35 @@ impl RateLimiter { shard.next_cleanup = now + CLEANUP_INTERVAL; } - if let Some(bucket) = shard.clients.get_mut(&ip) { + if let Some(bucket) = shard.clients.get_mut(key) { return bucket.try_consume(self.capacity, self.refill_per_sec, now); } if shard.clients.len() >= self.max_clients_per_shard { // At capacity: reclaim the most idle entry (closest to full, so the // least information is lost). O(shard len) but only under a - // distinct-IP flood, where correctness beats micro-latency. + // distinct-key flood, where correctness beats micro-latency. if let Some(stalest) = shard .clients .iter() .min_by_key(|(_, bucket)| bucket.last_refill) - .map(|(ip, _)| *ip) + .map(|(key, _)| key.clone()) { - shard.clients.remove(&stalest); + // Turbofish: without it inference unifies the lookup type with + // this method's `Q` instead of `K`. + shard.clients.remove::(&stalest); } } shard .clients - .entry(ip) + .entry(key.to_owned()) .or_insert_with(|| TokenBucket::full(self.capacity, now)) .try_consume(self.capacity, self.refill_per_sec, now) } - fn shard_index(&self, ip: IpAddr) -> usize { - (self.shard_hasher.hash_one(ip) as usize) % self.shards.len() + fn shard_index(&self, key: &Q) -> usize { + (self.shard_hasher.hash_one(key) as usize) % self.shards.len() } #[cfg(test)] @@ -276,6 +307,62 @@ pub(crate) fn client_ip(req: &Request) -> Option { req.extensions().get::().map(|remote| remote.0.ip()) } +/// Resolve the host a request addresses, for virtual-hosted-style bucket +/// extraction (HTTP/2 carries it in the URI authority, HTTP/1.1 in `Host`). +fn request_host(req: &Request) -> Option<&str> { + if let Some(authority) = req.uri().authority() { + return Some(authority.as_str()); + } + req.headers().get(http::header::HOST).and_then(|value| value.to_str().ok()) +} + +/// `bucket.domain` → `bucket` when `host` is a subdomain of `domain` +/// (ASCII-case-insensitive, port-inclusive — the configured domain set +/// already carries port variants). +fn strip_vh_prefix<'a>(host: &'a str, domain: &str) -> Option<&'a str> { + let (host_len, domain_len) = (host.len(), domain.len()); + if host_len > domain_len + 1 + && host.as_bytes()[host_len - domain_len - 1] == b'.' + && host[host_len - domain_len..].eq_ignore_ascii_case(domain) + { + Some(&host[..host_len - domain_len - 1]) + } else { + None + } +} + +/// Extract the bucket a request addresses, if any. +/// +/// Mirrors s3s host routing: on a configured virtual-hosted-style domain the +/// bucket is the host prefix (an exact domain match means path-style on that +/// domain); otherwise the first path segment. Admin and table-catalog +/// namespaces are not buckets. Best-effort by design — a request this cannot +/// classify simply skips the bucket dimension (the client dimension still +/// applies). +fn request_bucket<'a, B>(req: &'a Request, vh_domains: &[String]) -> Option<&'a str> { + let path = req.uri().path(); + if is_admin_path(path) || is_table_catalog_path(path) { + return None; + } + + if !vh_domains.is_empty() + && let Some(host) = request_host(req) + { + for domain in vh_domains { + if host.eq_ignore_ascii_case(domain) { + // Path-style request on the API domain itself. + break; + } + if let Some(bucket) = strip_vh_prefix(host, domain) { + return Some(bucket); + } + } + } + + let bucket = path.trim_start_matches('/').split('/').next().unwrap_or(""); + (!bucket.is_empty()).then_some(bucket) +} + /// Paths exempt from S3 API rate limiting. /// /// - Health/profiling probes: kubelet and load-balancer checks must never be @@ -357,45 +444,70 @@ fn s3_too_many_requests_response(request_id: Option<&HeaderValue>, limit_rpm: u3 /// Build the S3 API rate limit layer from `RUSTFS_API_RATE_LIMIT_*`. /// -/// Returns `None` (no layer in the stack, zero request-path change) unless -/// explicitly enabled with a non-zero RPM. -pub fn api_rate_limit_layer_from_env() -> Option { +/// `vh_domains` is the expanded virtual-hosted-style domain set (with port +/// variants) the s3s host router uses; pass an empty vec when virtual-hosted +/// routing is not configured. Returns `None` (no layer in the stack, zero +/// request-path change) unless explicitly enabled with a non-zero RPM on at +/// least one dimension. +pub fn api_rate_limit_layer_from_env(vh_domains: Vec) -> Option { if !rustfs_utils::get_env_bool(rustfs_config::ENV_API_RATE_LIMIT_ENABLE, rustfs_config::DEFAULT_API_RATE_LIMIT_ENABLE) { return None; } - let rpm = rustfs_utils::get_env_u32(rustfs_config::ENV_API_RATE_LIMIT_RPM, rustfs_config::DEFAULT_API_RATE_LIMIT_RPM); - let burst = rustfs_utils::get_env_u32(rustfs_config::ENV_API_RATE_LIMIT_BURST, rustfs_config::DEFAULT_API_RATE_LIMIT_BURST); - let Some(quota) = RateLimitQuota::per_minute(rpm, burst) else { + let client_quota = RateLimitQuota::per_minute( + rustfs_utils::get_env_u32(rustfs_config::ENV_API_RATE_LIMIT_RPM, rustfs_config::DEFAULT_API_RATE_LIMIT_RPM), + rustfs_utils::get_env_u32(rustfs_config::ENV_API_RATE_LIMIT_BURST, rustfs_config::DEFAULT_API_RATE_LIMIT_BURST), + ); + let bucket_quota = RateLimitQuota::per_minute( + rustfs_utils::get_env_u32( + rustfs_config::ENV_API_RATE_LIMIT_BUCKET_RPM, + rustfs_config::DEFAULT_API_RATE_LIMIT_BUCKET_RPM, + ), + rustfs_utils::get_env_u32( + rustfs_config::ENV_API_RATE_LIMIT_BUCKET_BURST, + rustfs_config::DEFAULT_API_RATE_LIMIT_BUCKET_BURST, + ), + ); + if client_quota.is_none() && bucket_quota.is_none() { warn!( component = LOG_COMPONENT_SERVER, subsystem = LOG_SUBSYSTEM_RATE_LIMIT, - "{} is enabled but {} is 0; API rate limiting stays inactive", + "{} is enabled but {} and {} are both 0; API rate limiting stays inactive", rustfs_config::ENV_API_RATE_LIMIT_ENABLE, - rustfs_config::ENV_API_RATE_LIMIT_RPM + rustfs_config::ENV_API_RATE_LIMIT_RPM, + rustfs_config::ENV_API_RATE_LIMIT_BUCKET_RPM ); return None; - }; - Some(RateLimitLayer::new(quota)) + } + Some(RateLimitLayer::new(client_quota, bucket_quota, vh_domains)) } -/// Tower layer enforcing the per-IP quota on the external S3 service stack. +/// Tower layer enforcing the per-client-IP and/or per-bucket quotas on the +/// external S3 service stack. /// -/// All clones share one [`RateLimiter`], so per-connection stack construction -/// keeps a single global view of client budgets. +/// All clones share the same [`RateLimiter`] instances, so per-connection +/// stack construction keeps a single global view of budgets. #[derive(Debug, Clone)] pub struct RateLimitLayer { - limiter: Arc, + client_limiter: Option>>, + bucket_limiter: Option>>, + vh_domains: Arc<[String]>, } impl RateLimitLayer { - pub fn new(quota: RateLimitQuota) -> Self { + pub fn new(client_quota: Option, bucket_quota: Option, vh_domains: Vec) -> Self { Self { - limiter: Arc::new(RateLimiter::new(quota)), + client_limiter: client_quota.map(|quota| Arc::new(RateLimiter::new(quota))), + bucket_limiter: bucket_quota.map(|quota| Arc::new(RateLimiter::new(quota))), + vh_domains: vh_domains.into(), } } - pub fn quota(&self) -> RateLimitQuota { - self.limiter.quota() + pub fn client_quota(&self) -> Option { + self.client_limiter.as_ref().map(|limiter| limiter.quota()) + } + + pub fn bucket_quota(&self) -> Option { + self.bucket_limiter.as_ref().map(|limiter| limiter.quota()) } } @@ -405,7 +517,9 @@ impl Layer for RateLimitLayer { fn layer(&self, inner: S) -> Self::Service { RateLimitService { inner, - limiter: self.limiter.clone(), + client_limiter: self.client_limiter.clone(), + bucket_limiter: self.bucket_limiter.clone(), + vh_domains: self.vh_domains.clone(), } } } @@ -416,7 +530,39 @@ impl Layer for RateLimitLayer { #[derive(Debug, Clone)] pub struct RateLimitService { inner: S, - limiter: Arc, + client_limiter: Option>>, + bucket_limiter: Option>>, + vh_domains: Arc<[String]>, +} + +fn rejected_response( + req: &Request, + dimension: &'static str, + key: &dyn std::fmt::Display, + limit_rpm: u32, + throttle: &ThrottleInfo, +) -> Either, E>>> { + counter!( + METRIC_HTTP_SERVER_REQUESTS_RATE_LIMITED_TOTAL, + LABEL_RATE_LIMIT_SCOPE => RATE_LIMIT_SCOPE_S3_API, + LABEL_RATE_LIMIT_DIMENSION => dimension + ) + .increment(1); + debug!( + event = EVENT_REQUEST_RATE_LIMITED, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_RATE_LIMIT, + scope = RATE_LIMIT_SCOPE_S3_API, + dimension, + key = %key, + retry_after_secs = throttle.retry_after_secs, + "Request rejected by API rate limit" + ); + Either::Right(ready(Ok(s3_too_many_requests_response( + req.headers().get(X_REQUEST_ID), + limit_rpm, + throttle, + )))) } impl Service> for RateLimitService @@ -435,38 +581,34 @@ where if is_rate_limit_exempt_path(req.uri().path()) { return Either::Left(self.inner.call(req)); } - let Some(ip) = client_ip(&req) else { - // Fail open: without a socket peer address there is no trustworthy - // client identity, and deriving one from spoofable headers would - // let clients pick their own bucket. - return Either::Left(self.inner.call(req)); - }; - match self.limiter.check(ip) { - RateLimitDecision::Allowed => Either::Left(self.inner.call(req)), - RateLimitDecision::Limited(throttle) => { - counter!( - METRIC_HTTP_SERVER_REQUESTS_RATE_LIMITED_TOTAL, - LABEL_RATE_LIMIT_SCOPE => RATE_LIMIT_SCOPE_S3_API - ) - .increment(1); - debug!( - event = EVENT_REQUEST_RATE_LIMITED, - component = LOG_COMPONENT_SERVER, - subsystem = LOG_SUBSYSTEM_RATE_LIMIT, - scope = RATE_LIMIT_SCOPE_S3_API, - client_ip = %ip, - retry_after_secs = throttle.retry_after_secs, - "Request rejected by API rate limit" - ); - let limit_rpm = self.limiter.quota().requests_per_minute; - Either::Right(ready(Ok(s3_too_many_requests_response( - req.headers().get(X_REQUEST_ID), - limit_rpm, + if let Some(limiter) = &self.client_limiter { + // A request without a socket peer address fails open: there is no + // trustworthy client identity, and deriving one from spoofable + // headers would let clients pick their own budget. + if let Some(ip) = client_ip(&req) + && let RateLimitDecision::Limited(throttle) = limiter.check(&ip) + { + return rejected_response( + &req, + RATE_LIMIT_DIMENSION_CLIENT_IP, + &ip, + limiter.quota().requests_per_minute, &throttle, - )))) + ); } } + + if let Some(limiter) = &self.bucket_limiter + && let Some(bucket) = request_bucket(&req, &self.vh_domains) + && let RateLimitDecision::Limited(throttle) = limiter.check(bucket) + { + let limit_rpm = limiter.quota().requests_per_minute; + let bucket = bucket.to_owned(); + return rejected_response(&req, RATE_LIMIT_DIMENSION_BUCKET, &bucket, limit_rpm, &throttle); + } + + Either::Left(self.inner.call(req)) } } @@ -496,48 +638,58 @@ mod tests { #[test] fn bucket_exhaustion_returns_429_hints_and_window_recovers() { - let limiter = RateLimiter::new(quota(60, 5)); // 1 token/sec, burst 5 + let limiter: RateLimiter = RateLimiter::new(quota(60, 5)); // 1 token/sec, burst 5 let start = Instant::now(); for _ in 0..5 { - assert_eq!(limiter.check_at(ip(1), start), RateLimitDecision::Allowed); + assert_eq!(limiter.check_at(&ip(1), start), RateLimitDecision::Allowed); } - let RateLimitDecision::Limited(throttle) = limiter.check_at(ip(1), start) else { + let RateLimitDecision::Limited(throttle) = limiter.check_at(&ip(1), start) else { panic!("sixth request within the same instant must be limited"); }; assert_eq!(throttle.retry_after_secs, 1); assert_eq!(throttle.reset_after_secs, 5); // One refill interval restores exactly one token. - assert_eq!(limiter.check_at(ip(1), start + Duration::from_secs(1)), RateLimitDecision::Allowed); + assert_eq!(limiter.check_at(&ip(1), start + Duration::from_secs(1)), RateLimitDecision::Allowed); assert!(matches!( - limiter.check_at(ip(1), start + Duration::from_secs(1)), + limiter.check_at(&ip(1), start + Duration::from_secs(1)), RateLimitDecision::Limited(_) )); // A full window restores the full burst. let later = start + Duration::from_secs(61); for _ in 0..5 { - assert_eq!(limiter.check_at(ip(1), later), RateLimitDecision::Allowed); + assert_eq!(limiter.check_at(&ip(1), later), RateLimitDecision::Allowed); } - assert!(matches!(limiter.check_at(ip(1), later), RateLimitDecision::Limited(_))); + assert!(matches!(limiter.check_at(&ip(1), later), RateLimitDecision::Limited(_))); } #[test] fn clients_have_independent_buckets() { - let limiter = RateLimiter::new(quota(60, 1)); + let limiter: RateLimiter = RateLimiter::new(quota(60, 1)); let now = Instant::now(); - assert_eq!(limiter.check_at(ip(1), now), RateLimitDecision::Allowed); - assert!(matches!(limiter.check_at(ip(1), now), RateLimitDecision::Limited(_))); - assert_eq!(limiter.check_at(ip(2), now), RateLimitDecision::Allowed); + assert_eq!(limiter.check_at(&ip(1), now), RateLimitDecision::Allowed); + assert!(matches!(limiter.check_at(&ip(1), now), RateLimitDecision::Limited(_))); + assert_eq!(limiter.check_at(&ip(2), now), RateLimitDecision::Allowed); + } + + #[test] + fn string_keys_share_budget_across_borrowed_lookups() { + let limiter: RateLimiter = RateLimiter::new(quota(60, 2)); + let now = Instant::now(); + assert_eq!(limiter.check_at("photos", now), RateLimitDecision::Allowed); + assert_eq!(limiter.check_at("photos", now), RateLimitDecision::Allowed); + assert!(matches!(limiter.check_at("photos", now), RateLimitDecision::Limited(_))); + assert_eq!(limiter.check_at("logs", now), RateLimitDecision::Allowed); } #[test] fn tracked_clients_stay_bounded_under_distinct_ip_flood() { - let limiter = RateLimiter::with_limits(quota(60, 1), 4, 64); + let limiter: RateLimiter = RateLimiter::with_limits(quota(60, 1), 4, 64); let now = Instant::now(); for i in 0..10_000u32 { - limiter.check_at(IpAddr::V4(Ipv4Addr::from(i)), now); + limiter.check_at(&IpAddr::V4(Ipv4Addr::from(i)), now); } assert!( limiter.tracked_clients() <= 64, @@ -546,23 +698,38 @@ mod tests { ); } + #[test] + fn tracked_buckets_stay_bounded_under_random_name_flood() { + // Bucket names are attacker-chosen strings; the cap is the defense. + let limiter: RateLimiter = RateLimiter::with_limits(quota(60, 1), 4, 64); + let now = Instant::now(); + for i in 0..10_000u32 { + limiter.check_at(format!("bucket-{i}").as_str(), now); + } + assert!( + limiter.tracked_clients() <= 64, + "tracked {} buckets, cap is 64", + limiter.tracked_clients() + ); + } + #[test] fn refilled_idle_buckets_are_swept() { - let limiter = RateLimiter::with_limits(quota(60, 5), 1, 100); + let limiter: RateLimiter = RateLimiter::with_limits(quota(60, 5), 1, 100); let start = Instant::now(); - limiter.check_at(ip(1), start); + limiter.check_at(&ip(1), start); assert_eq!(limiter.tracked_clients(), 1); // After burst/rate = 5s the bucket is full again; the next request past // the cleanup deadline sweeps it. let after_sweep = start + CLEANUP_INTERVAL + Duration::from_secs(6); - limiter.check_at(ip(2), after_sweep); + limiter.check_at(&ip(2), after_sweep); assert_eq!(limiter.tracked_clients(), 1, "idle refilled bucket must be dropped"); } #[test] fn concurrent_hammering_never_exceeds_burst() { - let limiter = Arc::new(RateLimiter::new(quota(60, 100))); + let limiter: Arc = Arc::new(RateLimiter::new(quota(60, 100))); let now = Instant::now(); let allowed = Arc::new(std::sync::atomic::AtomicUsize::new(0)); @@ -572,7 +739,7 @@ mod tests { let allowed = Arc::clone(&allowed); scope.spawn(move || { for _ in 0..50 { - if limiter.check_at(ip(9), now) == RateLimitDecision::Allowed { + if limiter.check_at(&ip(9), now) == RateLimitDecision::Allowed { allowed.fetch_add(1, std::sync::atomic::Ordering::Relaxed); } } @@ -612,6 +779,56 @@ mod tests { } } + // ---- bucket extraction ---- + + fn request_with_host(host: &str, path: &str) -> Request<()> { + let mut req = Request::builder().uri(path).body(()).expect("request"); + req.headers_mut() + .insert(http::header::HOST, HeaderValue::from_str(host).expect("host")); + req + } + + #[test] + fn request_bucket_takes_first_path_segment_for_path_style() { + let domains: Vec = vec![]; + let req = request_with_host("s3.example.com", "/photos/2024/cat.jpg"); + assert_eq!(request_bucket(&req, &domains), Some("photos")); + + let root = request_with_host("s3.example.com", "/"); + assert_eq!(request_bucket(&root, &domains), None); + } + + #[test] + fn request_bucket_resolves_virtual_hosted_style_against_domains() { + let domains = vec!["s3.example.com".to_string(), "s3.example.com:9000".to_string()]; + + let vh = request_with_host("photos.s3.example.com", "/2024/cat.jpg"); + assert_eq!(request_bucket(&vh, &domains), Some("photos")); + + let vh_port = request_with_host("photos.s3.example.com:9000", "/2024/cat.jpg"); + assert_eq!(request_bucket(&vh_port, &domains), Some("photos")); + + let vh_case = request_with_host("Photos.S3.Example.COM", "/2024/cat.jpg"); + assert_eq!(request_bucket(&vh_case, &domains), Some("Photos")); + + // Exact domain match means path-style on the API domain. + let path_style = request_with_host("s3.example.com", "/photos/cat.jpg"); + assert_eq!(request_bucket(&path_style, &domains), Some("photos")); + + // Unrelated hosts fall back to path-style extraction. + let other = request_with_host("cdn.other.net", "/photos/cat.jpg"); + assert_eq!(request_bucket(&other, &domains), Some("photos")); + } + + #[test] + fn request_bucket_skips_admin_and_catalog_namespaces() { + let domains: Vec = vec![]; + for path in ["/rustfs/admin/v3/info", "/minio/admin/v3/info", "/iceberg/v1/config"] { + let req = request_with_host("s3.example.com", path); + assert_eq!(request_bucket(&req, &domains), None, "{path} must not be a bucket"); + } + } + // ---- service-level tests ---- #[derive(Clone)] @@ -635,7 +852,7 @@ mod tests { } fn service_with_quota(rpm: u32, burst: u32) -> RateLimitService { - RateLimitLayer::new(quota(rpm, burst)).layer(OkService) + RateLimitLayer::new(Some(quota(rpm, burst)), None, vec![]).layer(OkService) } fn request_from(peer: IpAddr, path: &str) -> Request<()> { @@ -744,6 +961,77 @@ mod tests { } } + #[tokio::test] + async fn bucket_dimension_budget_is_shared_across_client_ips() { + // Bucket dimension only (client dimension off): two IPs, one bucket. + let mut service = RateLimitLayer::new(None, Some(quota(60, 2)), vec![]).layer(OkService); + + assert_eq!( + service.call(request_from(ip(1), "/photos/a.jpg")).await.expect("ok").status(), + StatusCode::OK + ); + assert_eq!( + service.call(request_from(ip(2), "/photos/b.jpg")).await.expect("ok").status(), + StatusCode::OK + ); + // Third hit on the same bucket is rejected regardless of the new IP. + assert_eq!( + service.call(request_from(ip(3), "/photos/c.jpg")).await.expect("ok").status(), + StatusCode::TOO_MANY_REQUESTS + ); + // A different bucket has its own budget. + assert_eq!( + service.call(request_from(ip(3), "/logs/d.log")).await.expect("ok").status(), + StatusCode::OK + ); + // Requests without a bucket (e.g. ListBuckets) skip the dimension. + assert_eq!(service.call(request_from(ip(3), "/")).await.expect("ok").status(), StatusCode::OK); + } + + #[tokio::test] + async fn bucket_dimension_resolves_virtual_hosted_requests() { + let domains = vec!["s3.example.com".to_string()]; + let mut service = RateLimitLayer::new(None, Some(quota(60, 1)), domains).layer(OkService); + + let mut vh = request_from(ip(1), "/a.jpg"); + vh.headers_mut() + .insert(http::header::HOST, HeaderValue::from_static("photos.s3.example.com")); + assert_eq!(service.call(vh).await.expect("ok").status(), StatusCode::OK); + + // Path-style hit on the same bucket shares the budget. + let path_style = request_from(ip(2), "/photos/b.jpg"); + assert_eq!(service.call(path_style).await.expect("ok").status(), StatusCode::TOO_MANY_REQUESTS); + } + + #[tokio::test] + async fn both_dimensions_apply_when_configured() { + // Client burst 2, bucket burst 1: the bucket dimension trips first for + // one bucket, the client dimension caps the client across buckets. + let mut service = RateLimitLayer::new(Some(quota(60, 2)), Some(quota(60, 1)), vec![]).layer(OkService); + + assert_eq!( + service.call(request_from(ip(1), "/photos/a.jpg")).await.expect("ok").status(), + StatusCode::OK + ); + assert_eq!( + service.call(request_from(ip(1), "/photos/b.jpg")).await.expect("ok").status(), + StatusCode::TOO_MANY_REQUESTS, + "bucket budget exhausted" + ); + // Client budget: the first request consumed 1, the rejected one also + // consumed 1 (arrivals are counted), so the client is now exhausted too. + assert_eq!( + service.call(request_from(ip(1), "/logs/c.log")).await.expect("ok").status(), + StatusCode::TOO_MANY_REQUESTS, + "client budget exhausted" + ); + // A different client can still reach a fresh bucket. + assert_eq!( + service.call(request_from(ip(2), "/logs/c.log")).await.expect("ok").status(), + StatusCode::OK + ); + } + #[test] #[serial] fn env_constructor_defaults_to_disabled() { @@ -752,24 +1040,29 @@ mod tests { (rustfs_config::ENV_API_RATE_LIMIT_ENABLE, None::<&str>), (rustfs_config::ENV_API_RATE_LIMIT_RPM, None), (rustfs_config::ENV_API_RATE_LIMIT_BURST, None), + (rustfs_config::ENV_API_RATE_LIMIT_BUCKET_RPM, None), + (rustfs_config::ENV_API_RATE_LIMIT_BUCKET_BURST, None), ], || { - assert!(api_rate_limit_layer_from_env().is_none()); + assert!(api_rate_limit_layer_from_env(vec![]).is_none()); }, ); } #[test] #[serial] - fn env_constructor_requires_enable_and_nonzero_rpm() { + fn env_constructor_requires_enable_and_a_nonzero_dimension() { temp_env::with_vars( [ (rustfs_config::ENV_API_RATE_LIMIT_ENABLE, Some("true")), (rustfs_config::ENV_API_RATE_LIMIT_RPM, None), - (rustfs_config::ENV_API_RATE_LIMIT_BURST, None), + (rustfs_config::ENV_API_RATE_LIMIT_BUCKET_RPM, None), ], || { - assert!(api_rate_limit_layer_from_env().is_none(), "enabled with rpm=0 stays inactive"); + assert!( + api_rate_limit_layer_from_env(vec![]).is_none(), + "enabled with all dimensions at 0 stays inactive" + ); }, ); @@ -777,10 +1070,9 @@ mod tests { [ (rustfs_config::ENV_API_RATE_LIMIT_ENABLE, Some("false")), (rustfs_config::ENV_API_RATE_LIMIT_RPM, Some("600")), - (rustfs_config::ENV_API_RATE_LIMIT_BURST, None), ], || { - assert!(api_rate_limit_layer_from_env().is_none(), "rpm without enable stays inactive"); + assert!(api_rate_limit_layer_from_env(vec![]).is_none(), "rpm without enable stays inactive"); }, ); @@ -789,15 +1081,37 @@ mod tests { (rustfs_config::ENV_API_RATE_LIMIT_ENABLE, Some("true")), (rustfs_config::ENV_API_RATE_LIMIT_RPM, Some("600")), (rustfs_config::ENV_API_RATE_LIMIT_BURST, Some("50")), + (rustfs_config::ENV_API_RATE_LIMIT_BUCKET_RPM, None), ], || { - let layer = api_rate_limit_layer_from_env().expect("enabled with rpm"); + let layer = api_rate_limit_layer_from_env(vec![]).expect("enabled with client rpm"); assert_eq!( - layer.quota(), - RateLimitQuota { + layer.client_quota(), + Some(RateLimitQuota { requests_per_minute: 600, burst: 50 - } + }) + ); + assert_eq!(layer.bucket_quota(), None); + }, + ); + + temp_env::with_vars( + [ + (rustfs_config::ENV_API_RATE_LIMIT_ENABLE, Some("true")), + (rustfs_config::ENV_API_RATE_LIMIT_RPM, None), + (rustfs_config::ENV_API_RATE_LIMIT_BUCKET_RPM, Some("6000")), + (rustfs_config::ENV_API_RATE_LIMIT_BUCKET_BURST, Some("100")), + ], + || { + let layer = api_rate_limit_layer_from_env(vec![]).expect("enabled with bucket rpm only"); + assert_eq!(layer.client_quota(), None); + assert_eq!( + layer.bucket_quota(), + Some(RateLimitQuota { + requests_per_minute: 6000, + burst: 100 + }) ); }, );