fix(server): align vhost domains with s3s port matching (#7051)

This commit is contained in:
houseme
2026-09-02 15:55:30 +08:00
committed by GitHub
parent 1bbfa71b11
commit de7422b508
3 changed files with 155 additions and 51 deletions
+66 -21
View File
@@ -28,6 +28,7 @@ use crate::server::{
StsQueryApiCompatLayer, VirtualHostStyleHintLayer, redact_sensitive_uri_query, StsQueryApiCompatLayer, VirtualHostStyleHintLayer, redact_sensitive_uri_query,
}, },
rate_limit::{RateLimitLayer, api_rate_limit_layer_from_env}, rate_limit::{RateLimitLayer, api_rate_limit_layer_from_env},
strip_valid_port_suffix,
tls_material::{ tls_material::{
TlsAcceptFailure, TlsAcceptorHolder, TlsHandshakeFailureKind, accept_tls_with_deadline, build_acceptor_from_loaded, TlsAcceptFailure, TlsAcceptorHolder, TlsHandshakeFailureKind, accept_tls_with_deadline, build_acceptor_from_loaded,
load_tls_material, spawn_reload_loop, load_tls_material, spawn_reload_loop,
@@ -162,6 +163,25 @@ fn rustfs_s3_config() -> S3Config {
s3_config s3_config
} }
fn s3_host_domains(config: &config::Config) -> Result<Option<Vec<String>>> {
if config.server_domains.is_empty() || config.console_enable {
return Ok(None);
}
let mut domains = Vec::with_capacity(config.server_domains.len());
let mut seen = std::collections::HashSet::with_capacity(config.server_domains.len());
for domain in &config.server_domains {
if seen.insert(strip_valid_port_suffix(domain).to_string()) {
domains.push(domain.clone());
}
}
MultiDomain::new(&domains)
.map_err(|err| Error::other(format!("invalid RUSTFS_SERVER_DOMAINS {:?}: {err}", config.server_domains)))?;
Ok(Some(domains))
}
const LOG_COMPONENT_SERVER: &str = "server"; const LOG_COMPONENT_SERVER: &str = "server";
const LOG_SUBSYSTEM_HTTP: &str = "http"; const LOG_SUBSYSTEM_HTTP: &str = "http";
const LOG_SUBSYSTEM_TRANSPORT: &str = "transport"; const LOG_SUBSYSTEM_TRANSPORT: &str = "transport";
@@ -1202,26 +1222,10 @@ pub async fn start_http_server(
); );
} }
// Expanded virtual-hosted-style domain set (with port variants); shared by // Canonical virtual-hosted-style base domains for the S3 listener. s3s
// the s3s host router below and the rate limit layer's bucket extraction. // matching is port-agnostic, so synthesized listener-port variants would
let host_domain_sets = if !config.server_domains.is_empty() && !config.console_enable { // overlap with the bare domains they were generated from.
MultiDomain::new(&config.server_domains).map_err(Error::other)?; // validate domains let host_domain_sets = s3_host_domains(config)?;
// 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(); let rate_limit_vh_domains: Vec<String> = host_domain_sets.iter().flatten().cloned().collect();
// Setup S3 service // Setup S3 service
@@ -1256,7 +1260,7 @@ pub async fn start_http_server(
let s3_config = rustfs_s3_config(); let s3_config = rustfs_s3_config();
b.set_config(Arc::new(StaticConfigProvider::new(Arc::new(s3_config)))); b.set_config(Arc::new(StaticConfigProvider::new(Arc::new(s3_config))));
// Virtual-hosted-style requests are only set up for S3 API when server domains are configured and console is disabled // Virtual-hosted-style requests are only set up for the S3 API listener when server domains are configured.
if let Some(domain_sets) = host_domain_sets { if let Some(domain_sets) = host_domain_sets {
info!( info!(
event = EVENT_HTTP_HOST_ROUTING, event = EVENT_HTTP_HOST_ROUTING,
@@ -2953,6 +2957,47 @@ mod tests {
assert_http1_early_response_accepts_streaming_body(true).await; assert_http1_early_response_accepts_streaming_body(true).await;
} }
fn test_config_with_domains(domains: &[&str]) -> config::Config {
let mut config = config::Config::new("127.0.0.1:9000", vec!["/tmp/rustfs-data".to_string()]);
config.server_domains = domains.iter().map(|domain| (*domain).to_string()).collect();
config.console_enable = false;
config
}
#[test]
fn s3_host_domains_accepts_bare_domain_without_listener_port_variant() {
let config = test_config_with_domains(&["oss.example.com"]);
let domains = s3_host_domains(&config)
.expect("bare domain should be valid")
.expect("domains");
assert_eq!(domains, vec!["oss.example.com"]);
}
#[test]
fn s3_host_domains_deduplicates_port_equivalent_domains() {
let config = test_config_with_domains(&["oss.example.com", "oss.example.com:9000", "logs.example.com:1234"]);
let domains = s3_host_domains(&config)
.expect("port-equivalent duplicates should collapse")
.expect("domains");
assert_eq!(domains, vec!["oss.example.com", "logs.example.com:1234"]);
}
#[test]
fn s3_host_domains_rejects_real_subdomain_overlap_with_context() {
let config = test_config_with_domains(&["example.com", "s3.example.com"]);
let err = s3_host_domains(&config).expect_err("real subdomain overlap must stay rejected");
let message = err.to_string();
assert!(message.contains("RUSTFS_SERVER_DOMAINS"), "{message}");
assert!(message.contains("example.com"), "{message}");
assert!(message.contains("s3.example.com"), "{message}");
}
#[test] #[test]
fn rustfs_s3_config_preserves_compatibility_over_s3s_defaults() { fn rustfs_s3_config_preserves_compatibility_over_s3s_defaults() {
let s3_config = rustfs_s3_config(); let s3_config = rustfs_s3_config();
+11
View File
@@ -83,6 +83,17 @@ pub(crate) use readiness::{collect_cluster_read_health_report, collect_cluster_w
pub use crate::shared_types::RemoteAddr; pub use crate::shared_types::RemoteAddr;
pub(crate) fn strip_valid_port_suffix(host: &str) -> &str {
if host.ends_with(']') {
return host;
}
match host.rsplit_once(':') {
Some((host, port)) if !port.is_empty() && port.bytes().all(|b| b.is_ascii_digit()) && port.parse::<u16>().is_ok() => host,
_ => host,
}
}
pub struct ShutdownHandle { pub struct ShutdownHandle {
shutdown_tx: Option<tokio::sync::broadcast::Sender<()>>, shutdown_tx: Option<tokio::sync::broadcast::Sender<()>>,
task_handle: Option<tokio::task::JoinHandle<()>>, task_handle: Option<tokio::task::JoinHandle<()>>,
+78 -30
View File
@@ -37,8 +37,9 @@
//! 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 //! - **Bucket extraction mirrors s3s host routing.** Virtual-hosted-style
//! requests resolve the bucket from the Host/authority against the same //! requests resolve the bucket from the Host/authority against the same
//! expanded server-domain set the s3s router uses; everything else takes the //! configured server-domain set the s3s router uses, ignoring valid explicit
//! first path segment. Admin and table-catalog namespaces are not buckets. //! ports the same way s3s does; 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).
@@ -55,7 +56,7 @@
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, is_admin_path, is_table_catalog_path, RPC_PREFIX, RemoteAddr, TONIC_PREFIX, has_path_prefix, is_admin_path, is_table_catalog_path, strip_valid_port_suffix,
}; };
use crate::storage_api::server::layer::request_context::RequestContext; use crate::storage_api::server::layer::request_context::RequestContext;
use bytes::Bytes; use bytes::Bytes;
@@ -65,7 +66,7 @@ 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::borrow::Borrow;
use std::collections::HashMap; use std::collections::{HashMap, HashSet};
use std::hash::{BuildHasher, Hash, 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};
@@ -318,14 +319,12 @@ fn request_host<B>(req: &Request<B>) -> Option<&str> {
req.headers().get(http::header::HOST).and_then(|value| value.to_str().ok()) req.headers().get(http::header::HOST).and_then(|value| value.to_str().ok())
} }
/// `bucket.domain` → `bucket` when `host` is a subdomain of `domain` /// `bucket.domain` → `bucket` when `host` is a subdomain of `domain_host`.
/// (ASCII-case-insensitive, port-inclusive — the configured domain set fn strip_vh_prefix<'a>(host: &'a str, domain_host: &str) -> Option<&'a str> {
/// already carries port variants). let (host_len, domain_len) = (host.len(), domain_host.len());
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 if host_len > domain_len + 1
&& host.as_bytes()[host_len - domain_len - 1] == b'.' && host.as_bytes()[host_len - domain_len - 1] == b'.'
&& host[host_len - domain_len..].eq_ignore_ascii_case(domain) && host[host_len - domain_len..].eq_ignore_ascii_case(domain_host)
{ {
Some(&host[..host_len - domain_len - 1]) Some(&host[..host_len - domain_len - 1])
} else { } else {
@@ -337,6 +336,20 @@ fn bounded_bucket_name(bucket: &str) -> Option<&str> {
(!bucket.is_empty() && bucket.len() <= MAX_BUCKET_NAME_LENGTH).then_some(bucket) (!bucket.is_empty() && bucket.len() <= MAX_BUCKET_NAME_LENGTH).then_some(bucket)
} }
fn rate_limit_vh_domain_hosts(vh_domains: Vec<String>) -> Arc<[String]> {
let mut domains = Vec::with_capacity(vh_domains.len());
let mut seen = HashSet::with_capacity(vh_domains.len());
for domain in vh_domains {
let domain_host = strip_valid_port_suffix(&domain);
if seen.insert(domain_host.to_ascii_lowercase()) {
domains.push(domain_host.to_string());
}
}
domains.into()
}
/// Extract the bucket a request addresses, if any. /// Extract the bucket a request addresses, if any.
/// ///
/// Mirrors s3s host routing: on a configured virtual-hosted-style domain the /// Mirrors s3s host routing: on a configured virtual-hosted-style domain the
@@ -345,21 +358,22 @@ fn bounded_bucket_name(bucket: &str) -> Option<&str> {
/// namespaces are not buckets. Best-effort by design — a request this cannot /// namespaces are not buckets. Best-effort by design — a request this cannot
/// classify simply skips the bucket dimension (the client dimension still /// classify simply skips the bucket dimension (the client dimension still
/// applies). /// applies).
fn request_bucket<'a, B>(req: &'a Request<B>, vh_domains: &[String]) -> Option<&'a str> { fn request_bucket<'a, B>(req: &'a Request<B>, vh_domain_hosts: &[String]) -> Option<&'a str> {
let path = req.uri().path(); let path = req.uri().path();
if is_admin_path(path) || is_table_catalog_path(path) { if is_admin_path(path) || is_table_catalog_path(path) {
return None; return None;
} }
if !vh_domains.is_empty() if !vh_domain_hosts.is_empty()
&& let Some(host) = request_host(req) && let Some(host) = request_host(req)
{ {
for domain in vh_domains { let host = strip_valid_port_suffix(host);
if host.eq_ignore_ascii_case(domain) { for domain_host in vh_domain_hosts {
if host.eq_ignore_ascii_case(domain_host) {
// Path-style request on the API domain itself. // Path-style request on the API domain itself.
break; break;
} }
if let Some(bucket) = strip_vh_prefix(host, domain) { if let Some(bucket) = strip_vh_prefix(host, domain_host) {
return bounded_bucket_name(bucket); return bounded_bucket_name(bucket);
} }
} }
@@ -445,11 +459,10 @@ fn s3_too_many_requests_response(request_id: Option<&str>, limit_rpm: u32, throt
/// Build the S3 API rate limit layer from `RUSTFS_API_RATE_LIMIT_*`. /// Build the S3 API rate limit layer from `RUSTFS_API_RATE_LIMIT_*`.
/// ///
/// `vh_domains` is the expanded virtual-hosted-style domain set (with port /// `vh_domains` is the configured virtual-hosted-style domain set the s3s host
/// variants) the s3s host router uses; pass an empty vec when virtual-hosted /// router uses; pass an empty vec when virtual-hosted routing is not configured.
/// routing is not configured. Returns `None` (no layer in the stack, zero /// Returns `None` (no layer in the stack, zero request-path change) unless
/// request-path change) unless explicitly enabled with a non-zero RPM on at /// explicitly enabled with a non-zero RPM on at least one dimension.
/// least one dimension.
pub fn api_rate_limit_layer_from_env(vh_domains: Vec<String>) -> Option<RateLimitLayer> { 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;
@@ -491,7 +504,7 @@ pub fn api_rate_limit_layer_from_env(vh_domains: Vec<String>) -> Option<RateLimi
pub struct RateLimitLayer { pub struct RateLimitLayer {
client_limiter: Option<Arc<RateLimiter<IpAddr>>>, client_limiter: Option<Arc<RateLimiter<IpAddr>>>,
bucket_limiter: Option<Arc<RateLimiter<String>>>, bucket_limiter: Option<Arc<RateLimiter<String>>>,
vh_domains: Arc<[String]>, vh_domain_hosts: Arc<[String]>,
} }
impl RateLimitLayer { impl RateLimitLayer {
@@ -499,7 +512,7 @@ impl RateLimitLayer {
Self { Self {
client_limiter: client_quota.map(|quota| 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))), bucket_limiter: bucket_quota.map(|quota| Arc::new(RateLimiter::new(quota))),
vh_domains: vh_domains.into(), vh_domain_hosts: rate_limit_vh_domain_hosts(vh_domains),
} }
} }
@@ -520,7 +533,7 @@ impl<S> Layer<S> for RateLimitLayer {
inner, inner,
client_limiter: self.client_limiter.clone(), client_limiter: self.client_limiter.clone(),
bucket_limiter: self.bucket_limiter.clone(), bucket_limiter: self.bucket_limiter.clone(),
vh_domains: self.vh_domains.clone(), vh_domain_hosts: self.vh_domain_hosts.clone(),
} }
} }
} }
@@ -533,7 +546,7 @@ pub struct RateLimitService<S> {
inner: S, inner: S,
client_limiter: Option<Arc<RateLimiter<IpAddr>>>, client_limiter: Option<Arc<RateLimiter<IpAddr>>>,
bucket_limiter: Option<Arc<RateLimiter<String>>>, bucket_limiter: Option<Arc<RateLimiter<String>>>,
vh_domains: Arc<[String]>, vh_domain_hosts: Arc<[String]>,
} }
fn rejected_response<F, E, ReqBody>( fn rejected_response<F, E, ReqBody>(
@@ -603,7 +616,7 @@ where
} }
if let Some(limiter) = &self.bucket_limiter if let Some(limiter) = &self.bucket_limiter
&& let Some(bucket) = request_bucket(&req, &self.vh_domains) && let Some(bucket) = request_bucket(&req, &self.vh_domain_hosts)
&& let RateLimitDecision::Limited(throttle) = limiter.check(bucket) && let RateLimitDecision::Limited(throttle) = limiter.check(bucket)
{ {
let limit_rpm = limiter.quota().requests_per_minute; let limit_rpm = limiter.quota().requests_per_minute;
@@ -791,9 +804,13 @@ mod tests {
req req
} }
fn vh_domain_hosts(domains: &[&str]) -> Arc<[String]> {
rate_limit_vh_domain_hosts(domains.iter().map(|domain| (*domain).to_string()).collect())
}
#[test] #[test]
fn request_bucket_takes_first_path_segment_for_path_style() { fn request_bucket_takes_first_path_segment_for_path_style() {
let domains: Vec<String> = vec![]; let domains = vh_domain_hosts(&[]);
let req = request_with_host("s3.example.com", "/photos/2024/cat.jpg"); let req = request_with_host("s3.example.com", "/photos/2024/cat.jpg");
assert_eq!(request_bucket(&req, &domains), Some("photos")); assert_eq!(request_bucket(&req, &domains), Some("photos"));
@@ -803,7 +820,7 @@ mod tests {
#[test] #[test]
fn request_bucket_rejects_oversized_path_segment() { fn request_bucket_rejects_oversized_path_segment() {
let domains: Vec<String> = vec![]; let domains = vh_domain_hosts(&[]);
let maximum = "a".repeat(MAX_BUCKET_NAME_LENGTH); let maximum = "a".repeat(MAX_BUCKET_NAME_LENGTH);
let maximum_path = format!("/{maximum}/object"); let maximum_path = format!("/{maximum}/object");
let maximum_req = request_with_host("s3.example.com", &maximum_path); let maximum_req = request_with_host("s3.example.com", &maximum_path);
@@ -816,7 +833,7 @@ mod tests {
#[test] #[test]
fn request_bucket_resolves_virtual_hosted_style_against_domains() { fn request_bucket_resolves_virtual_hosted_style_against_domains() {
let domains = vec!["s3.example.com".to_string(), "s3.example.com:9000".to_string()]; let domains = vh_domain_hosts(&["s3.example.com"]);
let vh = request_with_host("photos.s3.example.com", "/2024/cat.jpg"); let vh = request_with_host("photos.s3.example.com", "/2024/cat.jpg");
assert_eq!(request_bucket(&vh, &domains), Some("photos")); assert_eq!(request_bucket(&vh, &domains), Some("photos"));
@@ -836,9 +853,23 @@ mod tests {
assert_eq!(request_bucket(&other, &domains), Some("photos")); assert_eq!(request_bucket(&other, &domains), Some("photos"));
} }
#[test]
fn request_bucket_matches_virtual_hosted_domains_port_agnostically() {
let domains = vh_domain_hosts(&["s3.example.com:1234"]);
let vh_bare = request_with_host("photos.s3.example.com", "/2024/cat.jpg");
assert_eq!(request_bucket(&vh_bare, &domains), Some("photos"));
let vh_other_port = request_with_host("photos.s3.example.com:443", "/2024/cat.jpg");
assert_eq!(request_bucket(&vh_other_port, &domains), Some("photos"));
let malformed_port = request_with_host("photos.s3.example.com:http", "/2024/cat.jpg");
assert_eq!(request_bucket(&malformed_port, &domains), Some("2024"));
}
#[test] #[test]
fn request_bucket_rejects_oversized_virtual_host_prefix() { fn request_bucket_rejects_oversized_virtual_host_prefix() {
let domains = vec!["s3.example.com".to_string()]; let domains = vh_domain_hosts(&["s3.example.com"]);
let maximum = "a".repeat(MAX_BUCKET_NAME_LENGTH); let maximum = "a".repeat(MAX_BUCKET_NAME_LENGTH);
let maximum_req = request_with_host(&format!("{maximum}.s3.example.com"), "/object"); let maximum_req = request_with_host(&format!("{maximum}.s3.example.com"), "/object");
assert_eq!(request_bucket(&maximum_req, &domains), Some(maximum.as_str())); assert_eq!(request_bucket(&maximum_req, &domains), Some(maximum.as_str()));
@@ -850,7 +881,7 @@ mod tests {
#[test] #[test]
fn request_bucket_skips_admin_and_catalog_namespaces() { fn request_bucket_skips_admin_and_catalog_namespaces() {
let domains: Vec<String> = vec![]; let domains = vh_domain_hosts(&[]);
for path in ["/rustfs/admin/v3/info", "/minio/admin/v3/info", "/iceberg/v1/config"] { for path in ["/rustfs/admin/v3/info", "/minio/admin/v3/info", "/iceberg/v1/config"] {
let req = request_with_host("s3.example.com", path); let req = request_with_host("s3.example.com", path);
assert_eq!(request_bucket(&req, &domains), None, "{path} must not be a bucket"); assert_eq!(request_bucket(&req, &domains), None, "{path} must not be a bucket");
@@ -1050,6 +1081,23 @@ mod tests {
assert_eq!(service.call(path_style).await.expect("ok").status(), StatusCode::TOO_MANY_REQUESTS); assert_eq!(service.call(path_style).await.expect("ok").status(), StatusCode::TOO_MANY_REQUESTS);
} }
#[tokio::test]
async fn bucket_dimension_shares_budget_for_port_carrying_virtual_hosts() {
let domains = vec!["s3.example.com".to_string()];
let mut service = RateLimitLayer::new(None, Some(quota(60, 1)), domains).layer(OkService);
let mut with_port = request_from(ip(1), "/a.jpg");
with_port
.headers_mut()
.insert(http::header::HOST, HeaderValue::from_static("photos.s3.example.com:443"));
assert_eq!(service.call(with_port).await.expect("ok").status(), StatusCode::OK);
let mut bare = request_from(ip(2), "/b.jpg");
bare.headers_mut()
.insert(http::header::HOST, HeaderValue::from_static("photos.s3.example.com"));
assert_eq!(service.call(bare).await.expect("ok").status(), StatusCode::TOO_MANY_REQUESTS);
}
#[tokio::test] #[tokio::test]
async fn both_dimensions_apply_when_configured() { async fn both_dimensions_apply_when_configured() {
// Client burst 2, bucket burst 1: the bucket dimension trips first for // Client burst 2, bucket burst 1: the bucket dimension trips first for