gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-13 08:02:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(http3): add opt-in QUIC transport (#7903)

Browse Source 
* build(http3): add optional s3s-http3 dependency

Signed-off-by: tommady <tommady@users.noreply.github.com>

* feat(server): add opt-in HTTP/3 transport

Signed-off-by: tommady <tommady@users.noreply.github.com>

* test(server): cover HTTP/3 TLS configuration

* feat(server): add opt-in HTTP/3 support

Serve S3 requests over QUIC with the existing middleware stack.

Signed-off-by: tommady <tommady@users.noreply.github.com>

* fix(http3): cancel pending body reads without panicking

* test(http3): keep transport independent of s3s

---------

Signed-off-by: tommady <tommady@users.noreply.github.com>
Co-authored-by: overtrue <anzhengchao@gmail.com>
This commit is contained in:
Lindroos Hsu
2026-10-07 18:59:21 +08:00
committed by GitHub
parent 14eb28dcd6
commit db2abcf0a6
13 changed files with 1449 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/Cargo.toml
+4
View File
@@ -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 }
rustfs/src/server/http.rs
+517 -125
View File
@@ -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<http3::RequestBody>,
Response<HybridBody<s3s::Body, http_body_util::Empty<Bytes>>>,
Box<dyn std::error::Error + Send + Sync>,
>;
#[cfg(feature = "http3")]
type Http3ExternalService = tower::util::BoxCloneService<
HttpRequest<http3::RequestBody>,
Response<tower_http::body::UnsyncBoxBody<Bytes, Box<dyn std::error::Error + Send + Sync>>>,
Box<dyn std::error::Error + Send + Sync>,
>;
use tracing::{Level, Span, debug, error, info, instrument, trace, warn};
use tracing_opentelemetry::OpenTelemetrySpanExt;
@@ -1086,6 +1102,153 @@ fn make_http_trace_span<ReqBody>(request: &HttpRequest<ReqBody>) -> 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<RemoteAddr>` - 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::<ClientInfo>() {
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::<u64>().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<GlobalReadiness>,
@@ -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<s3s::Body>| {
response.map(|rest_body| HybridBody::<s3s::Body, http_body_util::Empty<Bytes>>::Rest { rest_body })
}))
.service(service);
#[cfg(not(feature = "swift"))]
let service = ServiceBuilder::new()
.map_err(|error: s3s::HttpError| -> Box<dyn std::error::Error + Send + Sync> { 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<RemoteAddr>` - 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::<u64>().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<Incoming>| 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<http3::RequestBody>| async move {
assert_eq!(request.version(), Version::HTTP_3);
assert_eq!(request.extensions().get::<Option<RemoteAddr>>().copied().flatten().expect("peer").0, peer);
let info = request.extensions().get::<ClientInfo>().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<dyn std::error::Error + Send + Sync>>(Response::new(HybridBody::<s3s::Body, http_body_util::Empty<Bytes>>::Rest { rest_body: s3s::Body::from(identity) }))
});
let inner = Http3CoreService::new(inner);
Http3ExternalService::new(external_service_stack!(inner, Some(RemoteAddr(peer)), None::<rustfs_trusted_proxies::TrustedProxyLayer>, 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<s3s::Body>| {
response.map(|rest_body| HybridBody::<s3s::Body, http_body_util::Empty<Bytes>>::Rest { rest_body })
})
.map_err(|error: s3s::HttpError| -> Box<dyn std::error::Error + Send + Sync> { 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::<rustfs_trusted_proxies::TrustedProxyLayer>,
false,
None::<RateLimitLayer>,
true,
readiness.clone(),
None::<Arc<rustfs_keystone::KeystoneAuthProvider>>,
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("<Code>RequestTimeout</Code>"));
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::dto::ListBucketsInput>,
) -> s3s::S3Result<s3s::S3Response<s3s::dto::ListBucketsOutput>> {
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(&timestamp)?;
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", &timestamp).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("<Name>http3-bucket</Name>"));
shutdown.send(())?;
drop(client);
quic.close(0u32.into(), b"test complete");
tokio::time::timeout(Duration::from_secs(5), task).await??;
driver.abort();
Ok(())
}
}
rustfs/src/server/http3.rs
+644
View File
@@ -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<Bytes, io::Error>;
type Resolver = RequestResolver<h3_quinn::Connection, Bytes>;
pub(crate) fn spawn<M, S, B, E>(
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<Request<RequestBody>, Response = Response<B>, Error = E> + Clone + Send + 'static,
S::Future: Send + 'static,
B: HttpBody<Data = Bytes> + 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<S, B, E>(
quic: quinn::Connection,
service: S,
body_timeout: Duration,
cancellation: CancellationToken,
) -> std::result::Result<(), h3::error::ConnectionError>
where
S: Service<Request<RequestBody>, Response = Response<B>, Error = E> + Clone + Send + 'static,
S::Future: Send + 'static,
B: HttpBody<Data = Bytes> + 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<S, B, E>(resolver: Resolver, service: S, body_timeout: Duration)
where
S: Service<Request<RequestBody>, Response = Response<B>, Error = E> + Send,
S::Future: Send,
B: HttpBody<Data = Bytes> + 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::<u64>().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<h3_quinn::RecvStream, Bytes>;
enum State {
Data,
Trailers,
Done,
}
// Dropping an unread body cancels the receive stream without a detached drain.
struct Body {
stream: Option<RecvStream>,
state: State,
expected_length: Option<u64>,
received_length: u64,
timeout: Duration,
timer: Option<Pin<Box<tokio::time::Sleep>>>,
}
impl Body {
fn new(stream: RecvStream, expected_length: Option<u64>, 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<Option<io::Result<Frame<Bytes>>>> {
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<Option<Result<Frame<Self::Data>, 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::<Vec<_>>();
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<T = ()> = std::result::Result<T, Box<dyn std::error::Error + Send + Sync>>;
pub(crate) type Client = h3::client::SendRequest<h3_quinn::OpenStreams, Bytes>;
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<CertificateDer<'static>>) -> TestResult<Endpoint> {
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<RequestBody>| 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<RequestBody>| 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(())
}
}
rustfs/src/server/layer.rs
+5 -3
View File
@@ -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<B>(req: &HttpRequest<B>) -> bool {
&& (path.is_empty() || path == "/rustfs" || path == "/index.html")
}
impl<S, RestBody, GrpcBody> Service<HttpRequest<Incoming>> for RedirectService<S>
impl<S, ReqBody, RestBody, GrpcBody> Service<HttpRequest<ReqBody>> for RedirectService<S>
where
S: Service<HttpRequest<Incoming>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
S: Service<HttpRequest<ReqBody>, Response = Response<HybridBody<RestBody, GrpcBody>>> + Clone + Send + 'static,
S::Future: Send + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>> + 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<Incoming>) -> Self::Future {
fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
let path = req.uri().path().trim_end_matches('/');
if is_console_redirect_request(&req) {
debug!("Redirecting browser request from {} to console", path);
rustfs/src/server/mod.rs
+3
View File
@@ -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.
rustfs/src/server/readiness.rs
+4 -4
View File
@@ -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<Box
response
}
impl<S, B> Service<HttpRequest<Incoming>> for ReadinessGateService<S>
impl<S, ReqBody, B> Service<HttpRequest<ReqBody>> for ReadinessGateService<S>
where
S: Service<HttpRequest<Incoming>, Response = Response<B>> + Clone + Send + 'static,
S: Service<HttpRequest<ReqBody>, Response = Response<B>> + Clone + Send + 'static,
S::Future: Send + 'static,
S::Error: Send + 'static,
ReqBody: Send + 'static,
B: Body<Data = Bytes> + Send + 'static,
B::Error: Into<BoxError> + Send + 'static,
{
@@ -208,7 +208,7 @@ where
self.inner.poll_ready(cx)
}
fn call(&mut self, req: HttpRequest<Incoming>) -> Self::Future {
fn call(&mut self, req: HttpRequest<ReqBody>) -> Self::Future {
let mut inner = self.inner.clone();
let readiness = self.readiness.clone();
Box::pin(async move {
rustfs/src/server/tls_material.rs
+214 -19
View File
@@ -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<Option<quinn::ServerConfig>, 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<Arc<dyn rustls::server::danger::ClientCertVerifier>>,
protocol: ServerProtocol,
) -> Result<rustls::ServerConfig, TlsMaterialError> {
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<RuntimeServerTlsMaterial>,
tls_dir: &Path,
holder: &TlsAcceptorHolder,
#[cfg(feature = "http3")] http3_endpoint: Option<&quinn::Endpoint>,
) -> Result<bool, TlsMaterialError> {
#[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<TlsAcceptorHolder>) {
pub(crate) fn spawn_reload_loop(
tls_path: String,
holder: Arc<TlsAcceptorHolder>,
#[cfg(feature = "http3")] http3_endpoint: Option<quinn::Endpoint>,
) -> Option<tokio::task::JoinHandle<()>> {
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<TlsAcceptorHolder>
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<TlsAcceptorHolder>
"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<TlsAcceptorHolder>
);
}
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<TlsAcceptorHolder>
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<TlsAcceptorHolder>
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<TlsAcceptorHolder>
}
}
}
});
}))
}
#[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::<Vec<CertificateDer<'static>>>()
.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");
}
}