diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f2bbe9b39..28b1cd0a2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -509,7 +509,7 @@ jobs: runs-on: sm-standard-4 timeout-minutes: 90 strategy: - # Preserve both independent results when one protocol fails, so diagnosis + # Preserve independent results when one protocol fails, so diagnosis # and a focused fix do not require rebuilding an interrupted sibling. fail-fast: false matrix: @@ -518,6 +518,8 @@ jobs: flags: "--features swift" - name: sftp flags: "--features sftp" + - name: http3 + flags: "--features http3,swift" env: FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" steps: diff --git a/Cargo.lock b/Cargo.lock index 5374c9ecd..e3eda47a6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5003,6 +5003,34 @@ dependencies = [ "tracing", ] +[[package]] +name = "h3" +version = "0.0.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10872b55cfb02a821b69dc7cf8dc6a71d6af25eb9a79662bec4a9d016056b3be" +dependencies = [ + "bytes", + "fastrand", + "futures-util", + "http 1.5.0", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "h3-quinn" +version = "0.0.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2e732c8d91a74731663ac8479ab505042fbf547b9a207213ab7fbcbfc4f8b4" +dependencies = [ + "bytes", + "futures", + "h3", + "quinn", + "tokio", + "tokio-util", +] + [[package]] name = "half" version = "2.7.1" @@ -8649,6 +8677,7 @@ checksum = "4051e23e9185c255a7e33ef59cdbca87a22d359052eecd22fc6b901fb37d9d11" dependencies = [ "bytes", "cfg_aliases", + "futures-io", "pin-project-lite", "quinn-proto", "quinn-udp", @@ -9487,6 +9516,8 @@ dependencies = [ "futures-lite", "futures-util", "google-cloud-auth", + "h3", + "h3-quinn", "hashbrown 0.17.1", "hex-simd", "hmac 0.13.0", @@ -9518,6 +9549,7 @@ dependencies = [ "prost 0.14.4", "pyroscope", "quick-xml", + "quinn", "rand 0.10.3", "rcgen", "regex", diff --git a/Cargo.toml b/Cargo.toml index edef4184e..dc105b562 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -178,6 +178,13 @@ tonic-prost = { version = "0.14.6" } tonic-prost-build = { version = "0.14.6" } tower = { version = "0.5.3" } tower-http = { version = "0.7.1" } +h3 = "0.0.8" +h3-quinn = "0.0.10" +quinn = { version = "0.11.11", default-features = false, features = [ + "futures-io", + "runtime-tokio", + "rustls-aws-lc-rs", +] } # Serialization and Data Formats apache-avro = { version = "0.22.0", features = ["snappy", "zstandard"] } diff --git a/crates/config/src/constants/tls.rs b/crates/config/src/constants/tls.rs index ec208ad5a..65c575926 100644 --- a/crates/config/src/constants/tls.rs +++ b/crates/config/src/constants/tls.rs @@ -207,3 +207,9 @@ pub const ENV_TLS_RELOAD_INTERVAL: &str = "RUSTFS_TLS_RELOAD_INTERVAL"; /// Default interval for TLS certificate reload check pub const DEFAULT_TLS_RELOAD_INTERVAL: u64 = 30; + +/// Environment variable to enable experimental HTTP/3 support. +pub const ENV_HTTP3_ENABLE: &str = "RUSTFS_HTTP3_ENABLE"; + +/// HTTP/3 is disabled unless explicitly enabled. +pub const DEFAULT_HTTP3_ENABLE: bool = false; diff --git a/crates/keystone/src/middleware.rs b/crates/keystone/src/middleware.rs index 769e9edd7..360dfb255 100644 --- a/crates/keystone/src/middleware.rs +++ b/crates/keystone/src/middleware.rs @@ -40,7 +40,6 @@ use futures::Future; use http::{HeaderMap, Request, Response, StatusCode}; use http_body::Body; use http_body_util::{BodyExt, Full}; -use hyper::body::Incoming; use rustfs_credentials::Credentials; use std::pin::Pin; use std::sync::Arc; @@ -109,11 +108,12 @@ pub struct KeystoneAuthMiddleware { type BoxError = Box; type BoxBody = http_body_util::combinators::UnsyncBoxBody; -impl Service> for KeystoneAuthMiddleware +impl Service> for KeystoneAuthMiddleware where - S: Service, Response = Response> + Clone + Send + 'static, + S: Service, Response = Response> + Clone + Send + 'static, S::Future: Send + 'static, S::Error: Send + 'static, + ReqBody: Send + 'static, B: Body + Send + 'static, B::Error: Into + Send + 'static, { @@ -125,7 +125,7 @@ where self.inner.poll_ready(cx) } - fn call(&mut self, req: Request) -> Self::Future { + fn call(&mut self, req: Request) -> Self::Future { let keystone_auth = self.keystone_auth.clone(); let mut inner = self.inner.clone(); diff --git a/docs/operations/reverse-proxy.md b/docs/operations/reverse-proxy.md index 27bc32937..be17688ef 100644 --- a/docs/operations/reverse-proxy.md +++ b/docs/operations/reverse-proxy.md @@ -6,6 +6,12 @@ RustFS speaks plain S3 over HTTP/1.1 and HTTP/2. Most proxy problems are not RustFS storage bugs: if the same request works directly against `:9000`, the fault is in proxy/CDN request forwarding. Use the checklist below to find which forwarding behavior broke. +## Experimental HTTP/3 + +Build with `cargo build -p rustfs --features http3` and set `RUSTFS_HTTP3_ENABLE=true` together with `RUSTFS_TLS_PATH`. HTTP/3 listens on UDP at the S3 listener's TCP address and port. Allow UDP through your firewall or proxy; clients must explicitly support HTTP/3. The console and internode RPC remain on TCP. HTTP/3 uses TLS 1.3 with `h3` ALPN, disables early data and connection migration, and follows the existing certificate reload settings. + +The external request stack is shared with TCP, including readiness, Keystone authentication, trusted proxy validation, rate limiting, CORS, and response compatibility. `RUSTFS_API_MAX_CONNECTIONS` bounds each transport listener separately; a value of `0` disables this cap. HTTP/3 refuses excess connections before their handshake. `RUSTFS_HTTP_REQUEST_BODY_READ_TIMEOUT` also bounds idle HTTP/3 body reads; unread or timed-out streams are canceled without draining them in a background task. Shutdown allows up to ten seconds for active HTTP/3 requests to finish. + ## What RustFS requires from the proxy S3 clients sign requests with AWS SigV4. RustFS (via `s3s`) re-derives the signature from the forwarded request and streams the request body to storage, so the proxy must forward the signed material and the body byte-for-byte: diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 06028243e..9bc4de49e 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -77,6 +77,7 @@ rio-v2 = ["rustfs-ecstore/rio-v2"] pyroscope = ["rustfs-obs/pyroscope", "dep:pyroscope", "dep:object", "dep:symbolic-demangle"] # Tokio runtime telemetry. Requires `--cfg tokio_unstable`; use `make build-profiling`. dial9 = ["rustfs-obs/dial9"] +http3 = ["dep:h3", "dep:h3-quinn", "dep:quinn"] hotpath = [ "hotpath/hotpath", "hotpath/tokio", @@ -300,6 +301,9 @@ tokio-util = { workspace = true, features = ["io", "compat", "time"] } tonic = { workspace = true, features = ["gzip", "deflate"] } tower = { workspace = true, features = ["timeout"] } tower-http = { workspace = true, features = ["trace", "compression-full", "cors", "catch-panic", "timeout", "limit", "request-id", "add-extension"] } +h3 = { workspace = true, optional = true } +h3-quinn = { workspace = true, optional = true } +quinn = { workspace = true, optional = true } # Serialization and Data Formats apache-avro = { workspace = true } diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 111c2c98c..8feed740a 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -36,6 +36,8 @@ use crate::server::{ load_tls_material, spawn_reload_loop, }, }; +#[cfg(feature = "http3")] +use crate::server::{http3, hybrid::HybridBody, tls_material::build_http3_server_config_from_loaded}; use crate::storage_api::server::http as storage; use crate::storage_api::server::http::rpc::InternodeRpcService; #[cfg(test)] @@ -101,6 +103,20 @@ use tower_http::classify::ServerErrorsFailureClass; use tower_http::compression::CompressionLayer; use tower_http::request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer}; use tower_http::trace::TraceLayer; + +#[cfg(feature = "http3")] +type Http3CoreService = tower::util::BoxCloneService< + HttpRequest, + Response>>, + Box, +>; + +#[cfg(feature = "http3")] +type Http3ExternalService = tower::util::BoxCloneService< + HttpRequest, + Response>>, + Box, +>; use tracing::{Level, Span, debug, error, info, instrument, trace, warn}; use tracing_opentelemetry::OpenTelemetrySpanExt; @@ -1086,6 +1102,153 @@ fn make_http_trace_span(request: &HttpRequest) -> Span { span } +// Keep TCP and QUIC on the same external request stack. A macro preserves the +// concrete Tower types for each streaming request body without boxing TCP. +macro_rules! external_service_stack { + ($service:expr, $remote_addr:expr, $trusted_proxy_layer:expr, $is_console:expr, + $rate_limit_layer:expr, $connection_is_tls:expr, $readiness:expr, $keystone_auth:expr, + $compression_config:expr, $server_ctx:expr, $server_domains_configured:expr) => {{ + let remote_addr = $remote_addr; + let trusted_proxy_layer = $trusted_proxy_layer; + let is_console = $is_console; + let rate_limit_layer = $rate_limit_layer; + let connection_is_tls = $connection_is_tls; + let readiness = $readiness; + let keystone_auth = $keystone_auth; + let compression_config = $compression_config; + let server_ctx = $server_ctx; + let server_domains_configured = $server_domains_configured; + ServiceBuilder::new() + // NOTE: Both extension types are intentionally inserted to maintain compatibility: + // 1. `Option` - Used by existing admin/storage handlers throughout the codebase + // 2. `std::net::SocketAddr` - Required by TrustedProxyMiddleware for proxy validation + // This dual insertion is necessary because the middleware expects the raw SocketAddr type + // while our application code uses the RemoteAddr wrapper. Consolidating these would + // require either modifying the third-party middleware or refactoring all existing handlers. + .layer(AddExtensionLayer::new(remote_addr)) + .option_layer(remote_addr.map(|ra| AddExtensionLayer::new(ra.0))) + // Add TrustedProxyLayer to handle X-Forwarded-For and other proxy headers + // This should be placed before TraceLayer so that logs reflect the real client IP + // Pre-computed in ConnectionContext to avoid per-connection is_enabled() check. + .option_layer(trusted_proxy_layer) + .map_request(move |mut request: HttpRequest<_>| { + if request.version() == Version::HTTP_3 { + if let Some(info) = request.extensions_mut().get_mut::() { + if !info.is_from_trusted_proxy { + info.forwarded_proto = Some("https".to_owned()); + } + } else if let Some(peer) = remote_addr { + let mut info = ClientInfo::direct(peer.0); + info.forwarded_proto = Some("https".to_owned()); + request.extensions_mut().insert(info); + } + } + request + }) + .layer(ExternalRequestContextLayer::new(is_console)) + .layer(StsQueryApiCompatLayer) + .layer(EmptyBodyContentLengthCompatLayer) + .layer(CatchPanicLayer::new()) + // Per-client API rate limit (backlog#1191): rejects over-limit + // requests with 429 before readiness/auth/tracing spend any + // work on them, but after the trusted-proxy layer has resolved + // a spoof-proof client IP. Absent (None) unless enabled via + // RUSTFS_API_RATE_LIMIT_ENABLE with a non-zero RPM. + .option_layer(rate_limit_layer) + // backlog#2369 P7.2: an SSE-C request carries the customer key + // in a header, so a plaintext hop leaks it permanently. Sits + // beside the rate limiter: after the trusted-proxy layer, which + // is what makes a forwarded `https` protocol trustworthy, and + // after the request context so a rejection can echo the request + // id. Reports by default; refuses only under + // RUSTFS_SSE_C_REQUIRE_TLS. + .layer(SsecTransportLayer::new(connection_is_tls)) + // CRITICAL: Insert ReadinessGateLayer before business logic + // This stops requests from hitting IAMAuth or Storage if they are not ready. + .layer(ReadinessGateLayer::new(readiness.clone())) + // Add Keystone authentication middleware + // This validates X-Auth-Token headers and stores credentials in task-local storage + // Must be placed AFTER ReadinessGateLayer but BEFORE business logic + // Pre-computed in ConnectionContext to avoid per-connection OnceLock read. + .layer(KeystoneAuthLayer::new(keystone_auth)) + // Maintain the in-flight request gauge with an RAII guard so it is + // decremented exactly once per request (backlog#806-35). Placed just + // outside TraceLayer so the counting window matches the old on_request + // timing while avoiding the 5xx double-decrement. + .layer(InFlightLayer) + .layer( + TraceLayer::new_for_http() + .make_span_with(make_http_trace_span) + .on_request(|request: &HttpRequest<_>, span: &Span| { + let _enter = span.enter(); + trace!("HTTP request started"); + let method_metrics = http_method_metrics(request.method()); + // In-flight counting is handled by InFlightLayer's RAII guard + // (backlog#806-35); do not adjust the active-requests gauge here. + method_metrics.requests.increment(1); + + if let Some(cl) = request.headers().get("content-length") + && let Some(len) = cl.to_str().ok().and_then(|s| s.parse::().ok()) + { + REQUEST_BODY_BYTES_COUNTER.increment(len); + method_metrics.request_body_size.record(len as f64); + } + }) + .on_response(trace_on_response) + .on_body_chunk(|chunk: &Bytes, latency: Duration, span: &Span| { + RESP_BODY_BYTES_COUNTER.increment(usize_to_u64_saturating(chunk.len())); + record_response_body_chunk_observation(chunk.len(), latency); + #[cfg(feature = "tracing-chunk-debug")] + { + let _enter = span.enter(); + debug!(chunk_bytes = chunk.len(), duration_ms = duration_ms(latency), "HTTP response body chunk sent"); + } + #[cfg(not(feature = "tracing-chunk-debug"))] + { + let _ = (latency, span); + } + }) + .on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, span: &Span| { + record_response_body_stream_duration(stream_duration); + #[cfg(feature = "tracing-chunk-debug")] + { + let _enter = span.enter(); + debug!(duration_ms = duration_ms(stream_duration), "HTTP response stream closed"); + } + #[cfg(not(feature = "tracing-chunk-debug"))] + { + let _ = (_trailers, stream_duration, span); + } + }) + .on_failure(|error, latency: Duration, span: &Span| { + let _enter = span.enter(); + // In-flight counting is handled by InFlightLayer's RAII guard + // (backlog#806-35). This hook previously also fired for 5xx + // responses (which ALSO hit on_response), double-decrementing + // the gauge; only the failure metric is recorded here now. + HTTP_TRANSPORT_FAILURES_COUNTER.increment(1); + record_http_transport_failure_if_body_error(&error); + trace!(error = ?error, duration_ms = duration_ms(latency), "HTTP request failure captured by trace layer"); + }), + ) + .layer(RequestLoggingLayer) + .layer(CompressionLayer::new().compress_when(PathAwareHttpCompressionPredicate::new(compression_config.clone()))) + .option_layer(compression_config.enabled.then_some(PathCategoryInjectionLayer)) + .layer(S3ErrorMessageCompatLayer) + .layer(IcebergRestErrorCompatLayer) + .layer(ObjectAttributesEtagFixLayer) + .layer(ConditionalCorsLayer::new()) + .option_layer(if is_console { Some(RedirectLayer) } else { None }) + .layer(BodylessStatusFixLayer) + .layer(HeadRequestBodyFixLayer) + .layer(PublicHealthEndpointLayer::new(server_ctx, readiness)) + .option_layer((!server_domains_configured && !is_console).then_some(VirtualHostStyleHintLayer)) + .layer(DoubleSlashListBucketsCompatLayer) + .layer(SigV4HeaderGuardLayer) + .service($service) + }}; +} + pub async fn start_http_server( config: &config::Config, readiness: Arc, @@ -1269,6 +1432,19 @@ pub async fn start_http_server( let tls_path = config.tls_path.as_deref().map(str::trim).unwrap_or_default(); let tls_path_configured = !tls_path.is_empty(); + + #[cfg(feature = "http3")] + let http3_enabled = !config.console_enable + && rustfs_utils::get_env_bool(rustfs_config::ENV_HTTP3_ENABLE, rustfs_config::DEFAULT_HTTP3_ENABLE); + + #[cfg(feature = "http3")] + if http3_enabled && !tls_path_configured { + return Err(Error::other("HTTP/3 requires TLS configuration")); + } + + #[cfg(feature = "http3")] + let mut http3_server_config = None; + // Load TLS materials and build server acceptor in a single pass. // Outbound material (root CAs, mTLS identity) was already published in main.rs; // this load is needed for the server-side TLS acceptor and reload loop. @@ -1279,6 +1455,14 @@ pub async fn start_http_server( tls_path, e )) })?; + + #[cfg(feature = "http3")] + if http3_enabled { + http3_server_config = build_http3_server_config_from_loaded(snapshot.server.as_ref(), std::path::Path::new(tls_path)) + .await + .map_err(|e| Error::other(format!("HTTP/3 TLS configuration failed: {e}")))?; + } + let acceptor = build_acceptor_from_loaded(snapshot.server, std::path::Path::new(tls_path)) .await .map_err(|e| Error::other(e.to_string()))?; @@ -1300,10 +1484,6 @@ pub async fn start_http_server( let tls_enabled = tls_acceptor.is_some(); let protocol = if tls_enabled { "https" } else { "http" }; - // Spawn background TLS certificate hot-reload loop (if enabled). - if let Some(holder) = &tls_acceptor { - spawn_reload_loop(tls_path.to_string(), holder.clone()); - } // Obtain the listener address let local_addr: SocketAddr = listener.local_addr()?; let local_ip = match rustfs_utils::get_local_ip() { @@ -1431,6 +1611,7 @@ pub async fn start_http_server( // Create shutdown channel let (shutdown_tx, mut shutdown_rx) = tokio::sync::broadcast::channel(1); + // Create compression configuration from environment variables let compression_config = HttpCompressionConfig::from_env(); if compression_config.enabled { @@ -1512,6 +1693,77 @@ pub async fn start_http_server( ); } + let http_request_body_read_timeout = Duration::from_secs(rustfs_utils::get_env_u64( + rustfs_config::ENV_HTTP_REQUEST_BODY_READ_TIMEOUT, + rustfs_config::DEFAULT_HTTP_REQUEST_BODY_READ_TIMEOUT, + )); + #[cfg(feature = "http3")] + let http3_server = if let Some(server_config) = http3_server_config { + let h3_s3_service = s3_service.clone(); + let h3_readiness = Arc::clone(&readiness); + let h3_server_ctx = Arc::clone(&server_ctx); + let h3_keystone = auth_keystone::get_keystone_auth(); + let h3_rate_limit = api_rate_limit_layer.clone(); + let h3_compression = compression_config.clone(); + let h3_trusted_proxy = rustfs_trusted_proxies::is_enabled().then(|| rustfs_trusted_proxies::layer().clone()); + let h3_domains_configured = !config.server_domains.is_empty(); + + let make_service = move |peer_addr: SocketAddr| -> Http3ExternalService { + let remote_addr = RemoteAddr(peer_addr); + + #[cfg(feature = "swift")] + let service = SwiftService::new(true, None, h3_s3_service.clone()); + #[cfg(not(feature = "swift"))] + let service = h3_s3_service.clone(); + let service = EarlyResponseBodyService::new(service, http_request_body_read_timeout); + let service = ServiceBuilder::new() + .layer(tower::util::MapResponseLayer::new(|response: Response| { + response.map(|rest_body| HybridBody::>::Rest { rest_body }) + })) + .service(service); + #[cfg(not(feature = "swift"))] + let service = ServiceBuilder::new() + .map_err(|error: s3s::HttpError| -> Box { error.into() }) + .service(service); + let service = Http3CoreService::new(service); + Http3ExternalService::new(external_service_stack!( + service, + Some(remote_addr), + h3_trusted_proxy.clone(), + false, + h3_rate_limit.clone(), + true, + h3_readiness.clone(), + h3_keystone.clone(), + h3_compression.clone(), + h3_server_ctx.clone(), + h3_domains_configured + )) + }; + + let (endpoint, task) = http3::spawn( + server_config, + local_addr, + make_service, + max_connections, + http_request_body_read_timeout, + shutdown_tx.subscribe(), + ) + .map_err(|e| Error::other(format!("HTTP/3 startup failed: {e}")))?; + + Some((endpoint, task)) + } else { + None + }; + let tls_reload_task = tls_acceptor.as_ref().and_then(|holder| { + spawn_reload_loop( + tls_path.to_string(), + holder.clone(), + #[cfg(feature = "http3")] + http3_server.as_ref().map(|(endpoint, _)| endpoint.clone()), + ) + }); + let is_console = config.console_enable; let server_domains_configured = !config.server_domains.is_empty(); let task_handle = tokio::spawn(async move { @@ -1550,10 +1802,6 @@ pub async fn start_http_server( rustfs_config::ENV_HTTP1_HEADER_READ_TIMEOUT, rustfs_config::DEFAULT_HTTP1_HEADER_READ_TIMEOUT, ); - let http_request_body_read_timeout = Duration::from_secs(rustfs_utils::get_env_u64( - rustfs_config::ENV_HTTP_REQUEST_BODY_READ_TIMEOUT, - rustfs_config::DEFAULT_HTTP_REQUEST_BODY_READ_TIMEOUT, - )); let http1_max_buf_size = rustfs_utils::get_env_usize(rustfs_config::ENV_HTTP1_MAX_BUF_SIZE, rustfs_config::DEFAULT_HTTP1_MAX_BUF_SIZE); @@ -1791,6 +2039,25 @@ pub async fn start_http_server( ); } } + + if let Some(task) = tls_reload_task { + task.abort(); + let _ = task.await; + } + + #[cfg(feature = "http3")] + if let Some((_, task)) = http3_server + && let Err(error) = task.await + { + error!( + event = EVENT_HTTP_TRANSPORT_FAILED, + component = LOG_COMPONENT_SERVER, + subsystem = LOG_SUBSYSTEM_TRANSPORT, + protocol = "http3", + error = ?error, + "HTTP/3 server task failed during shutdown" + ); + } }); Ok((ShutdownHandle::new(shutdown_tx, task_handle), local_addr)) @@ -2074,123 +2341,19 @@ fn process_connection( // transport/auth/observability subset needed by `/rustfs/rpc/...`. // ───────────────────────────────────────────────────────────── let build_external_stack = |service| { - ServiceBuilder::new() - // NOTE: Both extension types are intentionally inserted to maintain compatibility: - // 1. `Option` - Used by existing admin/storage handlers throughout the codebase - // 2. `std::net::SocketAddr` - Required by TrustedProxyMiddleware for proxy validation - // This dual insertion is necessary because the middleware expects the raw SocketAddr type - // while our application code uses the RemoteAddr wrapper. Consolidating these would - // require either modifying the third-party middleware or refactoring all existing handlers. - .layer(AddExtensionLayer::new(remote_addr)) - .option_layer(remote_addr.map(|ra| AddExtensionLayer::new(ra.0))) - // Add TrustedProxyLayer to handle X-Forwarded-For and other proxy headers - // This should be placed before TraceLayer so that logs reflect the real client IP - // Pre-computed in ConnectionContext to avoid per-connection is_enabled() check. - .option_layer(trusted_proxy_layer.clone()) - .layer(ExternalRequestContextLayer::new(is_console)) - .layer(StsQueryApiCompatLayer) - .layer(EmptyBodyContentLengthCompatLayer) - .layer(CatchPanicLayer::new()) - // Per-client API rate limit (backlog#1191): rejects over-limit - // requests with 429 before readiness/auth/tracing spend any - // work on them, but after the trusted-proxy layer has resolved - // a spoof-proof client IP. Absent (None) unless enabled via - // RUSTFS_API_RATE_LIMIT_ENABLE with a non-zero RPM. - .option_layer(rate_limit_layer.clone()) - // backlog#2369 P7.2: an SSE-C request carries the customer key - // in a header, so a plaintext hop leaks it permanently. Sits - // beside the rate limiter: after the trusted-proxy layer, which - // is what makes a forwarded `https` protocol trustworthy, and - // after the request context so a rejection can echo the request - // id. Reports by default; refuses only under - // RUSTFS_SSE_C_REQUIRE_TLS. - .layer(SsecTransportLayer::new(connection_is_tls)) - // CRITICAL: Insert ReadinessGateLayer before business logic - // This stops requests from hitting IAMAuth or Storage if they are not ready. - .layer(ReadinessGateLayer::new(readiness.clone())) - // Add Keystone authentication middleware - // This validates X-Auth-Token headers and stores credentials in task-local storage - // Must be placed AFTER ReadinessGateLayer but BEFORE business logic - // Pre-computed in ConnectionContext to avoid per-connection OnceLock read. - .layer(KeystoneAuthLayer::new(keystone_auth.clone())) - // Maintain the in-flight request gauge with an RAII guard so it is - // decremented exactly once per request (backlog#806-35). Placed just - // outside TraceLayer so the counting window matches the old on_request - // timing while avoiding the 5xx double-decrement. - .layer(InFlightLayer) - .layer( - TraceLayer::new_for_http() - .make_span_with(make_http_trace_span) - .on_request(|request: &HttpRequest<_>, span: &Span| { - let _enter = span.enter(); - trace!("HTTP request started"); - let method_metrics = http_method_metrics(request.method()); - // In-flight counting is handled by InFlightLayer's RAII guard - // (backlog#806-35); do not adjust the active-requests gauge here. - method_metrics.requests.increment(1); - - if let Some(cl) = request.headers().get("content-length") - && let Some(len) = cl.to_str().ok().and_then(|s| s.parse::().ok()) - { - REQUEST_BODY_BYTES_COUNTER.increment(len); - method_metrics.request_body_size.record(len as f64); - } - }) - .on_response(trace_on_response) - .on_body_chunk(|chunk: &Bytes, latency: Duration, span: &Span| { - RESP_BODY_BYTES_COUNTER.increment(usize_to_u64_saturating(chunk.len())); - record_response_body_chunk_observation(chunk.len(), latency); - #[cfg(feature = "tracing-chunk-debug")] - { - let _enter = span.enter(); - debug!(chunk_bytes = chunk.len(), duration_ms = duration_ms(latency), "HTTP response body chunk sent"); - } - #[cfg(not(feature = "tracing-chunk-debug"))] - { - let _ = (latency, span); - } - }) - .on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, span: &Span| { - record_response_body_stream_duration(stream_duration); - #[cfg(feature = "tracing-chunk-debug")] - { - let _enter = span.enter(); - debug!(duration_ms = duration_ms(stream_duration), "HTTP response stream closed"); - } - #[cfg(not(feature = "tracing-chunk-debug"))] - { - let _ = (_trailers, stream_duration, span); - } - }) - .on_failure(|error, latency: Duration, span: &Span| { - let _enter = span.enter(); - // In-flight counting is handled by InFlightLayer's RAII guard - // (backlog#806-35). This hook previously also fired for 5xx - // responses (which ALSO hit on_response), double-decrementing - // the gauge; only the failure metric is recorded here now. - HTTP_TRANSPORT_FAILURES_COUNTER.increment(1); - record_http_transport_failure_if_body_error(&error); - trace!(error = ?error, duration_ms = duration_ms(latency), "HTTP request failure captured by trace layer"); - }), - ) - .layer(RequestLoggingLayer) - .layer(CompressionLayer::new().compress_when(PathAwareHttpCompressionPredicate::new(compression_config.clone()))) - .option_layer(compression_config.enabled.then_some(PathCategoryInjectionLayer)) - .layer(S3ErrorMessageCompatLayer) - .layer(IcebergRestErrorCompatLayer) - .layer(ObjectAttributesEtagFixLayer) - .layer(ConditionalCorsLayer::new()) - .option_layer(if is_console { Some(RedirectLayer) } else { None }) - .layer(BodylessStatusFixLayer) - .layer(HeadRequestBodyFixLayer) - .layer(PublicHealthEndpointLayer::new( - Arc::clone(&server_ctx), - Arc::clone(&readiness), - )) - .option_layer((!server_domains_configured && !is_console).then_some(VirtualHostStyleHintLayer)) - .layer(DoubleSlashListBucketsCompatLayer) - .layer(SigV4HeaderGuardLayer) - .service(service) + external_service_stack!( + service, + remote_addr, + trusted_proxy_layer.clone(), + is_console, + rate_limit_layer.clone(), + connection_is_tls, + readiness.clone(), + keystone_auth.clone(), + compression_config.clone(), + server_ctx.clone(), + server_domains_configured + ) }; let build_internode_stack = |service| { ServiceBuilder::new() @@ -4245,4 +4408,233 @@ mod tests { assert!(result.is_err(), "ErrService must return an error"); assert_eq!(active_http_requests(), before, "gauge must net to zero on service error"); } + #[cfg(feature = "http3")] + #[tokio::test] + #[serial_test::serial] + async fn http3_external_stack_enforces_readiness_keystone_context_and_rate_limit() -> http3::tests::TestResult { + use crate::server::rate_limit::RateLimitQuota; + use http3::tests::{client_endpoint, configs, connect, request}; + use rustfs_keystone::{KeystoneAuthProvider, KeystoneClient, KeystoneVersion}; + temp_env::async_with_vars([("NO_PROXY", Some("127.0.0.1,localhost")), ("no_proxy", Some("127.0.0.1,localhost"))], async { + let mock = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let auth_url = format!("http://{}", mock.local_addr()?); + let mock_task = tokio::spawn(async move { + let (socket, _) = mock.accept().await.expect("Keystone connection"); + hyper::server::conn::http1::Builder::new().serve_connection(TokioIo::new(socket), hyper::service::service_fn(|request: HttpRequest| async move { + assert_eq!(request.uri().path(), "/v3/auth/tokens"); + let valid = request.headers().get("x-subject-token").is_some_and(|value| value == "valid-token"); + let mut response = Response::new(http_body_util::Full::new(Bytes::from_static(br#"{"token":{"user":{"id":"user-id","name":"test-user"},"expires_at":"2099-01-01T00:00:00Z"}}"#))); + if !valid { *response.status_mut() = http::StatusCode::UNAUTHORIZED; } + response.headers_mut().insert("content-type", HeaderValue::from_static("application/json")); + Ok::<_, std::convert::Infallible>(response) + })).await.expect("Keystone mock"); + }); + let provider = Arc::new(KeystoneAuthProvider::new(KeystoneClient::new(auth_url, KeystoneVersion::V3, + None, None, None, "Default".to_owned(), true, Duration::from_secs(5)), 10, Duration::from_secs(60), false)); + let readiness = Arc::new(GlobalReadiness::new()); + let server_readiness = readiness.clone(); + let rate = RateLimitLayer::new(Some(RateLimitQuota { requests_per_minute: 1, burst: 4 }), None, vec![]); + let (config, cert) = configs()?; + let (shutdown, receiver) = tokio::sync::broadcast::channel(1); + let (endpoint, task) = http3::spawn(config, "127.0.0.1:0".parse()?, move |peer| { + let inner = tower::service_fn(move |request: HttpRequest| async move { + assert_eq!(request.version(), Version::HTTP_3); + assert_eq!(request.extensions().get::>().copied().flatten().expect("peer").0, peer); + let info = request.extensions().get::().expect("policy context"); + assert_eq!(info.real_ip, peer.ip()); + assert_eq!(info.forwarded_proto.as_deref(), Some("https")); + let identity = rustfs_keystone::KEYSTONE_CREDENTIALS.try_with(Clone::clone).ok().flatten().map(|credentials| credentials.parent_user).unwrap_or_else(|| "anonymous".to_owned()); + if request.headers().contains_key("x-auth-token") { assert_eq!(identity, "test-user"); } + Ok::<_, Box>(Response::new(HybridBody::>::Rest { rest_body: s3s::Body::from(identity) })) + }); + let inner = Http3CoreService::new(inner); + Http3ExternalService::new(external_service_stack!(inner, Some(RemoteAddr(peer)), None::, false, + Some(rate.clone()), true, server_readiness.clone(), Some(provider.clone()), + HttpCompressionConfig::default(), ServerContextSlot::new(), false)) + }, 0, Duration::from_secs(1), receiver)?; + let client_endpoint = client_endpoint(vec![cert])?; + let (quic, mut client, driver) = connect(&client_endpoint, endpoint.local_addr()?).await?; + let make_request = || HttpRequest::builder().uri("https://localhost/bucket/key").body(()); + assert_eq!(request(&mut client, make_request()?, Bytes::new()).await?.0, http::StatusCode::SERVICE_UNAVAILABLE); + readiness.mark_stage(rustfs_common::SystemStage::FullReady); + assert_eq!(request(&mut client, HttpRequest::builder().uri("https://localhost/bucket/key").header("x-auth-token", "invalid-token").body(())?, Bytes::new()).await?.0, http::StatusCode::UNAUTHORIZED); + let (status, headers, body) = request(&mut client, HttpRequest::builder().uri("https://localhost/bucket/key").header("x-auth-token", "valid-token").header("x-forwarded-for", "203.0.113.1").body(())?, Bytes::new()).await?; + assert_eq!(status, http::StatusCode::OK); + assert_eq!(body, "test-user"); + assert!(headers.contains_key("x-amz-request-id"), "request context must survive HTTP/3 dispatch"); + assert_eq!(request(&mut client, make_request()?, Bytes::new()).await?.2, "anonymous", "Keystone task-local credentials must not leak between streams"); + assert_eq!(request(&mut client, make_request()?, Bytes::new()).await?.0, http::StatusCode::TOO_MANY_REQUESTS); + shutdown.send(())?; + drop(client); + quic.close(0u32.into(), b"test complete"); + tokio::time::timeout(Duration::from_secs(5), task).await??; + driver.abort(); + mock_task.abort(); + Ok(()) + }).await + } + #[cfg(feature = "http3")] + #[tokio::test] + async fn http3_upload_part_timeout_preserves_s3_error_and_other_streams() -> http3::tests::TestResult { + use bytes::Buf; + use http3::tests::{client_endpoint, configs, connect}; + let (config, cert) = configs()?; + let readiness = Arc::new(GlobalReadiness::new()); + readiness.mark_stage(rustfs_common::SystemStage::FullReady); + let s3 = S3ServiceBuilder::new(UploadPartTimeoutS3 { + timeout: Duration::from_secs(1), + }) + .build(); + let (shutdown, receiver) = tokio::sync::broadcast::channel(1); + let (endpoint, task) = http3::spawn( + config, + "127.0.0.1:0".parse()?, + move |peer| { + let inner = ServiceBuilder::new() + .map_response(|response: Response| { + response.map(|rest_body| HybridBody::>::Rest { rest_body }) + }) + .map_err(|error: s3s::HttpError| -> Box { error.into() }) + .service(EarlyResponseBodyService::new(s3.clone(), Duration::from_secs(1))); + let inner = Http3CoreService::new(inner); + Http3ExternalService::new(external_service_stack!( + inner, + Some(RemoteAddr(peer)), + None::, + false, + None::, + true, + readiness.clone(), + None::>, + HttpCompressionConfig::default(), + ServerContextSlot::new(), + false + )) + }, + 0, + Duration::from_secs(1), + receiver, + )?; + let client_endpoint = client_endpoint(vec![cert])?; + let (quic, mut client, driver) = connect(&client_endpoint, endpoint.local_addr()?).await?; + let mut stalled = client + .send_request( + HttpRequest::builder() + .method("PUT") + .uri("https://localhost/bucket/object?partNumber=1&uploadId=upload") + .header("content-length", 1024) + .body(())?, + ) + .await?; + stalled.send_data(Bytes::from_static(b"x")).await?; + let response = tokio::time::timeout(Duration::from_secs(5), stalled.recv_response()).await??; + assert_eq!(response.status(), http::StatusCode::BAD_REQUEST); + let mut xml = Vec::new(); + while let Some(mut chunk) = stalled.recv_data().await? { + xml.extend_from_slice(&chunk.copy_to_bytes(chunk.remaining())); + } + assert!(String::from_utf8(xml)?.contains("RequestTimeout")); + let mut subsequent = client + .send_request( + HttpRequest::builder() + .method("PUT") + .uri("https://localhost/bucket/object?partNumber=1&uploadId=upload") + .header("content-length", 0) + .body(())?, + ) + .await?; + subsequent.finish().await?; + let response = tokio::time::timeout(Duration::from_secs(5), subsequent.recv_response()).await??; + assert_eq!( + response.status(), + http::StatusCode::OK, + "timed-out uploads must not close unrelated HTTP/3 streams" + ); + while subsequent.recv_data().await?.is_some() {} + drop(stalled); + drop(subsequent); + shutdown.send(())?; + drop(client); + quic.close(0u32.into(), b"test complete"); + tokio::time::timeout(Duration::from_secs(5), task).await??; + driver.abort(); + Ok(()) + } + #[cfg(feature = "http3")] + struct SignedS3; + + #[cfg(feature = "http3")] + #[async_trait::async_trait] + impl s3s::S3 for SignedS3 { + async fn list_buckets( + &self, + req: s3s::S3Request, + ) -> s3s::S3Result> { + assert_eq!(req.credentials.expect("verified SigV4 credentials").access_key, "test-key"); + Ok(s3s::S3Response::new(s3s::dto::ListBucketsOutput { + buckets: Some(vec![s3s::dto::Bucket { + name: Some("http3-bucket".to_owned()), + ..Default::default() + }]), + ..Default::default() + })) + } + } + + #[cfg(feature = "http3")] + #[tokio::test] + async fn http3_s3_service_verifies_sigv4_before_dispatch() -> http3::tests::TestResult { + use http3::tests::{client_endpoint, configs, connect, request}; + use s3s_sigv4::{ + AmzDate, EMPTY_STRING_SHA256_HASH, Payload, calculate_signature, create_canonical_request, create_string_to_sign, + }; + let (config, cert) = configs()?; + let mut builder = s3s::service::S3ServiceBuilder::new(SignedS3); + builder.set_auth(s3s::auth::SimpleAuth::from_single("test-key", "test-secret")); + let service = builder.build(); + let (shutdown, receiver) = tokio::sync::broadcast::channel(1); + let (endpoint, task) = http3::spawn( + config, + "127.0.0.1:0".parse()?, + move |_| service.clone(), + 0, + Duration::from_secs(1), + receiver, + )?; + let endpoint_client = client_endpoint(vec![cert])?; + let (quic, mut client, driver) = connect(&endpoint_client, endpoint.local_addr()?).await?; + let timestamp = chrono::Utc::now().format("%Y%m%dT%H%M%SZ").to_string(); + let date = AmzDate::parse(×tamp)?; + let canonical = create_canonical_request( + "GET", + "/", + &[] as &[(&str, &str)], + [ + ("host", "localhost"), + ("x-amz-content-sha256", EMPTY_STRING_SHA256_HASH), + ("x-amz-date", timestamp.as_str()), + ], + Payload::SingleChunk(EMPTY_STRING_SHA256_HASH), + ); + let string_to_sign = create_string_to_sign(&canonical, &date, "us-east-1", "s3"); + let make_request = |secret| { + let signature = calculate_signature(&string_to_sign, secret, &date, "us-east-1", "s3"); + HttpRequest::builder().uri("https://localhost/").header("host", "localhost") + .header("x-amz-date", ×tamp).header("x-amz-content-sha256", EMPTY_STRING_SHA256_HASH) + .header("authorization", format!("AWS4-HMAC-SHA256 Credential=test-key/{}/us-east-1/s3/aws4_request, SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature={signature}", date.fmt_date())) + .body(()) + }; + let (status, _, body) = request(&mut client, make_request("wrong-secret")?, Bytes::new()).await?; + assert_eq!(status, http::StatusCode::FORBIDDEN); + assert!(String::from_utf8(body.to_vec())?.contains("SignatureDoesNotMatch")); + let (status, _, body) = request(&mut client, make_request("test-secret")?, Bytes::new()).await?; + assert_eq!(status, http::StatusCode::OK); + assert!(String::from_utf8(body.to_vec())?.contains("http3-bucket")); + shutdown.send(())?; + drop(client); + quic.close(0u32.into(), b"test complete"); + tokio::time::timeout(Duration::from_secs(5), task).await??; + driver.abort(); + Ok(()) + } } diff --git a/rustfs/src/server/http3.rs b/rustfs/src/server/http3.rs new file mode 100644 index 000000000..0b204bd11 --- /dev/null +++ b/rustfs/src/server/http3.rs @@ -0,0 +1,644 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: 2023-2026 The s3s Authors +// Copyright 2026 RustFS Team + +use bytes::{Buf, Bytes}; +use h3::error::Code; +use h3::server::{RequestResolver, RequestStream}; +use http::{HeaderMap, HeaderName, Request, Response, Version, header}; +use http_body::{Body as HttpBody, Frame, SizeHint}; +use http_body_util::BodyExt; +use quinn::{Endpoint, VarInt}; +use std::io; +use std::net::SocketAddr; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; +use std::time::Duration; +use tokio::sync::{Semaphore, broadcast}; +use tokio::task::{JoinHandle, JoinSet}; +use tokio_util::sync::CancellationToken; +use tower::{Service, ServiceExt}; + +const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10); + +pub(crate) type RequestBody = http_body_util::combinators::UnsyncBoxBody; + +type Resolver = RequestResolver; + +pub(crate) fn spawn( + mut server_config: quinn::ServerConfig, + bind_addr: SocketAddr, + make_service: M, + max_connections: usize, + body_timeout: Duration, + mut shutdown: broadcast::Receiver<()>, +) -> io::Result<(Endpoint, JoinHandle<()>)> +where + M: Fn(SocketAddr) -> S + Clone + Send + 'static, + S: Service, Response = Response, Error = E> + Clone + Send + 'static, + S::Future: Send + 'static, + B: HttpBody + Send + 'static, + B::Error: Send, + E: Send + 'static, +{ + // The service's peer address must remain valid for the connection lifetime. + server_config.migration(false); + let endpoint = Endpoint::server(server_config, bind_addr)?; + let server_endpoint = endpoint.clone(); + let limiter = (max_connections > 0).then(|| Arc::new(Semaphore::new(max_connections.min(Semaphore::MAX_PERMITS)))); + let task = tokio::spawn(async move { + let cancellation = CancellationToken::new(); + let mut connections = JoinSet::new(); + loop { + tokio::select! { + biased; + _ = shutdown.recv() => break, + _ = connections.join_next(), if !connections.is_empty() => {}, + incoming = server_endpoint.accept() => { + let Some(incoming) = incoming else { break }; + let permit = match &limiter { + Some(limiter) => match limiter.clone().try_acquire_owned() { + Ok(permit) => Some(permit), + Err(_) => { incoming.refuse(); continue; } + }, + None => None, + }; + let make_service = make_service.clone(); + let cancellation = cancellation.child_token(); + connections.spawn(async move { + let _permit = permit; + let connection = tokio::select! { + result = incoming => match result { Ok(connection) => connection, Err(_) => return }, + _ = cancellation.cancelled() => return, + }; + let service = make_service(connection.remote_address()); + let _ = serve_connection(connection, service, body_timeout, cancellation).await; + }); + } + } + } + server_endpoint.set_server_config(None); + cancellation.cancel(); + let drain = async { while connections.join_next().await.is_some() {} }; + let _ = tokio::time::timeout(SHUTDOWN_TIMEOUT, drain).await; + server_endpoint.close(VarInt::from_u32(0), b"server shutdown"); + connections.abort_all(); + while connections.join_next().await.is_some() {} + }); + Ok((endpoint, task)) +} + +async fn serve_connection( + quic: quinn::Connection, + service: S, + body_timeout: Duration, + cancellation: CancellationToken, +) -> std::result::Result<(), h3::error::ConnectionError> +where + S: Service, Response = Response, Error = E> + Clone + Send + 'static, + S::Future: Send + 'static, + B: HttpBody + Send + 'static, + B::Error: Send, + E: Send + 'static, +{ + let mut connection = h3::server::builder() + .max_field_section_size(u64::from(rustfs_config::DEFAULT_H2_MAX_HEADER_LIST_SIZE)) + .build(h3_quinn::Connection::new(quic.clone())) + .await?; + let mut requests = JoinSet::new(); + let mut shutting_down = false; + loop { + tokio::select! { + biased; + _ = cancellation.cancelled(), if !shutting_down => { + shutting_down = true; + connection.shutdown(0).await?; + }, + _ = requests.join_next(), if !requests.is_empty() => {}, + result = connection.accept() => match result? { + Some(resolver) => { requests.spawn(handle_request(resolver, service.clone(), body_timeout)); } + None => break, + }, + } + } + while requests.join_next().await.is_some() {} + // finish() queues bytes; keep QUIC alive until the peer has consumed them. + // The listener's shutdown deadline bounds peers that never close. + let _ = quic.closed().await; + Ok(()) +} + +async fn handle_request(resolver: Resolver, service: S, body_timeout: Duration) +where + S: Service, Response = Response, Error = E> + Send, + S::Future: Send, + B: HttpBody + Send, + B::Error: Send, +{ + let Ok((request, stream)) = resolver.resolve_request().await else { return }; + let (mut send, recv) = stream.split(); + if disallowed_request_field(request.headers()).is_some() { + send.stop_stream(Code::H3_MESSAGE_ERROR); + return; + } + if super::has_path_prefix(request.uri().path(), super::RPC_PREFIX) + || request + .headers() + .get(header::CONTENT_TYPE) + .is_some_and(|value| value.as_bytes().starts_with(b"application/grpc")) + { + let response = Response::builder() + .status(http::StatusCode::NOT_IMPLEMENTED) + .body(()) + .expect("static response"); + if send.send_response(response).await.is_ok() { + let _ = send + .send_data(Bytes::from_static(b"Use the TCP listener for internode RPC")) + .await; + let _ = send.finish().await; + } + return; + } + let content_length = match request.headers().get(header::CONTENT_LENGTH) { + Some(value) => match value.to_str().ok().and_then(|value| value.parse::().ok()) { + Some(length) => Some(length), + None => { + send.stop_stream(Code::H3_MESSAGE_ERROR); + return; + } + }, + None => None, + }; + let request = request.map(|()| Body::new(recv, content_length, body_timeout).boxed_unsync()); + let Ok(response) = service.oneshot(request).await else { + send.stop_stream(Code::H3_INTERNAL_ERROR); + return; + }; + let (mut parts, body) = response.into_parts(); + strip_hop_by_hop_headers(&mut parts.headers); + parts.version = Version::HTTP_3; + if send.send_response(Response::from_parts(parts, ())).await.is_err() { + return; + } + let mut body = std::pin::pin!(body); + while let Some(frame) = std::future::poll_fn(|cx| body.as_mut().poll_frame(cx)).await { + let Ok(frame) = frame else { + send.stop_stream(Code::H3_INTERNAL_ERROR); + return; + }; + match frame.into_data() { + Ok(data) => { + if send.send_data(data).await.is_err() { + return; + } + } + Err(frame) => { + if let Ok(mut trailers) = frame.into_trailers() { + strip_hop_by_hop_headers(&mut trailers); + if send.send_trailers(trailers).await.is_err() { + return; + } + break; + } + } + } + } + let _ = send.finish().await; +} + +type RecvStream = RequestStream; + +enum State { + Data, + Trailers, + Done, +} + +// Dropping an unread body cancels the receive stream without a detached drain. +struct Body { + stream: Option, + state: State, + expected_length: Option, + received_length: u64, + timeout: Duration, + timer: Option>>, +} + +impl Body { + fn new(stream: RecvStream, expected_length: Option, timeout: Duration) -> Self { + Self { + stream: Some(stream), + state: State::Data, + expected_length, + received_length: 0, + timeout, + timer: None, + } + } + + fn pending(&mut self, cx: &mut Context<'_>) -> Poll>>> { + if self.timeout.is_zero() { + return Poll::Pending; + } + let timer = self.timer.get_or_insert_with(|| Box::pin(tokio::time::sleep(self.timeout))); + if std::future::Future::poll(timer.as_mut(), cx).is_pending() { + return Poll::Pending; + } + // h3-quinn moves the receive stream into its pending read future. + // Dropping it cancels that read and sends STOP_SENDING without panic. + drop(self.stream.take()); + self.state = State::Done; + Poll::Ready(Some(Err(io::Error::new( + io::ErrorKind::TimedOut, + crate::error::ClientBodyReadTimeout { + timeout: self.timeout, + raw_bytes_received: self.received_length, + }, + )))) + } +} + +fn content_length_error(expected: u64, actual: u64) -> io::Error { + io::Error::new( + std::io::ErrorKind::InvalidData, + format!("HTTP/3 request body length mismatch: expected {expected}, received {actual}"), + ) +} + +impl HttpBody for Body { + type Data = Bytes; + type Error = io::Error; + + fn poll_frame(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll, Self::Error>>> { + let this = self.get_mut(); + + loop { + match this.state { + State::Data => { + let stream = this.stream.as_mut().expect("data state owns the receive stream"); + + match stream.poll_recv_data(cx) { + Poll::Pending => return this.pending(cx), + Poll::Ready(Err(error)) => { + this.state = State::Done; + return Poll::Ready(Some(Err(io::Error::other(error)))); + } + Poll::Ready(Ok(Some(mut data))) => { + let data = data.copy_to_bytes(data.remaining()); + if !data.is_empty() { + this.timer = None; + } + let Some(received) = this + .received_length + .checked_add(u64::try_from(data.len()).expect("Bytes length fits in u64")) + else { + stream.stop_sending(Code::H3_MESSAGE_ERROR); + this.state = State::Done; + return Poll::Ready(Some(Err(io::Error::new( + io::ErrorKind::InvalidData, + "HTTP/3 body length overflow", + )))); + }; + if let Some(expected) = this.expected_length + && received > expected + { + stream.stop_sending(Code::H3_MESSAGE_ERROR); + this.state = State::Done; + return Poll::Ready(Some(Err(content_length_error(expected, received)))); + } + this.received_length = received; + return Poll::Ready(Some(Ok(Frame::data(data)))); + } + Poll::Ready(Ok(None)) => { + if let Some(expected) = this.expected_length + && this.received_length != expected + { + stream.stop_sending(Code::H3_MESSAGE_ERROR); + this.state = State::Done; + return Poll::Ready(Some(Err(content_length_error(expected, this.received_length)))); + } + this.timer = None; + this.state = State::Trailers; + } + } + } + State::Trailers => { + let stream = this.stream.as_mut().expect("trailers state owns the receive stream"); + + match stream.poll_recv_trailers(cx) { + Poll::Pending => return this.pending(cx), + Poll::Ready(Err(error)) => { + this.state = State::Done; + return Poll::Ready(Some(Err(io::Error::other(error)))); + } + Poll::Ready(Ok(Some(trailers))) => { + this.state = State::Done; + return Poll::Ready(Some(Ok(Frame::trailers(trailers)))); + } + Poll::Ready(Ok(None)) => { + this.state = State::Done; + return Poll::Ready(None); + } + } + } + State::Done => return Poll::Ready(None), + } + } + } + + fn is_end_stream(&self) -> bool { + matches!(self.state, State::Done) + } + + fn size_hint(&self) -> SizeHint { + SizeHint::default() + } +} + +fn disallowed_request_field(headers: &HeaderMap) -> Option<&'static str> { + for name in ["connection", "transfer-encoding", "upgrade", "keep-alive", "proxy-connection"] { + if headers.contains_key(name) { + return Some(name); + } + } + + if headers + .get_all(header::TE) + .iter() + .any(|value| !trim_ows(value.as_bytes()).eq_ignore_ascii_case(b"trailers")) + { + return Some("te"); + } + + None +} + +/// Removes the optional whitespace (SP / HTAB) around a field value, which +/// RFC 9110 §5.5 requires before the value is evaluated. Other whitespace +/// characters are left in place: CR, LF, and NUL make a field value invalid +/// rather than equivalent. +fn trim_ows(value: &[u8]) -> &[u8] { + let start = value + .iter() + .position(|byte| !matches!(*byte, b' ' | b'\t')) + .unwrap_or(value.len()); + let end = value + .iter() + .rposition(|byte| !matches!(*byte, b' ' | b'\t')) + .map_or(start, |index| index + 1); + + value.get(start..end).unwrap_or_default() +} + +fn strip_hop_by_hop_headers(headers: &mut HeaderMap) { + let connection_headers = headers + .get_all(header::CONNECTION) + .iter() + .filter_map(|value| value.to_str().ok()) + .flat_map(|value| value.split(',')) + .filter_map(|name| HeaderName::from_bytes(name.trim().as_bytes()).ok()) + .collect::>(); + + for name in connection_headers { + headers.remove(name); + } + + for name in [ + header::CONNECTION, + header::PROXY_AUTHENTICATE, + header::PROXY_AUTHORIZATION, + header::TE, + header::TRAILER, + header::TRANSFER_ENCODING, + header::UPGRADE, + HeaderName::from_static("keep-alive"), + ] { + headers.remove(name); + } +} + +#[cfg(test)] +pub(super) mod tests { + use super::*; + use http_body_util::BodyExt; + use rustls::pki_types::{CertificateDer, PrivateKeyDer}; + use std::convert::Infallible; + + pub(crate) type TestResult = std::result::Result>; + pub(crate) type Client = h3::client::SendRequest; + + pub(crate) fn configs() -> TestResult<(quinn::ServerConfig, CertificateDer<'static>)> { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()])?; + let der = cert.cert.der().clone(); + let mut tls = rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(vec![der.clone()], PrivateKeyDer::try_from(cert.signing_key.serialize_der())?)?; + tls.alpn_protocols = vec![b"h3".to_vec()]; + Ok(( + quinn::ServerConfig::with_crypto(Arc::new(quinn::crypto::rustls::QuicServerConfig::try_from(tls)?)), + der, + )) + } + + pub(crate) fn client_endpoint(certs: Vec>) -> TestResult { + let mut roots = rustls::RootCertStore::empty(); + for cert in certs { + roots.add(cert)?; + } + let mut tls = rustls::ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(); + tls.alpn_protocols = vec![b"h3".to_vec()]; + let mut endpoint = Endpoint::client("127.0.0.1:0".parse()?)?; + endpoint.set_default_client_config(quinn::ClientConfig::new(Arc::new( + quinn::crypto::rustls::QuicClientConfig::try_from(tls)?, + ))); + Ok(endpoint) + } + + pub(crate) async fn connect( + endpoint: &Endpoint, + addr: SocketAddr, + ) -> TestResult<(quinn::Connection, Client, JoinHandle<()>)> { + let quic = tokio::time::timeout(Duration::from_secs(5), endpoint.connect(addr, "localhost")?).await??; + let (mut driver, client) = h3::client::builder().build(h3_quinn::Connection::new(quic.clone())).await?; + let task = tokio::spawn(async move { + let _ = std::future::poll_fn(|cx| driver.poll_close(cx)).await; + }); + Ok((quic, client, task)) + } + + pub(crate) async fn request( + client: &mut Client, + request: Request<()>, + data: Bytes, + ) -> TestResult<(http::StatusCode, HeaderMap, Bytes)> { + let mut stream = client.send_request(request).await?; + if !data.is_empty() { + stream.send_data(data).await?; + } + stream.finish().await?; + let response = tokio::time::timeout(Duration::from_secs(5), stream.recv_response()).await??; + let mut body = Vec::new(); + while let Some(mut chunk) = stream.recv_data().await? { + body.extend_from_slice(&chunk.copy_to_bytes(chunk.remaining())); + } + Ok((response.status(), response.headers().clone(), Bytes::from(body))) + } + + #[tokio::test] + async fn http3_bounds_connections_and_releases_permits() -> TestResult { + let (config, cert) = configs()?; + let (shutdown, receiver) = broadcast::channel(1); + let (endpoint, task) = spawn( + config, + "127.0.0.1:0".parse()?, + |_| { + tower::service_fn(|_: Request| async { + Ok::<_, Infallible>(Response::new(http_body_util::Full::new(Bytes::from_static(b"bounded")))) + }) + }, + 1, + Duration::from_secs(1), + receiver, + )?; + let client_endpoint = client_endpoint(vec![cert])?; + let addr = endpoint.local_addr()?; + let (first, mut client, driver) = connect(&client_endpoint, addr).await?; + assert_eq!( + request(&mut client, Request::builder().uri("https://localhost/").body(())?, Bytes::new()) + .await? + .2, + "bounded" + ); + let rejected = tokio::time::timeout(Duration::from_secs(5), client_endpoint.connect(addr, "localhost")?).await?; + assert!(rejected.is_err(), "a second QUIC connection must be refused at the configured cap"); + first.close(0u32.into(), b"release permit"); + drop(client); + driver.abort(); + tokio::time::timeout(Duration::from_secs(5), async { + while endpoint.open_connections() > 0 { + tokio::task::yield_now().await; + } + }) + .await?; + // Quinn releases the connection before its serving task releases the permit. + let (second, mut client, driver) = tokio::time::timeout(Duration::from_secs(5), async { + loop { + if let Ok(connection) = connect(&client_endpoint, addr).await { + break connection; + } + tokio::task::yield_now().await; + } + }) + .await?; + assert_eq!( + request(&mut client, Request::builder().uri("https://localhost/").body(())?, Bytes::new()) + .await? + .0, + http::StatusCode::OK + ); + shutdown.send(())?; + drop(client); + second.close(0u32.into(), b"test complete"); + tokio::time::timeout(Duration::from_secs(5), task).await??; + assert!( + !matches!( + tokio::time::timeout(Duration::from_millis(250), client_endpoint.connect(addr, "localhost")?).await, + Ok(Ok(_)) + ), + "shutdown must stop new handshakes" + ); + driver.abort(); + client_endpoint.close(0u32.into(), b"test complete"); + Ok(()) + } + + #[tokio::test] + async fn http3_streams_body_and_times_out_stalled_upload() -> TestResult { + let (config, cert) = configs()?; + let (shutdown, receiver) = broadcast::channel(1); + let (endpoint, task) = spawn( + config, + "127.0.0.1:0".parse()?, + |_| { + tower::service_fn(|request: Request| async { + let mut response = match request.into_body().collect().await { + Ok(body) => Response::new(http_body_util::Full::new(body.to_bytes())), + Err(_) => { + let mut response = Response::new(http_body_util::Full::new(Bytes::new())); + *response.status_mut() = http::StatusCode::REQUEST_TIMEOUT; + response + } + }; + response + .headers_mut() + .insert(header::CONNECTION, http::HeaderValue::from_static("close")); + Ok::<_, Infallible>(response) + }) + }, + 0, + Duration::from_secs(1), + receiver, + )?; + let client_endpoint = client_endpoint(vec![cert])?; + let (quic, mut client, driver) = connect(&client_endpoint, endpoint.local_addr()?).await?; + let (status, _, body) = request( + &mut client, + Request::builder() + .method("POST") + .uri("https://localhost/node_service.NodeService/Ping") + .header(header::CONTENT_TYPE, "application/grpc") + .body(())?, + Bytes::new(), + ) + .await?; + assert_eq!(status, http::StatusCode::NOT_IMPLEMENTED); + assert_eq!(body, "Use the TCP listener for internode RPC"); + let payload = Bytes::from(vec![b'x'; 256 * 1024]); + let (status, headers, body) = request( + &mut client, + Request::builder() + .method("PUT") + .uri("https://localhost/bucket/key") + .header(header::CONTENT_LENGTH, payload.len()) + .body(())?, + payload.clone(), + ) + .await?; + assert_eq!(status, http::StatusCode::OK); + assert_eq!(body, payload, "the full streamed body must survive transport adaptation"); + assert!(!headers.contains_key(header::CONNECTION)); + let mut stalled = client + .send_request( + Request::builder() + .method("PUT") + .uri("https://localhost/bucket/key") + .header(header::CONTENT_LENGTH, 10) + .body(())?, + ) + .await?; + stalled.send_data(Bytes::from_static(b"x")).await?; + let response = tokio::time::timeout(Duration::from_secs(5), stalled.recv_response()).await??; + assert_eq!(response.status(), http::StatusCode::REQUEST_TIMEOUT); + drop(stalled); + let (_, _, body) = request( + &mut client, + Request::builder() + .method("PUT") + .uri("https://localhost/bucket/key") + .body(())?, + Bytes::from_static(b"after timeout"), + ) + .await?; + assert_eq!(body, "after timeout", "a canceled upload must leave the QUIC connection usable"); + shutdown.send(())?; + drop(client); + quic.close(0u32.into(), b"test complete"); + tokio::time::timeout(Duration::from_secs(5), task).await??; + driver.abort(); + Ok(()) + } +} diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index fc04368ed..c6657cddf 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -34,6 +34,7 @@ use futures::future::Either; use http::{HeaderMap, HeaderValue, Method, Request as HttpRequest, Response, StatusCode, Uri}; use http_body::Body; use http_body_util::{BodyExt, Full}; +#[cfg(test)] use hyper::body::Incoming; use pin_project_lite::pin_project; use quick_xml::events::Event; @@ -639,11 +640,12 @@ fn is_console_redirect_request(req: &HttpRequest) -> bool { && (path.is_empty() || path == "/rustfs" || path == "/index.html") } -impl Service> for RedirectService +impl Service> for RedirectService where - S: Service, Response = Response>> + Clone + Send + 'static, + S: Service, Response = Response>> + Clone + Send + 'static, S::Future: Send + 'static, S::Error: Into> + Send + 'static, + ReqBody: Send + 'static, RestBody: Default + Send + 'static, GrpcBody: Send + 'static, { @@ -655,7 +657,7 @@ where self.inner.poll_ready(cx).map_err(Into::into) } - fn call(&mut self, req: HttpRequest) -> Self::Future { + fn call(&mut self, req: HttpRequest) -> Self::Future { let path = req.uri().path().trim_end_matches('/'); if is_console_redirect_request(&req) { debug!("Redirecting browser request from {} to console", path); diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index 7d79791ef..7f919d6f7 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -30,6 +30,9 @@ mod service_state; mod ssec_transport; pub mod tls_material; +#[cfg(feature = "http3")] +mod http3; + use tracing::warn; // Items used by main.rs (binary crate) and/or embedded.rs — must be fully pub. diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index 56fa9b885..712522e1a 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -27,7 +27,6 @@ use http::HeaderValue; use http::{Request as HttpRequest, Response, StatusCode}; use http_body::Body; use http_body_util::{BodyExt, Full}; -use hyper::body::Incoming; use metrics::{counter, gauge}; use rustfs_common::GlobalReadiness; use rustfs_madmin::{BackendInfo, Disk, StorageInfo}; @@ -192,11 +191,12 @@ fn service_not_ready_response(stage: rustfs_common::SystemStage) -> Response Service> for ReadinessGateService +impl Service> for ReadinessGateService where - S: Service, Response = Response> + Clone + Send + 'static, + S: Service, Response = Response> + Clone + Send + 'static, S::Future: Send + 'static, S::Error: Send + 'static, + ReqBody: Send + 'static, B: Body + Send + 'static, B::Error: Into + Send + 'static, { @@ -208,7 +208,7 @@ where self.inner.poll_ready(cx) } - fn call(&mut self, req: HttpRequest) -> Self::Future { + fn call(&mut self, req: HttpRequest) -> Self::Future { let mut inner = self.inner.clone(); let readiness = self.readiness.clone(); Box::pin(async move { diff --git a/rustfs/src/server/tls_material.rs b/rustfs/src/server/tls_material.rs index 01329ae38..4d6239eb2 100644 --- a/rustfs/src/server/tls_material.rs +++ b/rustfs/src/server/tls_material.rs @@ -102,7 +102,7 @@ pub(crate) async fn build_acceptor_from_loaded( match server { Some(RuntimeServerTlsMaterial::SingleCert { certs, key }) => { - let config = build_server_config(ServerCertSource::SingleCert { certs, key }, mtls_verifier)?; + let config = build_server_config(ServerCertSource::SingleCert { certs, key }, mtls_verifier, ServerProtocol::Tcp)?; info!( component = LOG_COMPONENT_TLS, subsystem = LOG_SUBSYSTEM_TLS, @@ -116,7 +116,7 @@ pub(crate) async fn build_acceptor_from_loaded( Some(RuntimeServerTlsMaterial::MultiCert { cert_key_pairs }) => { let resolver = create_multi_cert_resolver(cert_key_pairs) .map_err(|e| TlsMaterialError::Parse(format!("build multi-cert resolver: {e}")))?; - let config = build_server_config(ServerCertSource::Resolver(Arc::new(resolver)), mtls_verifier)?; + let config = build_server_config(ServerCertSource::Resolver(Arc::new(resolver)), mtls_verifier, ServerProtocol::Tcp)?; info!( component = LOG_COMPONENT_TLS, subsystem = LOG_SUBSYSTEM_TLS, @@ -131,6 +131,47 @@ pub(crate) async fn build_acceptor_from_loaded( } } +#[cfg(feature = "http3")] +pub(crate) async fn build_http3_server_config_from_loaded( + server: Option<&RuntimeServerTlsMaterial>, + tls_dir: &Path, +) -> Result, TlsMaterialError> { + let Some(server) = server else { + return Ok(None); + }; + + let mtls_verifier = build_webpki_client_verifier( + WebPkiClientVerifierOptions::builder(tls_dir, RUSTFS_CLIENT_CA_CERT_FILENAME, RUSTFS_CA_CERT) + .enabled(get_env_bool(ENV_SERVER_MTLS_ENABLE, DEFAULT_SERVER_MTLS_ENABLE)) + .build(), + ) + .map_err(|e| TlsMaterialError::Io(format!("build mTLS verifier: {e}")))?; + + let source = match server { + RuntimeServerTlsMaterial::SingleCert { certs, key } => ServerCertSource::SingleCert { + certs: certs.clone(), + key: key.clone_key(), + }, + RuntimeServerTlsMaterial::MultiCert { cert_key_pairs } => { + let pairs = cert_key_pairs + .iter() + .map(|(domain, (certs, key))| (domain.clone(), (certs.clone(), key.clone_key()))) + .collect(); + + let resolver = create_multi_cert_resolver(pairs) + .map_err(|e| TlsMaterialError::Parse(format!("build HTTP/3 SNI resolver: {e}")))?; + + ServerCertSource::Resolver(Arc::new(resolver)) + } + }; + + let tls = build_server_config(source, mtls_verifier, ServerProtocol::Http3)?; + let crypto = quinn::crypto::rustls::QuicServerConfig::try_from(tls) + .map_err(|e| TlsMaterialError::Io(format!("configure QUIC TLS: {e}")))?; + + Ok(Some(quinn::ServerConfig::with_crypto(Arc::new(crypto)))) +} + // ── Outbound Enrichment ── /// Enrich the outbound TLS material with platform-specific additions that the @@ -422,30 +463,39 @@ enum ServerCertSource { }, } +enum ServerProtocol { + Tcp, + #[cfg(feature = "http3")] + Http3, +} + fn build_server_config( cert_source: ServerCertSource, mtls_verifier: Option>, + protocol: ServerProtocol, ) -> Result { + let builder = match protocol { + ServerProtocol::Tcp => rustls::ServerConfig::builder(), + #[cfg(feature = "http3")] + ServerProtocol::Http3 => rustls::ServerConfig::builder_with_protocol_versions(&[&rustls::version::TLS13]), + }; + let mut config = match cert_source { ServerCertSource::Resolver(resolver) => { if let Some(verifier) = mtls_verifier { - rustls::ServerConfig::builder() - .with_client_cert_verifier(verifier) - .with_cert_resolver(resolver) + builder.with_client_cert_verifier(verifier).with_cert_resolver(resolver) } else { - rustls::ServerConfig::builder() - .with_no_client_auth() - .with_cert_resolver(resolver) + builder.with_no_client_auth().with_cert_resolver(resolver) } } ServerCertSource::SingleCert { certs, key } => { if let Some(verifier) = mtls_verifier { - rustls::ServerConfig::builder() + builder .with_client_cert_verifier(verifier) .with_single_cert(certs, key) .map_err(|e| TlsMaterialError::Io(format!("configure single cert with mTLS: {e}")))? } else { - rustls::ServerConfig::builder() + builder .with_no_client_auth() .with_single_cert(certs, key) .map_err(|e| TlsMaterialError::Io(format!("configure single cert: {e}")))? @@ -453,7 +503,19 @@ fn build_server_config( } }; - config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()]; + config.alpn_protocols = match protocol { + ServerProtocol::Tcp => { + vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()] + } + #[cfg(feature = "http3")] + ServerProtocol::Http3 => vec![b"h3".to_vec()], + }; + + #[cfg(feature = "http3")] + if matches!(protocol, ServerProtocol::Http3) { + config.max_early_data_size = 0; + } + config.session_storage = rustls::server::ServerSessionMemoryCache::new(10000); if tls_key_log() { @@ -584,12 +646,45 @@ impl TlsAcceptorHolder { } } +// Build both configurations before publishing either listener's new certificate. +async fn reload_acceptors( + server: Option, + tls_dir: &Path, + holder: &TlsAcceptorHolder, + #[cfg(feature = "http3")] http3_endpoint: Option<&quinn::Endpoint>, +) -> Result { + #[cfg(feature = "http3")] + let http3_config = if http3_endpoint.is_some() { + let config = build_http3_server_config_from_loaded(server.as_ref(), tls_dir).await?; + let Some(mut config) = config else { + return Ok(false); + }; + config.migration(false); + Some(config) + } else { + None + }; + let Some(new_holder) = build_acceptor_from_loaded(server, tls_dir).await? else { + return Ok(false); + }; + #[cfg(feature = "http3")] + if let (Some(endpoint), Some(config)) = (http3_endpoint, http3_config) { + endpoint.set_server_config(Some(config)); + } + holder.swap(&new_holder); + Ok(true) +} + // ── Reload Loop ── /// Spawn a background task that periodically checks for TLS certificate changes. /// Single load per tick: loads once via tls-runtime, enriches, publishes outbound, /// and builds acceptor — no double reads. -pub(crate) fn spawn_reload_loop(tls_path: String, holder: Arc) { +pub(crate) fn spawn_reload_loop( + tls_path: String, + holder: Arc, + #[cfg(feature = "http3")] http3_endpoint: Option, +) -> Option> { let enabled = get_env_bool(ENV_TLS_RELOAD_ENABLE, DEFAULT_TLS_RELOAD_ENABLE); if !enabled { debug!( @@ -600,7 +695,7 @@ pub(crate) fn spawn_reload_loop(tls_path: String, holder: Arc env_var = ENV_TLS_RELOAD_ENABLE, "TLS reload state changed" ); - return; + return None; } let interval_secs = rustfs_utils::get_env_u64(ENV_TLS_RELOAD_INTERVAL, DEFAULT_TLS_RELOAD_INTERVAL).max(5); @@ -615,7 +710,7 @@ pub(crate) fn spawn_reload_loop(tls_path: String, holder: Arc "TLS reload state changed" ); - tokio::spawn(async move { + Some(tokio::spawn(async move { let mut interval = tokio::time::interval(Duration::from_secs(interval_secs)); let tls_dir = PathBuf::from(&tls_path); loop { @@ -650,8 +745,16 @@ pub(crate) fn spawn_reload_loop(tls_path: String, holder: Arc ); } - match build_acceptor_from_loaded(snapshot.server, &tls_dir).await { - Ok(Some(new_holder)) => { + match reload_acceptors( + snapshot.server, + &tls_dir, + &holder, + #[cfg(feature = "http3")] + http3_endpoint.as_ref(), + ) + .await + { + Ok(true) => { info!( component = LOG_COMPONENT_TLS, subsystem = LOG_SUBSYSTEM_TLS, @@ -660,7 +763,6 @@ pub(crate) fn spawn_reload_loop(tls_path: String, holder: Arc generation, "TLS reload state changed" ); - holder.swap(&new_holder); startup_runtime_sources::record_tls_reload_result( "rustfs_server_reload_loop", "ok", @@ -668,7 +770,7 @@ pub(crate) fn spawn_reload_loop(tls_path: String, holder: Arc Some(generation), ); } - Ok(None) => { + Ok(false) => { startup_runtime_sources::record_tls_reload_skipped("rustfs_server_reload_loop", "no_acceptor"); warn!( component = LOG_COMPONENT_TLS, @@ -711,7 +813,7 @@ pub(crate) fn spawn_reload_loop(tls_path: String, holder: Arc } } } - }); + })) } #[cfg(test)] @@ -961,4 +1063,97 @@ mod tests { assert!(acceptor.is_some()); } + + #[cfg(feature = "http3")] + #[test] + fn http3_server_config_uses_h3_and_disables_early_data() { + ensure_rustls_crypto_provider(); + + let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).expect("certificate should generate"); + let config = build_server_config( + ServerCertSource::SingleCert { + certs: vec![cert.cert.der().clone()], + key: rustls::pki_types::PrivateKeyDer::try_from(cert.signing_key.serialize_der()) + .expect("private key should convert"), + }, + None, + ServerProtocol::Http3, + ) + .expect("HTTP/3 TLS config should build"); + + assert_eq!(config.alpn_protocols, vec![b"h3".to_vec()]); + assert_eq!(config.max_early_data_size, 0); + } + #[cfg(feature = "http3")] + #[tokio::test] + #[serial_test::serial] + async fn http3_certificate_reload_updates_handshakes_and_retains_valid_config() { + use crate::server::http3::tests::{client_endpoint, connect}; + let dir = TempDir::new().expect("TLS directory"); + write_test_cert_pair(dir.path(), "localhost"); + let snapshot = load_tls_material(dir.path().to_str().expect("TLS path")) + .await + .expect("load first cert"); + let holder = build_acceptor_from_loaded( + load_tls_material(dir.path().to_str().expect("TLS path")) + .await + .expect("load TCP cert") + .server, + dir.path(), + ) + .await + .expect("TCP config") + .expect("TCP holder"); + let config = build_http3_server_config_from_loaded(snapshot.server.as_ref(), dir.path()) + .await + .expect("QUIC config") + .expect("server cert"); + let endpoint = quinn::Endpoint::server(config, "127.0.0.1:0".parse().expect("address")).expect("QUIC endpoint"); + let first_cert = match snapshot.server.expect("cert") { + RuntimeServerTlsMaterial::SingleCert { certs, .. } => certs[0].clone(), + _ => panic!("single certificate expected"), + }; + write_test_cert_pair(dir.path(), "localhost"); + let snapshot = load_tls_material(dir.path().to_str().expect("TLS path")) + .await + .expect("load rotated cert"); + let second_cert = match snapshot.server.as_ref().expect("cert") { + RuntimeServerTlsMaterial::SingleCert { certs, .. } => certs[0].clone(), + _ => panic!("single certificate expected"), + }; + assert_ne!(first_cert, second_cert); + let old_tcp = holder.get(); + assert!( + reload_acceptors(snapshot.server, dir.path(), &holder, Some(&endpoint)) + .await + .expect("reload both listeners") + ); + assert!(!Arc::ptr_eq(&old_tcp, &holder.get()), "TCP acceptor must also rotate"); + let valid_tcp = holder.get(); + assert!( + !reload_acceptors(None, dir.path(), &holder, Some(&endpoint)) + .await + .expect("empty snapshot") + ); + assert!(Arc::ptr_eq(&valid_tcp, &holder.get()), "failed reload must retain the valid TCP acceptor"); + let client = client_endpoint(vec![second_cert.clone()]).expect("trust rotated cert"); + let addr = endpoint.local_addr().expect("server address"); + let handshake = tokio::spawn(async move { connect(&client, addr).await }); + let incoming = endpoint.accept().await.expect("new connection"); + let server = incoming.await.expect("rotated TLS handshake"); + let (client, _, driver) = handshake.await.expect("client task").expect("client trusts new certificate"); + let identity = client + .peer_identity() + .expect("peer certs") + .downcast::>>() + .expect("rustls identity"); + assert_eq!( + identity[0], second_cert, + "QUIC must serve the rotated certificate even after a rejected reload" + ); + client.close(0u32.into(), b"test complete"); + server.close(0u32.into(), b"test complete"); + driver.abort(); + endpoint.close(0u32.into(), b"test complete"); + } }