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.
This commit is contained in:
Zhengchao An
2026-07-17 19:01:18 +08:00
committed by GitHub
parent e1fc4b12ea
commit 97edb2e5cf
5 changed files with 536 additions and 144 deletions
+23
View File
@@ -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). /// Default for `RUSTFS_API_RATE_LIMIT_BURST` (`0` = same as RPM).
pub const DEFAULT_API_RATE_LIMIT_BURST: u32 = 0; 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;
@@ -86,6 +86,52 @@ async fn api_rate_limit_enforces_429_with_retry_after_when_enabled() -> TestResu
Ok(()) 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] #[tokio::test]
#[serial] #[serial]
async fn api_rate_limit_stays_inert_by_default() -> TestResult { async fn api_rate_limit_stays_inert_by_default() -> TestResult {
+7 -5
View File
@@ -17,8 +17,9 @@ use crate::admin::storage_api::access::RequestContext;
use crate::license::has_valid_license; use crate::license::has_valid_license;
use crate::server::has_path_prefix; use crate::server::has_path_prefix;
use crate::server::rate_limit::{ use crate::server::rate_limit::{
LABEL_RATE_LIMIT_SCOPE, METRIC_HTTP_SERVER_REQUESTS_RATE_LIMITED_TOTAL, RATE_LIMIT_SCOPE_CONSOLE, RateLimitDecision, LABEL_RATE_LIMIT_DIMENSION, LABEL_RATE_LIMIT_SCOPE, METRIC_HTTP_SERVER_REQUESTS_RATE_LIMITED_TOTAL,
RateLimitQuota, RateLimiter, apply_throttle_headers, client_ip, RATE_LIMIT_DIMENSION_CLIENT_IP, RATE_LIMIT_SCOPE_CONSOLE, RateLimitDecision, RateLimitQuota, RateLimiter,
apply_throttle_headers, client_ip,
}; };
use crate::server::{ use crate::server::{
CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, HeaderMapCarrier, HealthProbe, LICENSE, RUSTFS_ADMIN_PREFIX, 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. // without one fail open rather than trusting spoofable headers.
if rate_limit_enable { if rate_limit_enable {
if let Some(quota) = RateLimitQuota::per_minute(rate_limit_rpm, 0) { if let Some(quota) = RateLimitQuota::per_minute(rate_limit_rpm, 0) {
let limiter = Arc::new(RateLimiter::new(quota)); let limiter: Arc<RateLimiter> = Arc::new(RateLimiter::new(quota));
app = app.layer(middleware::from_fn(move |req: Request, next: middleware::Next| { app = app.layer(middleware::from_fn(move |req: Request, next: middleware::Next| {
let limiter = limiter.clone(); let limiter = limiter.clone();
async move { async move {
match client_ip(&req).map(|ip| limiter.check(ip)) { match client_ip(&req).map(|ip| limiter.check(&ip)) {
Some(RateLimitDecision::Limited(throttle)) => { Some(RateLimitDecision::Limited(throttle)) => {
metrics::counter!( metrics::counter!(
METRIC_HTTP_SERVER_REQUESTS_RATE_LIMITED_TOTAL, 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); .increment(1);
debug!(retry_after_secs = throttle.retry_after_secs, "Console request rejected by rate limit"); debug!(retry_after_secs = throttle.retry_after_secs, "Console request rejected by rate limit");
+32 -25
View File
@@ -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<String> = host_domain_sets.iter().flatten().cloned().collect();
// Setup S3 service // Setup S3 service
// This project uses the S3S library to implement S3 services // This project uses the S3S library to implement S3 services
let s3_service = { let s3_service = {
@@ -620,24 +642,6 @@ pub async fn start_http_server(
let access_key = config.access_key.clone(); let access_key = config.access_key.clone();
let secret_key = config.secret_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 let metadata_route_host = host_domain_sets
.as_ref() .as_ref()
.map(MultiDomain::new) .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 // Per-client / per-bucket S3 API rate limiting (backlog#1191). Built once
// connection's stack shares the same limiter state; `None` (the default) // so every connection's stack shares the same limiter state; `None` (the
// leaves the request path unchanged. // default) leaves the request path unchanged.
let api_rate_limit_layer = api_rate_limit_layer_from_env(); let api_rate_limit_layer = api_rate_limit_layer_from_env(rate_limit_vh_domains);
match &api_rate_limit_layer { match &api_rate_limit_layer {
Some(layer) => { Some(layer) => {
let quota = layer.quota(); let client_quota = layer.client_quota();
let bucket_quota = layer.bucket_quota();
info!( info!(
event = EVENT_API_RATE_LIMIT_STATE, event = EVENT_API_RATE_LIMIT_STATE,
component = LOG_COMPONENT_SERVER, component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_HTTP, subsystem = LOG_SUBSYSTEM_HTTP,
state = "enabled", state = "enabled",
requests_per_minute = quota.requests_per_minute, client_rpm = client_quota.map(|q| q.requests_per_minute).unwrap_or(0),
burst = quota.burst, 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" "API rate limit state changed"
); );
} }
+428 -114
View File
@@ -14,35 +14,48 @@
//! Per-client request rate limiting for the S3 API (backlog#1191). //! Per-client request rate limiting for the S3 API (backlog#1191).
//! //!
//! A token-bucket limiter keyed by client IP. Over-limit requests are rejected //! A token-bucket limiter with two optional dimensions:
//! with `429 Too Many Requests`, a `Retry-After` header, and the //! - **client IP** (`RUSTFS_API_RATE_LIMIT_RPM`/`_BURST`): per-client fairness
//! `x-ratelimit-*` headers already used by the Swift protocol error mapping. //! 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: //! Scope and invariants:
//! - **Opt-in, default off.** [`api_rate_limit_layer_from_env`] returns `None` //! - **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`; //! unless `RUSTFS_API_RATE_LIMIT_ENABLE=true` and at least one dimension has
//! the layer is then absent from the service stack and the request path is //! a non-zero RPM; the layer is then absent from the service stack and the
//! unchanged. //! request path is unchanged.
//! - **Client identity is never taken from request headers.** The key is the //! - **Client identity is never taken from request headers.** The key is the
//! validated [`ClientInfo::real_ip`] inserted by the trusted-proxy layer //! validated [`ClientInfo::real_ip`] inserted by the trusted-proxy layer
//! (which only honors `X-Forwarded-For` from configured proxies) or, absent //! (which only honors `X-Forwarded-For` from configured proxies) or, absent
//! that, the socket peer address ([`RemoteAddr`]). A raw `X-Forwarded-For` //! that, the socket peer address ([`RemoteAddr`]). A raw `X-Forwarded-For`
//! header can therefore not be used to escape into an attacker-chosen bucket. //! 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 //! - **Infra traffic is exempt** ([`is_rate_limit_exempt_path`]): health and
//! profiling probes, internode RPC/gRPC, and the console (which has its own //! profiling probes, internode RPC/gRPC, and the console (which has its own
//! limiter, sharing this module's [`RateLimiter`] core). //! limiter, sharing this module's [`RateLimiter`] core).
//! - This is client-facing abuse protection, not internal backpressure — it is //! - This is client-facing abuse protection, not internal backpressure — it is
//! unrelated to `workload_admission`, which schedules already-admitted work. //! unrelated to `workload_admission`, which schedules already-admitted work.
//! //!
//! Memory is bounded: buckets live in [`SHARD_COUNT`] independently locked //! Memory is bounded per dimension: buckets live in [`SHARD_COUNT`]
//! shards, each capped and periodically swept. A bucket that has been idle //! independently locked shards, each capped and periodically swept. A bucket
//! long enough to have refilled completely is indistinguishable from a fresh //! that has been idle long enough to have refilled completely is
//! one, so eviction never makes the limiter more permissive than a restart. //! 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::{ use crate::server::{
CONSOLE_PREFIX, FAVICON_PATH, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_HEALTH_CLUSTER_PATH, 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, 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 bytes::Bytes;
use futures::future::{Either, Ready, ready}; use futures::future::{Either, Ready, ready};
@@ -50,8 +63,9 @@ use http::{HeaderMap, HeaderValue, Request, Response, StatusCode};
use http_body_util::{BodyExt, Full}; use http_body_util::{BodyExt, Full};
use metrics::counter; use metrics::counter;
use rustfs_trusted_proxies::ClientInfo; use rustfs_trusted_proxies::ClientInfo;
use std::borrow::Borrow;
use std::collections::HashMap; use std::collections::HashMap;
use std::hash::{BuildHasher, RandomState}; use std::hash::{BuildHasher, Hash, RandomState};
use std::net::IpAddr; use std::net::IpAddr;
use std::sync::{Arc, Mutex, PoisonError}; use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; 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 LABEL_RATE_LIMIT_SCOPE: &str = "scope";
pub(crate) const RATE_LIMIT_SCOPE_S3_API: &str = "s3_api"; pub(crate) const RATE_LIMIT_SCOPE_S3_API: &str = "s3_api";
pub(crate) const RATE_LIMIT_SCOPE_CONSOLE: &str = "console"; 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). // Header names shared with the Swift error mapping (crates/protocols swift/errors.rs).
const X_RATE_LIMIT_LIMIT: &str = "x-ratelimit-limit"; 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. /// Shards bound lock contention: each request locks 1/32 of the key space.
const SHARD_COUNT: usize = 32; const SHARD_COUNT: usize = 32;
/// Upper bound on tracked client IPs across all shards (~100 bytes each, so /// Upper bound on tracked keys per limiter instance (~100 bytes each, so
/// worst case a few MiB). Real distinct-IP cardinality is bounded by actual /// worst case a few MiB). For the client dimension real cardinality is
/// TCP peers (or validated proxy clients), never by spoofable headers. /// 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; const MAX_TRACKED_CLIENTS: usize = 100_000;
/// Per-shard sweep cadence for dropping refilled-idle buckets. /// Per-shard sweep cadence for dropping refilled-idle buckets.
const CLEANUP_INTERVAL: Duration = Duration::from_secs(30); 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)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct RateLimitQuota { pub struct RateLimitQuota {
/// Sustained budget in requests per minute. /// Sustained budget in requests per minute.
@@ -120,8 +138,8 @@ pub enum RateLimitDecision {
Limited(ThrottleInfo), Limited(ThrottleInfo),
} }
/// One client's bucket. Capacity and refill rate are limiter-wide (single /// One key's bucket. Capacity and refill rate are limiter-wide (single quota
/// quota for all clients), so they are not duplicated per entry. /// for all keys), so they are not duplicated per entry.
#[derive(Debug)] #[derive(Debug)]
struct TokenBucket { struct TokenBucket {
tokens: f64, tokens: f64,
@@ -163,16 +181,17 @@ impl TokenBucket {
} }
#[derive(Debug)] #[derive(Debug)]
struct Shard { struct Shard<K> {
clients: HashMap<IpAddr, TokenBucket>, clients: HashMap<K, TokenBucket>,
next_cleanup: Instant, next_cleanup: Instant,
} }
/// Sharded per-IP token-bucket limiter. Shared by the S3 API tower layer and /// Sharded token-bucket limiter over an arbitrary key dimension (client IP,
/// the console middleware (each with its own instance and quota). /// bucket name, ...). Shared by the S3 API tower layer and the console
/// middleware, each instance with its own quota.
#[derive(Debug)] #[derive(Debug)]
pub struct RateLimiter { pub struct RateLimiter<K = IpAddr> {
shards: Vec<Mutex<Shard>>, shards: Vec<Mutex<Shard<K>>>,
shard_hasher: RandomState, shard_hasher: RandomState,
quota: RateLimitQuota, quota: RateLimitQuota,
capacity: f64, capacity: f64,
@@ -180,7 +199,7 @@ pub struct RateLimiter {
max_clients_per_shard: usize, max_clients_per_shard: usize,
} }
impl RateLimiter { impl<K: Hash + Eq + Clone> RateLimiter<K> {
pub fn new(quota: RateLimitQuota) -> Self { pub fn new(quota: RateLimitQuota) -> Self {
Self::with_limits(quota, SHARD_COUNT, MAX_TRACKED_CLIENTS) Self::with_limits(quota, SHARD_COUNT, MAX_TRACKED_CLIENTS)
} }
@@ -208,13 +227,23 @@ impl RateLimiter {
self.quota self.quota
} }
/// Consume one token for `ip`, creating its bucket on first sight. /// Consume one token for `key`, creating its bucket on first sight. The
pub fn check(&self, ip: IpAddr) -> RateLimitDecision { /// borrowed-key form keeps the hit path allocation-free for owned key
self.check_at(ip, Instant::now()) /// types (e.g. `&str` lookups against `String` keys).
pub fn check<Q>(&self, key: &Q) -> RateLimitDecision
where
K: Borrow<Q>,
Q: Hash + Eq + ToOwned<Owned = K> + ?Sized,
{
self.check_at(key, Instant::now())
} }
fn check_at(&self, ip: IpAddr, now: Instant) -> RateLimitDecision { fn check_at<Q>(&self, key: &Q, now: Instant) -> RateLimitDecision
let shard = &self.shards[self.shard_index(ip)]; where
K: Borrow<Q>,
Q: Hash + Eq + ToOwned<Owned = K> + ?Sized,
{
let shard = &self.shards[self.shard_index(key)];
let mut shard = shard.lock().unwrap_or_else(PoisonError::into_inner); let mut shard = shard.lock().unwrap_or_else(PoisonError::into_inner);
if now >= shard.next_cleanup { if now >= shard.next_cleanup {
@@ -225,33 +254,35 @@ impl RateLimiter {
shard.next_cleanup = now + CLEANUP_INTERVAL; 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); return bucket.try_consume(self.capacity, self.refill_per_sec, now);
} }
if shard.clients.len() >= self.max_clients_per_shard { if shard.clients.len() >= self.max_clients_per_shard {
// At capacity: reclaim the most idle entry (closest to full, so the // At capacity: reclaim the most idle entry (closest to full, so the
// least information is lost). O(shard len) but only under a // 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 if let Some(stalest) = shard
.clients .clients
.iter() .iter()
.min_by_key(|(_, bucket)| bucket.last_refill) .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::<K>(&stalest);
} }
} }
shard shard
.clients .clients
.entry(ip) .entry(key.to_owned())
.or_insert_with(|| TokenBucket::full(self.capacity, now)) .or_insert_with(|| TokenBucket::full(self.capacity, now))
.try_consume(self.capacity, self.refill_per_sec, now) .try_consume(self.capacity, self.refill_per_sec, now)
} }
fn shard_index(&self, ip: IpAddr) -> usize { fn shard_index<Q: Hash + ?Sized>(&self, key: &Q) -> usize {
(self.shard_hasher.hash_one(ip) as usize) % self.shards.len() (self.shard_hasher.hash_one(key) as usize) % self.shards.len()
} }
#[cfg(test)] #[cfg(test)]
@@ -276,6 +307,62 @@ pub(crate) fn client_ip<B>(req: &Request<B>) -> Option<IpAddr> {
req.extensions().get::<RemoteAddr>().map(|remote| remote.0.ip()) req.extensions().get::<RemoteAddr>().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<B>(req: &Request<B>) -> 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<B>, 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. /// Paths exempt from S3 API rate limiting.
/// ///
/// - Health/profiling probes: kubelet and load-balancer checks must never be /// - 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_*`. /// Build the S3 API rate limit layer from `RUSTFS_API_RATE_LIMIT_*`.
/// ///
/// Returns `None` (no layer in the stack, zero request-path change) unless /// `vh_domains` is the expanded virtual-hosted-style domain set (with port
/// explicitly enabled with a non-zero RPM. /// variants) the s3s host router uses; pass an empty vec when virtual-hosted
pub fn api_rate_limit_layer_from_env() -> Option<RateLimitLayer> { /// 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<String>) -> Option<RateLimitLayer> {
if !rustfs_utils::get_env_bool(rustfs_config::ENV_API_RATE_LIMIT_ENABLE, rustfs_config::DEFAULT_API_RATE_LIMIT_ENABLE) { if !rustfs_utils::get_env_bool(rustfs_config::ENV_API_RATE_LIMIT_ENABLE, rustfs_config::DEFAULT_API_RATE_LIMIT_ENABLE) {
return None; return None;
} }
let rpm = rustfs_utils::get_env_u32(rustfs_config::ENV_API_RATE_LIMIT_RPM, rustfs_config::DEFAULT_API_RATE_LIMIT_RPM); let client_quota = RateLimitQuota::per_minute(
let burst = rustfs_utils::get_env_u32(rustfs_config::ENV_API_RATE_LIMIT_BURST, rustfs_config::DEFAULT_API_RATE_LIMIT_BURST); rustfs_utils::get_env_u32(rustfs_config::ENV_API_RATE_LIMIT_RPM, rustfs_config::DEFAULT_API_RATE_LIMIT_RPM),
let Some(quota) = RateLimitQuota::per_minute(rpm, burst) else { 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!( warn!(
component = LOG_COMPONENT_SERVER, component = LOG_COMPONENT_SERVER,
subsystem = LOG_SUBSYSTEM_RATE_LIMIT, 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_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; 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 /// All clones share the same [`RateLimiter`] instances, so per-connection
/// keeps a single global view of client budgets. /// stack construction keeps a single global view of budgets.
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct RateLimitLayer { pub struct RateLimitLayer {
limiter: Arc<RateLimiter>, client_limiter: Option<Arc<RateLimiter<IpAddr>>>,
bucket_limiter: Option<Arc<RateLimiter<String>>>,
vh_domains: Arc<[String]>,
} }
impl RateLimitLayer { impl RateLimitLayer {
pub fn new(quota: RateLimitQuota) -> Self { pub fn new(client_quota: Option<RateLimitQuota>, bucket_quota: Option<RateLimitQuota>, vh_domains: Vec<String>) -> Self {
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 { pub fn client_quota(&self) -> Option<RateLimitQuota> {
self.limiter.quota() self.client_limiter.as_ref().map(|limiter| limiter.quota())
}
pub fn bucket_quota(&self) -> Option<RateLimitQuota> {
self.bucket_limiter.as_ref().map(|limiter| limiter.quota())
} }
} }
@@ -405,7 +517,9 @@ impl<S> Layer<S> for RateLimitLayer {
fn layer(&self, inner: S) -> Self::Service { fn layer(&self, inner: S) -> Self::Service {
RateLimitService { RateLimitService {
inner, 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<S> Layer<S> for RateLimitLayer {
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct RateLimitService<S> { pub struct RateLimitService<S> {
inner: S, inner: S,
limiter: Arc<RateLimiter>, client_limiter: Option<Arc<RateLimiter<IpAddr>>>,
bucket_limiter: Option<Arc<RateLimiter<String>>>,
vh_domains: Arc<[String]>,
}
fn rejected_response<F, E, ReqBody>(
req: &Request<ReqBody>,
dimension: &'static str,
key: &dyn std::fmt::Display,
limit_rpm: u32,
throttle: &ThrottleInfo,
) -> Either<F, Ready<Result<Response<BoxBody>, 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<S, ReqBody> Service<Request<ReqBody>> for RateLimitService<S> impl<S, ReqBody> Service<Request<ReqBody>> for RateLimitService<S>
@@ -435,38 +581,34 @@ where
if is_rate_limit_exempt_path(req.uri().path()) { if is_rate_limit_exempt_path(req.uri().path()) {
return Either::Left(self.inner.call(req)); 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) { if let Some(limiter) = &self.client_limiter {
RateLimitDecision::Allowed => Either::Left(self.inner.call(req)), // A request without a socket peer address fails open: there is no
RateLimitDecision::Limited(throttle) => { // trustworthy client identity, and deriving one from spoofable
counter!( // headers would let clients pick their own budget.
METRIC_HTTP_SERVER_REQUESTS_RATE_LIMITED_TOTAL, if let Some(ip) = client_ip(&req)
LABEL_RATE_LIMIT_SCOPE => RATE_LIMIT_SCOPE_S3_API && let RateLimitDecision::Limited(throttle) = limiter.check(&ip)
) {
.increment(1); return rejected_response(
debug!( &req,
event = EVENT_REQUEST_RATE_LIMITED, RATE_LIMIT_DIMENSION_CLIENT_IP,
component = LOG_COMPONENT_SERVER, &ip,
subsystem = LOG_SUBSYSTEM_RATE_LIMIT, limiter.quota().requests_per_minute,
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,
&throttle, &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] #[test]
fn bucket_exhaustion_returns_429_hints_and_window_recovers() { 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(); let start = Instant::now();
for _ in 0..5 { 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"); panic!("sixth request within the same instant must be limited");
}; };
assert_eq!(throttle.retry_after_secs, 1); assert_eq!(throttle.retry_after_secs, 1);
assert_eq!(throttle.reset_after_secs, 5); assert_eq!(throttle.reset_after_secs, 5);
// One refill interval restores exactly one token. // 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!( assert!(matches!(
limiter.check_at(ip(1), start + Duration::from_secs(1)), limiter.check_at(&ip(1), start + Duration::from_secs(1)),
RateLimitDecision::Limited(_) RateLimitDecision::Limited(_)
)); ));
// A full window restores the full burst. // A full window restores the full burst.
let later = start + Duration::from_secs(61); let later = start + Duration::from_secs(61);
for _ in 0..5 { 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] #[test]
fn clients_have_independent_buckets() { fn clients_have_independent_buckets() {
let limiter = RateLimiter::new(quota(60, 1)); let limiter: RateLimiter = RateLimiter::new(quota(60, 1));
let now = Instant::now(); let now = Instant::now();
assert_eq!(limiter.check_at(ip(1), now), RateLimitDecision::Allowed); assert_eq!(limiter.check_at(&ip(1), now), RateLimitDecision::Allowed);
assert!(matches!(limiter.check_at(ip(1), now), RateLimitDecision::Limited(_))); 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(2), now), RateLimitDecision::Allowed);
}
#[test]
fn string_keys_share_budget_across_borrowed_lookups() {
let limiter: RateLimiter<String> = 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] #[test]
fn tracked_clients_stay_bounded_under_distinct_ip_flood() { 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(); let now = Instant::now();
for i in 0..10_000u32 { 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!( assert!(
limiter.tracked_clients() <= 64, 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<String> = 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] #[test]
fn refilled_idle_buckets_are_swept() { 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(); let start = Instant::now();
limiter.check_at(ip(1), start); limiter.check_at(&ip(1), start);
assert_eq!(limiter.tracked_clients(), 1); assert_eq!(limiter.tracked_clients(), 1);
// After burst/rate = 5s the bucket is full again; the next request past // After burst/rate = 5s the bucket is full again; the next request past
// the cleanup deadline sweeps it. // the cleanup deadline sweeps it.
let after_sweep = start + CLEANUP_INTERVAL + Duration::from_secs(6); 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"); assert_eq!(limiter.tracked_clients(), 1, "idle refilled bucket must be dropped");
} }
#[test] #[test]
fn concurrent_hammering_never_exceeds_burst() { fn concurrent_hammering_never_exceeds_burst() {
let limiter = Arc::new(RateLimiter::new(quota(60, 100))); let limiter: Arc<RateLimiter> = Arc::new(RateLimiter::new(quota(60, 100)));
let now = Instant::now(); let now = Instant::now();
let allowed = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let allowed = Arc::new(std::sync::atomic::AtomicUsize::new(0));
@@ -572,7 +739,7 @@ mod tests {
let allowed = Arc::clone(&allowed); let allowed = Arc::clone(&allowed);
scope.spawn(move || { scope.spawn(move || {
for _ in 0..50 { 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); 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<String> = 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<String> = 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 ---- // ---- service-level tests ----
#[derive(Clone)] #[derive(Clone)]
@@ -635,7 +852,7 @@ mod tests {
} }
fn service_with_quota(rpm: u32, burst: u32) -> RateLimitService<OkService> { fn service_with_quota(rpm: u32, burst: u32) -> RateLimitService<OkService> {
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<()> { 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] #[test]
#[serial] #[serial]
fn env_constructor_defaults_to_disabled() { 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_ENABLE, None::<&str>),
(rustfs_config::ENV_API_RATE_LIMIT_RPM, None), (rustfs_config::ENV_API_RATE_LIMIT_RPM, None),
(rustfs_config::ENV_API_RATE_LIMIT_BURST, 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] #[test]
#[serial] #[serial]
fn env_constructor_requires_enable_and_nonzero_rpm() { fn env_constructor_requires_enable_and_a_nonzero_dimension() {
temp_env::with_vars( temp_env::with_vars(
[ [
(rustfs_config::ENV_API_RATE_LIMIT_ENABLE, Some("true")), (rustfs_config::ENV_API_RATE_LIMIT_ENABLE, Some("true")),
(rustfs_config::ENV_API_RATE_LIMIT_RPM, None), (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_ENABLE, Some("false")),
(rustfs_config::ENV_API_RATE_LIMIT_RPM, Some("600")), (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_ENABLE, Some("true")),
(rustfs_config::ENV_API_RATE_LIMIT_RPM, Some("600")), (rustfs_config::ENV_API_RATE_LIMIT_RPM, Some("600")),
(rustfs_config::ENV_API_RATE_LIMIT_BURST, Some("50")), (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!( assert_eq!(
layer.quota(), layer.client_quota(),
RateLimitQuota { Some(RateLimitQuota {
requests_per_minute: 600, requests_per_minute: 600,
burst: 50 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
})
); );
}, },
); );