From da52303f2a05ef1bcff1295715dddcfc12e901ee Mon Sep 17 00:00:00 2001 From: Hauser Date: Tue, 13 Oct 2026 20:32:27 +0800 Subject: [PATCH] fix(server): restore socket peer fallback for client rate limits (#8519) Read the optional socket peer extension supplied by external ingress so API and Console client limits remain effective without trusted proxies. Keep validated client precedence and the existing missing-peer policy. Exercise the full external stack across HTTP request versions, direct IPv4/IPv6 peers, forged headers, and distinct trusted proxy clients. Align API and Console fixtures with the production extension contract. Co-authored-by: heihutu Co-authored-by: zhi22915 --- rustfs/src/admin/console.rs | 10 +- rustfs/src/server/http.rs | 157 ++++++++++++++++++++++++++++++++ rustfs/src/server/rate_limit.rs | 18 ++-- 3 files changed, 175 insertions(+), 10 deletions(-) diff --git a/rustfs/src/admin/console.rs b/rustfs/src/admin/console.rs index 4019c9876..f03a3f3ee 100644 --- a/rustfs/src/admin/console.rs +++ b/rustfs/src/admin/console.rs @@ -1051,10 +1051,12 @@ mod tests { .uri(format!("{CONSOLE_PREFIX}/index.html")) .body(Body::empty()) .expect("failed to build request"); - request.extensions_mut().insert(crate::server::RemoteAddr(SocketAddr::new( - IpAddr::from([198, 51, 100, last_octet]), - 50000, - ))); + request + .extensions_mut() + .insert(Some(crate::server::RemoteAddr(SocketAddr::new( + IpAddr::from([198, 51, 100, last_octet]), + 50000, + )))); request }; diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index d5a90e6bc..e0615835a 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -2818,6 +2818,163 @@ mod tests { use tokio::sync::{Notify, mpsc}; use tower::{Layer, Service, ServiceBuilder}; + type RateLimitTestCore = tower::util::BoxCloneService< + HttpRequest, + Response>>, + Box, + >; + type RateLimitTestStack = tower::util::BoxCloneService< + HttpRequest, + Response>>, + Box, + >; + + fn rate_limited_external_stack( + peer: SocketAddr, + proxy_layer: Option, + inner: RateLimitTestCore, + ) -> RateLimitTestStack { + use crate::server::rate_limit::RateLimitQuota; + + let readiness = Arc::new(GlobalReadiness::new()); + readiness.mark_stage(rustfs_common::SystemStage::FullReady); + RateLimitTestStack::new(external_service_stack!( + inner, + Some(RemoteAddr(peer)), + proxy_layer, + false, + Some(RateLimitLayer::new( + Some(RateLimitQuota { + requests_per_minute: 1, + burst: 1 + }), + None, + vec![] + )), + false, + readiness, + None::>, + HttpCompressionConfig::default(), + ServerContextSlot::new(), + false + )) + } + + #[tokio::test] + async fn external_stack_limits_socket_peers_without_trusted_proxies() { + use crate::server::hybrid::HybridBody; + use tower::ServiceExt; + + for peer in ["192.0.2.10:51000", "[2001:db8::10]:51000"] { + let peer: SocketAddr = peer.parse().expect("socket peer"); + for version in [Version::HTTP_11, Version::HTTP_2, Version::HTTP_3] { + let calls = Arc::new(AtomicUsize::new(0)); + let inner_calls = calls.clone(); + let inner = tower::service_fn(move |request: HttpRequest| { + assert_eq!( + request + .extensions() + .get::>() + .copied() + .flatten() + .map(|addr| addr.0), + Some(peer) + ); + inner_calls.fetch_add(1, Ordering::SeqCst); + async { + Ok::<_, Box>(Response::new( + HybridBody::>::Rest { + rest_body: s3s::Body::empty(), + }, + )) + } + }); + let mut service = rate_limited_external_stack(peer, None, RateLimitTestCore::new(inner)); + + for (forwarded, expected) in [ + ("203.0.113.1", StatusCode::OK), + ("203.0.113.2", StatusCode::TOO_MANY_REQUESTS), + ] { + let request = HttpRequest::builder() + .uri("/bucket/object") + .version(version) + .header("x-forwarded-for", forwarded) + .header("x-real-ip", forwarded) + .body(s3s::Body::empty()) + .expect("request"); + let response = service + .ready() + .await + .expect("service ready") + .call(request) + .await + .expect("response"); + assert_eq!(response.status(), expected, "peer={peer}, version={version:?}"); + if expected == StatusCode::TOO_MANY_REQUESTS { + assert!(response.headers().contains_key(http::header::RETRY_AFTER)); + let body = response.into_body().collect().await.expect("body").to_bytes(); + assert!(String::from_utf8_lossy(&body).contains("TooManyRequests")); + } + } + assert_eq!(calls.load(Ordering::SeqCst), 1, "throttled requests must not reach the inner service"); + } + } + } + + #[tokio::test] + async fn external_stack_limits_validated_proxy_clients_independently() { + use crate::server::hybrid::HybridBody; + use rustfs_trusted_proxies::{ + LegacyTrustedProxyLayer, TrustedProxy, TrustedProxyConfig, TrustedProxyLayer, ValidationMode, + }; + use tower::ServiceExt; + + let peer: SocketAddr = "192.0.2.100:51000".parse().expect("proxy peer"); + for version in [Version::HTTP_11, Version::HTTP_2, Version::HTTP_3] { + let proxy_config = + TrustedProxyConfig::new(vec![TrustedProxy::Single(peer.ip())], ValidationMode::HopByHop, false, 3, false, vec![]); + let proxy_layer = TrustedProxyLayer::legacy(LegacyTrustedProxyLayer::new(proxy_config, None, true)); + let calls = Arc::new(AtomicUsize::new(0)); + let inner_calls = calls.clone(); + let inner = tower::service_fn(move |request: HttpRequest| { + let info = request.extensions().get::().expect("validated client context"); + assert!(info.is_from_trusted_proxy); + assert_ne!(info.real_ip, peer.ip()); + inner_calls.fetch_add(1, Ordering::SeqCst); + async { + Ok::<_, Box>(Response::new( + HybridBody::>::Rest { + rest_body: s3s::Body::empty(), + }, + )) + } + }); + let mut service = rate_limited_external_stack(peer, Some(proxy_layer), RateLimitTestCore::new(inner)); + + for (client, expected) in [ + ("203.0.113.1", StatusCode::OK), + ("203.0.113.2", StatusCode::OK), + ("203.0.113.1", StatusCode::TOO_MANY_REQUESTS), + ] { + let request = HttpRequest::builder() + .uri("/bucket/object") + .version(version) + .header("x-forwarded-for", client) + .body(s3s::Body::empty()) + .expect("request"); + let response = service + .ready() + .await + .expect("service ready") + .call(request) + .await + .expect("response"); + assert_eq!(response.status(), expected, "client={client}, version={version:?}"); + } + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + } + #[test] fn peer_transport_close_kind_uses_io_error_sources() { #[derive(Debug)] diff --git a/rustfs/src/server/rate_limit.rs b/rustfs/src/server/rate_limit.rs index 483b805a0..ca39b8d28 100644 --- a/rustfs/src/server/rate_limit.rs +++ b/rustfs/src/server/rate_limit.rs @@ -301,14 +301,16 @@ impl RateLimiter { /// Resolve the client identity for rate limiting. /// /// Only trusted sources are consulted: `ClientInfo` is inserted by the -/// trusted-proxy layer after validating the proxy chain, and `RemoteAddr` is -/// the raw socket peer. Spoofable headers (`X-Forwarded-For`, `X-Real-IP`) -/// are deliberately never read here. +/// trusted-proxy layer after validating the proxy chain, and `Option` +/// carries the socket peer injected by the external HTTP stack. Spoofable +/// headers (`X-Forwarded-For`, `X-Real-IP`) are deliberately never read here. pub(crate) fn client_ip(req: &Request) -> Option { if let Some(info) = req.extensions().get::() { return Some(info.real_ip); } - req.extensions().get::().map(|remote| remote.0.ip()) + req.extensions() + .get::>() + .and_then(|remote| remote.map(|remote| remote.0.ip())) } /// Resolve the host a request addresses, for virtual-hosted-style bucket @@ -929,7 +931,7 @@ mod tests { fn request_from(peer: IpAddr, path: &str) -> Request<()> { let mut req = Request::builder().uri(path).body(()).expect("request"); - req.extensions_mut().insert(RemoteAddr(SocketAddr::new(peer, 51000))); + req.extensions_mut().insert(Some(RemoteAddr(SocketAddr::new(peer, 51000)))); req } @@ -1047,7 +1049,11 @@ mod tests { async fn missing_client_identity_fails_open() { let mut service = service_with_quota(60, 1); for _ in 0..3 { - let req = Request::builder().uri("/bucket/object").body(()).expect("request"); + let mut req = Request::builder().uri("/bucket/object").body(()).expect("request"); + req.extensions_mut().insert(None::); + req.headers_mut() + .insert("x-forwarded-for", HeaderValue::from_static("203.0.113.1")); + req.headers_mut().insert("x-real-ip", HeaderValue::from_static("203.0.113.2")); assert_eq!(service.call(req).await.expect("ok").status(), StatusCode::OK); } }