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

refactor(server): unify TLS loading, optimize HTTP transport, add hot reload (#2388)

Browse Source
This commit is contained in:
houseme
2026-04-04 06:37:24 +08:00
committed by GitHub
parent 0ac3b5b992
commit 372f004a8d
8 changed files with 1015 additions and 471 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/config/src/constants/tls.rs
+66
View File
@@ -84,3 +84,69 @@ pub const ENV_SERVER_MTLS_ENABLE: &str = "RUSTFS_SERVER_MTLS_ENABLE";
/// By default, RustFS server mTLS is disabled. /// By default, RustFS server mTLS is disabled.
/// To change this behavior, set the environment variable RUSTFS_SERVER_MTLS_ENABLE=1 /// To change this behavior, set the environment variable RUSTFS_SERVER_MTLS_ENABLE=1
pub const DEFAULT_SERVER_MTLS_ENABLE: bool = false; pub const DEFAULT_SERVER_MTLS_ENABLE: bool = false;
// ── HTTP Transport Tuning Parameters ──
/// Environment variable for HTTP/2 initial stream window size (bytes)
/// Default: 4194304 (4 MB)
pub const ENV_H2_INITIAL_STREAM_WINDOW_SIZE: &str = "RUSTFS_H2_INITIAL_STREAM_WINDOW_SIZE";
pub const DEFAULT_H2_INITIAL_STREAM_WINDOW_SIZE: u32 = 4 * 1024 * 1024; // 4 MB
/// Environment variable for HTTP/2 initial connection window size (bytes)
/// Default: 8388608 (8 MB)
pub const ENV_H2_INITIAL_CONN_WINDOW_SIZE: &str = "RUSTFS_H2_INITIAL_CONN_WINDOW_SIZE";
pub const DEFAULT_H2_INITIAL_CONN_WINDOW_SIZE: u32 = 8 * 1024 * 1024; // 8 MB
/// Environment variable for HTTP/2 max frame size (bytes)
/// Range: 16384 (16 KB) to 16777216 (16 MB) per RFC 7540
/// Default: 524288 (512 KB)
pub const ENV_H2_MAX_FRAME_SIZE: &str = "RUSTFS_H2_MAX_FRAME_SIZE";
pub const DEFAULT_H2_MAX_FRAME_SIZE: u32 = 512 * 1024; // 512 KB
/// Environment variable for HTTP/2 max header list size (bytes)
/// Default: 65536 (64 KB)
pub const ENV_H2_MAX_HEADER_LIST_SIZE: &str = "RUSTFS_H2_MAX_HEADER_LIST_SIZE";
pub const DEFAULT_H2_MAX_HEADER_LIST_SIZE: u32 = 64 * 1024; // 64 KB
/// Environment variable for HTTP/2 max concurrent streams
/// Default: 2048
pub const ENV_H2_MAX_CONCURRENT_STREAMS: &str = "RUSTFS_H2_MAX_CONCURRENT_STREAMS";
pub const DEFAULT_H2_MAX_CONCURRENT_STREAMS: u32 = 2048;
/// Environment variable for HTTP/2 keep-alive interval (seconds)
/// Default: 20
pub const ENV_H2_KEEP_ALIVE_INTERVAL: &str = "RUSTFS_H2_KEEP_ALIVE_INTERVAL";
pub const DEFAULT_H2_KEEP_ALIVE_INTERVAL: u64 = 20;
/// Environment variable for HTTP/2 keep-alive timeout (seconds)
/// Default: 10
pub const ENV_H2_KEEP_ALIVE_TIMEOUT: &str = "RUSTFS_H2_KEEP_ALIVE_TIMEOUT";
pub const DEFAULT_H2_KEEP_ALIVE_TIMEOUT: u64 = 10;
/// Environment variable for HTTP/1.1 header read timeout (seconds)
/// Default: 5
pub const ENV_HTTP1_HEADER_READ_TIMEOUT: &str = "RUSTFS_HTTP1_HEADER_READ_TIMEOUT";
pub const DEFAULT_HTTP1_HEADER_READ_TIMEOUT: u64 = 5;
/// Environment variable for HTTP/1.1 max buffer size (bytes)
/// Default: 65536 (64 KB)
pub const ENV_HTTP1_MAX_BUF_SIZE: &str = "RUSTFS_HTTP1_MAX_BUF_SIZE";
pub const DEFAULT_HTTP1_MAX_BUF_SIZE: usize = 64 * 1024; // 64 KB
// ── TLS Hot Reload Parameters ──
/// Environment variable to enable TLS certificate hot reload
/// Default: false
/// To enable, set the environment variable RUSTFS_TLS_RELOAD_ENABLE=1
pub const ENV_TLS_RELOAD_ENABLE: &str = "RUSTFS_TLS_RELOAD_ENABLE";
/// Default value for TLS certificate hot reload
/// By default, RustFS does not reload TLS certificates automatically.
pub const DEFAULT_TLS_RELOAD_ENABLE: bool = false;
/// Environment variable for TLS certificate reload interval (seconds)
/// Default: 30 seconds. Minimum: 5 seconds.
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;
rustfs/Cargo.toml
+1
View File
@@ -47,6 +47,7 @@ webdav = ["rustfs-protocols/webdav"]
license = [] license = []
direct-io = [] # Aligned direct I/O reader support (uses aligned pread, does not set O_DIRECT) direct-io = [] # Aligned direct I/O reader support (uses aligned pread, does not set O_DIRECT)
io-scheduler-debug = [] # Enable debug information in I/O scheduler io-scheduler-debug = [] # Enable debug information in I/O scheduler
tracing-chunk-debug = [] # Enable per-chunk tracing in data plane (high noise, for debugging only)
full = ["metrics-gpu", "ftps", "swift", "webdav", "direct-io"] full = ["metrics-gpu", "ftps", "swift", "webdav", "direct-io"]
manual-test-runners = [] manual-test-runners = []
rustfs/src/main.rs
+9 -6
View File
@@ -43,7 +43,7 @@ use crate::init::init_webdav_system;
use crate::capacity::capacity_integration::init_capacity_management; use crate::capacity::capacity_integration::init_capacity_management;
use crate::server::{ use crate::server::{
SHUTDOWN_TIMEOUT, ServiceState, ServiceStateManager, ShutdownSignal, init_cert, init_event_notifier, shutdown_event_notifier, SHUTDOWN_TIMEOUT, ServiceState, ServiceStateManager, ShutdownSignal, init_event_notifier, shutdown_event_notifier,
start_audit_system, start_http_server, stop_audit_system, wait_for_shutdown, start_audit_system, start_http_server, stop_audit_system, wait_for_shutdown,
}; };
use license::{current_license, init_license, license_status}; use license::{current_license, init_license, license_status};
@@ -229,15 +229,18 @@ async fn async_main() -> Result<()> {
// A crypto provider is already installed (e.g. by the host process); this is fine. // A crypto provider is already installed (e.g. by the host process); this is fine.
debug!("rustls crypto provider already installed, skipping aws-lc-rs default install"); debug!("rustls crypto provider already installed, skipping aws-lc-rs default install");
} }
// Initialize TLS if a certificate path is provided // Initialize TLS outbound material (root CAs, mTLS identity) if configured.
// Server-side TLS acceptor is built separately inside start_http_server()
// using the same TlsMaterialSnapshot loading logic.
if let Some(tls_path) = &config.tls_path { if let Some(tls_path) = &config.tls_path {
match init_cert(tls_path).await { match crate::server::tls_material::TlsMaterialSnapshot::load(tls_path).await {
Ok(_) => { Ok(snapshot) => {
info!(target: "rustfs::main", "TLS initialized successfully with certs from {}", tls_path); snapshot.apply_outbound().await;
info!(target: "rustfs::main", "TLS outbound material initialized from {}", tls_path);
} }
Err(e) => { Err(e) => {
error!("Failed to initialize TLS from {}: {}", tls_path, e); error!("Failed to initialize TLS from {}: {}", tls_path, e);
return Err(Error::other(e)); return Err(Error::other(e.to_string()));
} }
} }
} }
rustfs/src/server/cert.rs
-296
View File
@@ -1,296 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use rustfs_common::{MtlsIdentityPem, set_global_mtls_identity, set_global_root_cert};
use rustfs_config::{RUSTFS_CA_CERT, RUSTFS_PUBLIC_CERT, RUSTFS_TLS_CERT};
use rustls::pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject};
use std::path::{Path, PathBuf};
use tracing::{debug, info};
#[derive(Debug)]
pub enum RustFSError {
Cert(String),
}
impl std::fmt::Display for RustFSError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
RustFSError::Cert(msg) => write!(f, "Certificate error: {msg}"),
}
}
}
impl std::error::Error for RustFSError {}
/// Parse PEM-encoded certificates into DER format.
/// Returns a vector of DER-encoded certificates.
///
/// # Arguments
/// * `pem` - A byte slice containing the PEM-encoded certificates.
///
/// # Returns
/// A vector of `CertificateDer` containing the DER-encoded certificates.
///
/// # Errors
/// Returns `RustFSError` if parsing fails.
fn parse_pem_certs(pem: &[u8]) -> Result<Vec<CertificateDer<'static>>, RustFSError> {
let mut out = Vec::new();
let mut reader = std::io::Cursor::new(pem);
for item in CertificateDer::pem_reader_iter(&mut reader) {
let c = item.map_err(|e| RustFSError::Cert(format!("parse cert pem: {e}")))?;
out.push(c);
}
Ok(out)
}
/// Parse a PEM-encoded private key into DER format.
/// Supports PKCS#8 and RSA private keys.
///
/// # Arguments
/// * `pem` - A byte slice containing the PEM-encoded private key.
///
/// # Returns
/// A `PrivateKeyDer` containing the DER-encoded private key.
///
/// # Errors
/// Returns `RustFSError` if parsing fails or no key is found.
fn parse_pem_private_key(pem: &[u8]) -> Result<PrivateKeyDer<'static>, RustFSError> {
let mut reader = std::io::Cursor::new(pem);
PrivateKeyDer::from_pem_reader(&mut reader).map_err(|e| RustFSError::Cert(format!("parse private key pem: {e}")))
}
/// Helper function to read a file and return its contents.
/// Returns the file contents as a vector of bytes.
/// # Errors
/// Returns `RustFSError` if reading fails.
async fn read_file(path: &PathBuf, desc: &str) -> Result<Vec<u8>, RustFSError> {
tokio::fs::read(path)
.await
.map_err(|e| RustFSError::Cert(format!("read {desc} {path:?}: {e}")))
}
/// Initialize TLS material for both server and outbound client connections.
///
/// Loads roots from:
/// - `${RUSTFS_TLS_PATH}/ca.crt` (or `tls/ca.crt`)
/// - `${RUSTFS_TLS_PATH}/public.crt` (optional additional root bundle)
/// - system roots if `RUSTFS_TRUST_SYSTEM_CA=true` (default: false)
/// - if `RUSTFS_TRUST_LEAF_CERT_AS_CA=true`, also loads leaf cert(s) from
/// `${RUSTFS_TLS_PATH}/rustfs_cert.pem` into the root store.
///
/// Loads mTLS client identity (optional) from:
/// - `${RUSTFS_TLS_PATH}/client_cert.pem`
/// - `${RUSTFS_TLS_PATH}/client_key.pem`
///
/// Environment overrides:
/// - RUSTFS_TLS_PATH
/// - RUSTFS_MTLS_CLIENT_CERT
/// - RUSTFS_MTLS_CLIENT_KEY
pub(crate) async fn init_cert(tls_path: &str) -> Result<(), RustFSError> {
if tls_path.is_empty() {
info!("No TLS path configured; skipping certificate initialization");
return Ok(());
}
let tls_dir = PathBuf::from(tls_path);
// Load root certificates
load_root_certs(&tls_dir).await?;
// Load optional mTLS identity
load_mtls_identity(&tls_dir).await?;
Ok(())
}
/// Load root certificates from various sources.
async fn load_root_certs(tls_dir: &Path) -> Result<(), RustFSError> {
let mut cert_data = Vec::new();
let trust_leaf_as_ca =
rustfs_utils::get_env_bool(rustfs_config::ENV_TRUST_LEAF_CERT_AS_CA, rustfs_config::DEFAULT_TRUST_LEAF_CERT_AS_CA);
if trust_leaf_as_ca {
walk_dir(tls_dir.to_path_buf(), RUSTFS_TLS_CERT, &mut cert_data).await;
info!("Loaded leaf certificate(s) as root CA as per RUSTFS_TRUST_LEAF_CERT_AS_CA");
}
// Try public.crt and ca.crt
let public_cert_path = tls_dir.join(RUSTFS_PUBLIC_CERT);
load_cert_file(public_cert_path.to_str().unwrap_or_default(), &mut cert_data, "CA certificate").await;
let ca_cert_path = tls_dir.join(RUSTFS_CA_CERT);
load_cert_file(ca_cert_path.to_str().unwrap_or_default(), &mut cert_data, "CA certificate").await;
// Load system root certificates if enabled
let trust_system_ca = rustfs_utils::get_env_bool(rustfs_config::ENV_TRUST_SYSTEM_CA, rustfs_config::DEFAULT_TRUST_SYSTEM_CA);
if trust_system_ca {
let system_ca_paths = [
"/etc/ssl/certs/ca-certificates.crt", // Debian/Ubuntu/Alpine
"/etc/pki/tls/certs/ca-bundle.crt", // Fedora/RHEL/CentOS
"/etc/ssl/ca-bundle.pem", // OpenSUSE
"/etc/pki/tls/cacert.pem", // OpenELEC
"/etc/ssl/cert.pem", // macOS/FreeBSD
"/usr/local/etc/openssl/cert.pem", // macOS/Homebrew OpenSSL
"/usr/local/share/certs/ca-root-nss.crt", // FreeBSD
"/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem", // RHEL
"/usr/share/pki/ca-trust-legacy/ca-bundle.legacy.crt", // RHEL legacy
];
let mut system_cert_loaded = false;
for path in system_ca_paths {
if load_cert_file(path, &mut cert_data, "system root certificates").await {
system_cert_loaded = true;
info!("Loaded system root certificates from {}", path);
break;
}
}
if !system_cert_loaded {
debug!("Could not find system root certificates in common locations.");
}
} else {
info!("Loading system root certificates disabled via RUSTFS_TRUST_SYSTEM_CA");
}
if !cert_data.is_empty() {
set_global_root_cert(cert_data).await;
info!("Configured custom root certificates for inter-node communication");
}
Ok(())
}
/// Load optional mTLS identity.
async fn load_mtls_identity(tls_dir: &Path) -> Result<(), RustFSError> {
let client_cert_path = match rustfs_utils::get_env_opt_str(rustfs_config::ENV_MTLS_CLIENT_CERT) {
Some(p) => PathBuf::from(p),
None => tls_dir.join(rustfs_config::RUSTFS_CLIENT_CERT_FILENAME),
};
let client_key_path = match rustfs_utils::get_env_opt_str(rustfs_config::ENV_MTLS_CLIENT_KEY) {
Some(p) => PathBuf::from(p),
None => tls_dir.join(rustfs_config::RUSTFS_CLIENT_KEY_FILENAME),
};
if client_cert_path.exists() && client_key_path.exists() {
let cert_bytes = read_file(&client_cert_path, "client cert").await?;
let key_bytes = read_file(&client_key_path, "client key").await?;
// Validate parse-ability early; store as PEM bytes for tonic.
parse_pem_certs(&cert_bytes)?;
parse_pem_private_key(&key_bytes)?;
let identity_pem = MtlsIdentityPem {
cert_pem: cert_bytes,
key_pem: key_bytes,
};
set_global_mtls_identity(Some(identity_pem)).await;
info!("Loaded mTLS client identity cert={:?} key={:?}", client_cert_path, client_key_path);
} else {
set_global_mtls_identity(None).await;
info!(
"mTLS client identity not configured (missing {:?} and/or {:?}); proceeding with server-only TLS",
client_cert_path, client_key_path
);
}
Ok(())
}
/// Helper function to load a certificate file and append to cert_data.
/// Returns true if the file was successfully loaded.
async fn load_cert_file(path: &str, cert_data: &mut Vec<u8>, desc: &str) -> bool {
if tokio::fs::metadata(path).await.is_ok() {
if let Ok(data) = tokio::fs::read(path).await {
cert_data.extend(data);
cert_data.push(b'\n');
info!("Loaded {} from {}", desc, path);
true
} else {
debug!("Failed to read {} from {}", desc, path);
false
}
} else {
debug!("{} file not found at {}", desc, path);
false
}
}
/// Load the certificate file if its name matches `cert_name`.
/// If it matches, the certificate data is appended to `cert_data`.
///
/// # Parameters
/// - `entry`: The directory entry to check.
/// - `cert_name`: The name of the certificate file to match.
/// - `cert_data`: A mutable vector to append loaded certificate data.
async fn load_if_matches(entry: &tokio::fs::DirEntry, cert_name: &str, cert_data: &mut Vec<u8>) {
let fname = entry.file_name().to_string_lossy().to_string();
if fname == cert_name {
let p = entry.path();
load_cert_file(&p.to_string_lossy(), cert_data, "certificate").await;
}
}
/// Search the directory at `path` and one level of subdirectories to find and load
/// certificates matching `cert_name`. Loaded certificate data is appended to
/// `cert_data`.
/// # Parameters
/// - `path`: The starting directory path to search for certificates.
/// - `cert_name`: The name of the certificate file to look for.
/// - `cert_data`: A mutable vector to append loaded certificate data.
async fn walk_dir(path: PathBuf, cert_name: &str, cert_data: &mut Vec<u8>) {
if let Ok(mut rd) = tokio::fs::read_dir(&path).await {
while let Ok(Some(entry)) = rd.next_entry().await {
if let Ok(ft) = entry.file_type().await {
if ft.is_file() {
load_if_matches(&entry, cert_name, cert_data).await;
} else if ft.is_dir() {
// Only check direct subdirectories, no deeper recursion
if let Ok(mut sub_rd) = tokio::fs::read_dir(&entry.path()).await {
while let Ok(Some(sub_entry)) = sub_rd.next_entry().await {
if let Ok(sub_ft) = sub_entry.file_type().await
&& sub_ft.is_file()
{
load_if_matches(&sub_entry, cert_name, cert_data).await;
}
// Ignore subdirectories and symlinks in subdirs to limit to one level
}
}
} else if ft.is_symlink() {
// Follow symlink and treat target as file or directory, but limit to one level
if let Ok(meta) = tokio::fs::metadata(&entry.path()).await {
if meta.is_file() {
load_if_matches(&entry, cert_name, cert_data).await;
} else if meta.is_dir() {
// Treat as directory but only check its direct contents
if let Ok(mut sub_rd) = tokio::fs::read_dir(&entry.path()).await {
while let Ok(Some(sub_entry)) = sub_rd.next_entry().await {
if let Ok(sub_ft) = sub_entry.file_type().await
&& sub_ft.is_file()
{
load_if_matches(&sub_entry, cert_name, cert_data).await;
}
// Ignore deeper levels
}
}
}
}
}
}
}
} else {
debug!("Certificate directory not found: {}", path.display());
}
}
rustfs/src/server/compress.rs
+196
View File
@@ -50,6 +50,11 @@ use std::str::FromStr;
use tower_http::compression::predicate::Predicate; use tower_http::compression::predicate::Predicate;
use tracing::debug; use tracing::debug;
/// Response extension key for storing the request path category.
/// Set by `PathCategoryInjectionLayer` before the compression predicate evaluates.
#[derive(Debug, Clone, Copy)]
pub(crate) struct RequestPathCategory(pub(crate) PathCategory);
/// Configuration for HTTP response compression. /// Configuration for HTTP response compression.
/// ///
/// This structure holds the whitelist-based compression settings: /// This structure holds the whitelist-based compression settings:
@@ -319,6 +324,156 @@ impl Predicate for CompressionPredicate {
} }
} }
// ── Path-Aware Compression ──
/// Classifies request paths to determine if compression should apply.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum PathCategory {
/// S3 data plane (bucket/key operations) — compression applies via whitelist
S3DataPlane,
/// Admin API paths — skip compression (small JSON responses)
AdminApi,
/// Console paths — skip compression (static assets, already optimized)
Console,
/// Internode RPC paths — skip compression (binary protocol data)
InternodeRpc,
/// Health/probe paths — skip compression (tiny responses)
Probe,
}
impl PathCategory {
/// Classify a request URI path into a category.
pub(crate) fn classify(path: &str) -> Self {
if path.starts_with("/rustfs/rpc/") || path.starts_with("/rustfs/peer/") {
PathCategory::InternodeRpc
} else if path.starts_with("/rustfs/admin/") || path.starts_with("/minio/admin/") {
PathCategory::AdminApi
} else if path.starts_with("/rustfs/console") {
PathCategory::Console
} else if path.starts_with("/minio/health/") {
PathCategory::Probe
} else {
PathCategory::S3DataPlane
}
}
/// Returns true if compression should be considered for this path category.
/// Only S3 data plane paths go through the full compression predicate.
#[inline]
pub(crate) fn should_evaluate_compression(self) -> bool {
matches!(self, PathCategory::S3DataPlane)
}
}
/// A compression predicate that first checks the request path category
/// before evaluating the full compression rules.
///
/// This avoids running MIME type / extension matching for admin, RPC, console,
/// and health probe paths where compression is never beneficial.
#[derive(Clone, Debug)]
pub(crate) struct PathAwareCompressionPredicate {
inner: CompressionPredicate,
}
impl PathAwareCompressionPredicate {
pub(crate) fn new(config: CompressionConfig) -> Self {
Self {
inner: CompressionPredicate::new(config),
}
}
}
impl Predicate for PathAwareCompressionPredicate {
fn should_compress<B>(&self, response: &Response<B>) -> bool
where
B: http_body::Body,
{
// Fast path: skip full predicate evaluation for non-S3 paths
if let Some(RequestPathCategory(category)) = response.extensions().get::<RequestPathCategory>()
&& !category.should_evaluate_compression()
{
return false;
}
self.inner.should_compress(response)
}
}
use http::Request;
use http_body::Body;
use std::pin::Pin;
use std::task::{Context, Poll};
use tower::{Layer, Service};
/// Tower layer that injects `RequestPathCategory` into each response's extensions
/// based on the incoming request URI path. Must be placed before `CompressionLayer`.
#[derive(Clone, Copy, Debug)]
pub(crate) struct PathCategoryInjectionLayer;
impl<S> Layer<S> for PathCategoryInjectionLayer {
type Service = PathCategoryInjectionService<S>;
fn layer(&self, inner: S) -> Self::Service {
PathCategoryInjectionService { inner }
}
}
/// Service wrapper that adds `RequestPathCategory` to response extensions.
#[derive(Clone)]
pub(crate) struct PathCategoryInjectionService<S> {
inner: S,
}
pin_project_lite::pin_project! {
/// Future for `PathCategoryInjectionService` that injects path category into response.
#[project = InjectCategoryFutProj]
pub(crate) struct InjectCategoryFut<F> {
#[pin]
inner: F,
category: PathCategory,
}
}
impl<F, ResBody, E> std::future::Future for InjectCategoryFut<F>
where
F: std::future::Future<Output = Result<Response<ResBody>, E>>,
{
type Output = Result<Response<ResBody>, E>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();
match this.inner.poll(cx) {
Poll::Ready(Ok(mut resp)) => {
resp.extensions_mut().insert(RequestPathCategory(*this.category));
Poll::Ready(Ok(resp))
}
Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
Poll::Pending => Poll::Pending,
}
}
}
impl<S, ReqBody, ResBody> Service<Request<ReqBody>> for PathCategoryInjectionService<S>
where
S: Service<Request<ReqBody>, Response = Response<ResBody>>,
ResBody: Body,
{
type Response = Response<ResBody>;
type Error = S::Error;
type Future = InjectCategoryFut<S::Future>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
let category = PathCategory::classify(req.uri().path());
InjectCategoryFut {
inner: self.inner.call(req),
category,
}
}
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
@@ -471,4 +626,45 @@ mod tests {
assert_eq!(predicate.config.mime_patterns.len(), 2); assert_eq!(predicate.config.mime_patterns.len(), 2);
assert_eq!(predicate.config.min_size, 1000); assert_eq!(predicate.config.min_size, 1000);
} }
#[test]
fn test_path_category_classify_s3() {
assert_eq!(PathCategory::classify("/"), PathCategory::S3DataPlane);
assert_eq!(PathCategory::classify("/mybucket"), PathCategory::S3DataPlane);
assert_eq!(PathCategory::classify("/mybucket/mykey"), PathCategory::S3DataPlane);
assert_eq!(PathCategory::classify("/bucket?list-type=2"), PathCategory::S3DataPlane);
}
#[test]
fn test_path_category_classify_admin() {
assert_eq!(PathCategory::classify("/rustfs/admin/v3/service"), PathCategory::AdminApi);
assert_eq!(PathCategory::classify("/minio/admin/v3/info"), PathCategory::AdminApi);
}
#[test]
fn test_path_category_classify_console() {
assert_eq!(PathCategory::classify("/rustfs/console/index.html"), PathCategory::Console);
assert_eq!(PathCategory::classify("/rustfs/console"), PathCategory::Console);
}
#[test]
fn test_path_category_classify_rpc() {
assert_eq!(PathCategory::classify("/rustfs/rpc/read_file_stream"), PathCategory::InternodeRpc);
assert_eq!(PathCategory::classify("/rustfs/peer/health"), PathCategory::InternodeRpc);
}
#[test]
fn test_path_category_classify_probe() {
assert_eq!(PathCategory::classify("/minio/health/live"), PathCategory::Probe);
assert_eq!(PathCategory::classify("/minio/health/ready"), PathCategory::Probe);
}
#[test]
fn test_path_category_should_evaluate() {
assert!(PathCategory::S3DataPlane.should_evaluate_compression());
assert!(!PathCategory::AdminApi.should_evaluate_compression());
assert!(!PathCategory::Console.should_evaluate_compression());
assert!(!PathCategory::InternodeRpc.should_evaluate_compression());
assert!(!PathCategory::Probe.should_evaluate_compression());
}
} }
rustfs/src/server/http.rs
+232 -167
View File
@@ -19,9 +19,10 @@ use crate::auth_keystone;
use crate::config; use crate::config;
use crate::server::{ use crate::server::{
ReadinessGateLayer, RemoteAddr, ServiceState, ServiceStateManager, ReadinessGateLayer, RemoteAddr, ServiceState, ServiceStateManager,
compress::{CompressionConfig, CompressionPredicate}, compress::{CompressionConfig, PathAwareCompressionPredicate, PathCategoryInjectionLayer},
hybrid::hybrid, hybrid::hybrid,
layer::{AdminChunkedContentLengthCompatLayer, ConditionalCorsLayer, ObjectAttributesEtagFixLayer, RedirectLayer}, layer::{AdminChunkedContentLengthCompatLayer, ConditionalCorsLayer, ObjectAttributesEtagFixLayer, RedirectLayer},
tls_material::{TlsAcceptorHolder, TlsHandshakeFailureKind, TlsMaterialSnapshot, spawn_reload_loop},
}; };
use crate::storage; use crate::storage;
use crate::storage::rpc::InternodeRpcService; use crate::storage::rpc::InternodeRpcService;
@@ -38,7 +39,6 @@ use metrics::{counter, histogram};
use opentelemetry::global; use opentelemetry::global;
use opentelemetry::trace::TraceContextExt; use opentelemetry::trace::TraceContextExt;
use rustfs_common::GlobalReadiness; use rustfs_common::GlobalReadiness;
use rustfs_config::{RUSTFS_TLS_CERT, RUSTFS_TLS_KEY};
use rustfs_ecstore::rpc::{TONIC_RPC_PREFIX, verify_rpc_signature}; use rustfs_ecstore::rpc::{TONIC_RPC_PREFIX, verify_rpc_signature};
use rustfs_keystone::KeystoneAuthLayer; use rustfs_keystone::KeystoneAuthLayer;
#[cfg(feature = "swift")] #[cfg(feature = "swift")]
@@ -46,7 +46,6 @@ use rustfs_protocols::SwiftService;
use rustfs_protos::proto_gen::node_service::node_service_server::NodeServiceServer; use rustfs_protos::proto_gen::node_service::node_service_server::NodeServiceServer;
use rustfs_trusted_proxies::ClientInfo; use rustfs_trusted_proxies::ClientInfo;
use rustfs_utils::net::parse_and_resolve_address; use rustfs_utils::net::parse_and_resolve_address;
use rustls::ServerConfig;
use s3s::{host::MultiDomain, service::S3Service, service::S3ServiceBuilder}; use s3s::{host::MultiDomain, service::S3Service, service::S3ServiceBuilder};
use socket2::{SockRef, TcpKeepalive}; use socket2::{SockRef, TcpKeepalive};
use std::io::{Error, Result}; use std::io::{Error, Result};
@@ -54,7 +53,6 @@ use std::net::SocketAddr;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tokio::net::{TcpListener, TcpStream}; use tokio::net::{TcpListener, TcpStream};
use tokio_rustls::TlsAcceptor;
use tonic::{Request, Status}; use tonic::{Request, Status};
use tower::ServiceBuilder; use tower::ServiceBuilder;
use tower_http::add_extension::AddExtensionLayer; use tower_http::add_extension::AddExtensionLayer;
@@ -156,9 +154,24 @@ pub async fn start_http_server(
TcpListener::from_std(socket.into())? TcpListener::from_std(socket.into())?
}; };
let tls_acceptor = setup_tls_acceptor(config.tls_path.as_deref().unwrap_or_default()).await?; let tls_path = config.tls_path.as_deref().unwrap_or_default();
// Load TLS materials and build server acceptor.
// Note: outbound material (root CAs, mTLS identity) is already applied in main.rs.
let tls_snapshot = TlsMaterialSnapshot::load(tls_path)
.await
.map_err(|e| std::io::Error::other(e.to_string()))?;
let tls_acceptor = tls_snapshot
.build_tls_acceptor(tls_path)
.await
.map_err(|e| std::io::Error::other(e.to_string()))?;
let tls_enabled = tls_acceptor.is_some(); let tls_enabled = tls_acceptor.is_some();
let protocol = if tls_enabled { "https" } else { "http" }; 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 // Obtain the listener address
let local_addr: SocketAddr = listener.local_addr()?; let local_addr: SocketAddr = listener.local_addr()?;
let local_ip = match rustfs_utils::get_local_ip() { let local_ip = match rustfs_utils::get_local_ip() {
@@ -273,11 +286,53 @@ pub async fn start_http_server(
(sigterm_inner, sigint_inner) (sigterm_inner, sigint_inner)
}; };
// RustFS Transport Layer Configuration Constants - Optimized for S3 Workloads // ── HTTP Transport Tuning (configurable via env vars) ──
const H2_INITIAL_STREAM_WINDOW_SIZE: u32 = 1024 * 1024 * 4; // 4MB: Optimize large file throughput // Read all transport parameters from environment, falling back to defaults.
const H2_INITIAL_CONN_WINDOW_SIZE: u32 = 1024 * 1024 * 8; // 8MB: Link-level flow control // H2 frame size is clamped to RFC 7540 range: 2^14 (16KB) to 2^24 (16MB).
const H2_MAX_FRAME_SIZE: u32 = 512 * 1024; // 512KB: Reduce framing overhead for large objects
const H2_MAX_HEADER_LIST_SIZE: u32 = 64 * 1024; // 64KB: Conservative header limit to mitigate DoS risk let h2_stream_window = rustfs_utils::get_env_u32(
rustfs_config::ENV_H2_INITIAL_STREAM_WINDOW_SIZE,
rustfs_config::DEFAULT_H2_INITIAL_STREAM_WINDOW_SIZE,
);
let h2_conn_window = rustfs_utils::get_env_u32(
rustfs_config::ENV_H2_INITIAL_CONN_WINDOW_SIZE,
rustfs_config::DEFAULT_H2_INITIAL_CONN_WINDOW_SIZE,
);
let h2_max_frame_size =
rustfs_utils::get_env_u32(rustfs_config::ENV_H2_MAX_FRAME_SIZE, rustfs_config::DEFAULT_H2_MAX_FRAME_SIZE)
.clamp(16_384, 16_777_216); // RFC 7540
let h2_max_header_list_size =
rustfs_utils::get_env_u32(rustfs_config::ENV_H2_MAX_HEADER_LIST_SIZE, rustfs_config::DEFAULT_H2_MAX_HEADER_LIST_SIZE);
let h2_max_concurrent_streams = rustfs_utils::get_env_u32(
rustfs_config::ENV_H2_MAX_CONCURRENT_STREAMS,
rustfs_config::DEFAULT_H2_MAX_CONCURRENT_STREAMS,
)
.max(1);
let h2_keep_alive_interval =
rustfs_utils::get_env_u64(rustfs_config::ENV_H2_KEEP_ALIVE_INTERVAL, rustfs_config::DEFAULT_H2_KEEP_ALIVE_INTERVAL);
let h2_keep_alive_timeout =
rustfs_utils::get_env_u64(rustfs_config::ENV_H2_KEEP_ALIVE_TIMEOUT, rustfs_config::DEFAULT_H2_KEEP_ALIVE_TIMEOUT);
let http1_header_read_timeout = rustfs_utils::get_env_u64(
rustfs_config::ENV_HTTP1_HEADER_READ_TIMEOUT,
rustfs_config::DEFAULT_HTTP1_HEADER_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);
info!(
"HTTP transport parameters: h2_stream_window={}, h2_conn_window={}, h2_max_frame={}, \
h2_max_header_list={}, h2_max_concurrent_streams={}, h2_keepalive_interval={}s, \
h2_keepalive_timeout={}s, http1_header_timeout={}s, http1_max_buf={}",
h2_stream_window,
h2_conn_window,
h2_max_frame_size,
h2_max_header_list_size,
h2_max_concurrent_streams,
h2_keep_alive_interval,
h2_keep_alive_timeout,
http1_header_read_timeout,
http1_max_buf_size,
);
let mut conn_builder = ConnBuilder::new(TokioExecutor::new()); let mut conn_builder = ConnBuilder::new(TokioExecutor::new());
@@ -286,8 +341,8 @@ pub async fn start_http_server(
.http1() .http1()
.timer(TokioTimer::new()) .timer(TokioTimer::new())
.keep_alive(true) .keep_alive(true)
.header_read_timeout(Duration::from_secs(5)) .header_read_timeout(Duration::from_secs(http1_header_read_timeout))
.max_buf_size(64 * 1024) .max_buf_size(http1_max_buf_size)
.writev(true); .writev(true);
// Optimize for HTTP/2 (AI/Data Lake high concurrency synchronization) // Optimize for HTTP/2 (AI/Data Lake high concurrency synchronization)
@@ -295,13 +350,13 @@ pub async fn start_http_server(
.http2() .http2()
.timer(TokioTimer::new()) .timer(TokioTimer::new())
.adaptive_window(true) .adaptive_window(true)
.initial_stream_window_size(H2_INITIAL_STREAM_WINDOW_SIZE) .initial_stream_window_size(h2_stream_window)
.initial_connection_window_size(H2_INITIAL_CONN_WINDOW_SIZE) .initial_connection_window_size(h2_conn_window)
.max_frame_size(H2_MAX_FRAME_SIZE) .max_frame_size(h2_max_frame_size)
.max_concurrent_streams(Some(2048)) .max_concurrent_streams(Some(h2_max_concurrent_streams))
.max_header_list_size(H2_MAX_HEADER_LIST_SIZE) .max_header_list_size(h2_max_header_list_size)
.keep_alive_interval(Some(Duration::from_secs(20))) .keep_alive_interval(Some(Duration::from_secs(h2_keep_alive_interval)))
.keep_alive_timeout(Duration::from_secs(10)); .keep_alive_timeout(Duration::from_secs(h2_keep_alive_timeout));
let http_server = Arc::new(conn_builder); let http_server = Arc::new(conn_builder);
let mut ctrl_c = std::pin::pin!(tokio::signal::ctrl_c()); let mut ctrl_c = std::pin::pin!(tokio::signal::ctrl_c());
@@ -310,10 +365,7 @@ pub async fn start_http_server(
// service ready // service ready
worker_state_manager.update(ServiceState::Ready); worker_state_manager.update(ServiceState::Ready);
let tls_acceptor = tls_acceptor.map(Arc::new); // tls_acceptor is already Option<Arc<TlsAcceptorHolder>>, clone for the loop
// Initialize keepalive configuration once to avoid recreation in the loop
let keepalive_conf = get_default_tcp_keepalive();
loop { loop {
debug!("Waiting for new connection..."); debug!("Waiting for new connection...");
@@ -374,28 +426,29 @@ pub async fn start_http_server(
let socket_ref = SockRef::from(&socket); let socket_ref = SockRef::from(&socket);
// Enable TCP Keepalive to detect dead clients (e.g. power loss) // ── POST-ACCEPT SOCKET SYSCALLS ──
if let Err(err) = socket_ref.set_tcp_keepalive(&keepalive_conf) { // The listening socket already sets TCP_NODELAY, TCP_KEEPALIVE,
warn!(?err, "Failed to set TCP_KEEPALIVE"); // SO_RCVBUF, and SO_SNDBUF. On Linux/BSD, these are inherited by
} // accepted sockets, so we skip redundant re-application here.
//
// Only TCP_QUICKACK (Linux) is kept — it is inherently per-connection
// and NOT inherited from the listening socket.
//
// T03 optimized: syscall count reduced from 5 → 1 (Linux) / 0 (other)
// Disable Nagle algorithm: Critical for 4KB Payload, achieving ultra-low latency // Enable TCP QuickAck to reduce latency for small requests (Linux only)
if let Err(err) = socket_ref.set_tcp_nodelay(true) {
warn!(?err, "Failed to set TCP_NODELAY");
}
// Enable TCP QuickAck to reduce latency for small requests
#[cfg(target_os = "linux")] #[cfg(target_os = "linux")]
if let Err(err) = socket_ref.set_tcp_quickack(true) { if let Err(err) = socket_ref.set_tcp_quickack(true) {
debug!(?err, "Failed to set TCP_QUICKACK"); debug!(?err, "Failed to set TCP_QUICKACK");
} }
// Increase receive/send buffer to support BDP at GB-level throughput // Debug-only: verify listening socket options were inherited
if let Err(err) = socket_ref.set_recv_buffer_size(4 * rustfs_config::MI_B) { #[cfg(debug_assertions)]
warn!(?err, "Failed to set set_recv_buffer_size"); {
} debug!(
if let Err(err) = socket_ref.set_send_buffer_size(4 * rustfs_config::MI_B) { nodelay = socket_ref.tcp_nodelay().unwrap_or(false),
warn!(?err, "Failed to set set_send_buffer_size"); "TCP_NODELAY inherited from listening socket"
);
} }
let connection_ctx = ConnectionContext { let connection_ctx = ConnectionContext {
@@ -404,6 +457,8 @@ pub async fn start_http_server(
compression_config: compression_config.clone(), compression_config: compression_config.clone(),
is_console, is_console,
readiness: readiness.clone(), readiness: readiness.clone(),
keystone_auth: auth_keystone::get_keystone_auth(),
trusted_proxy_layer: rustfs_trusted_proxies::is_enabled().then(|| rustfs_trusted_proxies::layer().clone()),
}; };
process_connection(socket, tls_acceptor.clone(), connection_ctx, graceful.clone()); process_connection(socket, tls_acceptor.clone(), connection_ctx, graceful.clone());
@@ -433,88 +488,6 @@ pub async fn start_http_server(
Ok(shutdown_tx) Ok(shutdown_tx)
} }
/// Sets up the TLS acceptor if certificates are available.
#[instrument(skip(tls_path))]
async fn setup_tls_acceptor(tls_path: &str) -> Result<Option<TlsAcceptor>> {
if tls_path.is_empty() || tokio::fs::metadata(tls_path).await.is_err() {
debug!("TLS path is not provided or does not exist, starting with HTTP");
return Ok(None);
}
debug!("Found TLS directory, checking for certificates");
let mtls_verifier = rustfs_utils::build_webpki_client_verifier(tls_path)?;
// 1. Attempt to load all certificates in the directory (multi-certificate support, for SNI)
if let Ok(cert_key_pairs) = rustfs_utils::load_all_certs_from_directory(tls_path)
&& !cert_key_pairs.is_empty()
{
debug!("Found {} certificates, creating SNI-aware multi-cert resolver", cert_key_pairs.len());
// Create an SNI-enabled certificate resolver
let resolver = rustfs_utils::create_multi_cert_resolver(cert_key_pairs)?;
// Configure the server to enable SNI support
let mut server_config = if let Some(verifier) = mtls_verifier.clone() {
ServerConfig::builder()
.with_client_cert_verifier(verifier)
.with_cert_resolver(Arc::new(resolver))
} else {
ServerConfig::builder()
.with_no_client_auth()
.with_cert_resolver(Arc::new(resolver))
};
// Configure ALPN protocol priority
server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()];
// Enable session resumption to reduce handshake overhead for returning clients
server_config.session_storage = rustls::server::ServerSessionMemoryCache::new(10000);
// Log SNI requests
if rustfs_utils::tls_key_log() {
server_config.key_log = Arc::new(rustls::KeyLogFile::new());
}
return Ok(Some(TlsAcceptor::from(Arc::new(server_config))));
}
// 2. Revert to the traditional single-certificate mode
let key_path = format!("{tls_path}/{RUSTFS_TLS_KEY}");
let cert_path = format!("{tls_path}/{RUSTFS_TLS_CERT}");
if tokio::try_join!(tokio::fs::metadata(&key_path), tokio::fs::metadata(&cert_path)).is_ok() {
debug!("Found legacy single TLS certificate, starting with HTTPS");
let certs = rustfs_utils::load_certs(&cert_path).map_err(|e| rustfs_utils::certs_error(e.to_string()))?;
let key = rustfs_utils::load_private_key(&key_path).map_err(|e| rustfs_utils::certs_error(e.to_string()))?;
let mut server_config = if let Some(verifier) = mtls_verifier {
ServerConfig::builder()
.with_client_cert_verifier(verifier)
.with_single_cert(certs, key)
.map_err(|e| rustfs_utils::certs_error(e.to_string()))?
} else {
ServerConfig::builder()
.with_no_client_auth()
.with_single_cert(certs, key)
.map_err(|e| rustfs_utils::certs_error(e.to_string()))?
};
// Configure ALPN protocol priority
server_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()];
// Enable session resumption to reduce handshake overhead for returning clients
server_config.session_storage = rustls::server::ServerSessionMemoryCache::new(10000);
// Log SNI requests
if rustfs_utils::tls_key_log() {
server_config.key_log = Arc::new(rustls::KeyLogFile::new());
}
return Ok(Some(TlsAcceptor::from(Arc::new(server_config))));
}
debug!("No valid TLS certificates found in the directory, starting with HTTP");
Ok(None)
}
#[derive(Clone)] #[derive(Clone)]
struct ConnectionContext { struct ConnectionContext {
http_server: Arc<ConnBuilder<TokioExecutor>>, http_server: Arc<ConnBuilder<TokioExecutor>>,
@@ -522,6 +495,10 @@ struct ConnectionContext {
compression_config: CompressionConfig, compression_config: CompressionConfig,
is_console: bool, is_console: bool,
readiness: Arc<GlobalReadiness>, readiness: Arc<GlobalReadiness>,
/// Pre-computed Keystone auth provider (avoids per-connection OnceLock read).
keystone_auth: Option<std::sync::Arc<rustfs_keystone::KeystoneAuthProvider>>,
/// Pre-computed trusted proxy layer (avoids per-connection is_enabled() check).
trusted_proxy_layer: Option<rustfs_trusted_proxies::TrustedProxyLayer>,
} }
/// Adapter that implements the OpenTelemetry [`Extractor`] trait for Hyper's /// Adapter that implements the OpenTelemetry [`Extractor`] trait for Hyper's
@@ -569,7 +546,7 @@ impl<'a> opentelemetry::propagation::Extractor for HeaderMapCarrier<'a> {
))] ))]
fn process_connection( fn process_connection(
socket: TcpStream, socket: TcpStream,
tls_acceptor: Option<Arc<TlsAcceptor>>, tls_acceptor: Option<Arc<TlsAcceptorHolder>>,
context: ConnectionContext, context: ConnectionContext,
graceful: Arc<GracefulShutdown>, graceful: Arc<GracefulShutdown>,
) { ) {
@@ -580,15 +557,16 @@ fn process_connection(
compression_config, compression_config,
is_console, is_console,
readiness, readiness,
keystone_auth,
trusted_proxy_layer,
} = context; } = context;
// Build services inside each connected task to avoid passing complex service types across tasks, // Build the hybrid service per-connection.
// It also ensures that each connection has an independent service instance. // Note: NodeService is not Clone (holds LocalPeerS3Client), and the SwiftService
// type is feature-gated, so we cannot pre-build the full hybrid service.
// The construction cost is negligible (struct wrapping only, no I/O).
let rpc_service = NodeServiceServer::with_interceptor(make_server(), check_auth); let rpc_service = NodeServiceServer::with_interceptor(make_server(), check_auth);
// Wrap S3 service with Swift service to handle Swift API requests
// Swift API is only available when compiled with the 'swift' feature
// When enabled, Swift routes are handled at /v1/AUTH_* paths by default
#[cfg(feature = "swift")] #[cfg(feature = "swift")]
let http_service = SwiftService::new(true, None, s3_service); let http_service = SwiftService::new(true, None, s3_service);
#[cfg(not(feature = "swift"))] #[cfg(not(feature = "swift"))]
@@ -607,6 +585,26 @@ fn process_connection(
None None
} }
}; };
// ── Canonical Middleware Stack Order (outermost → innermost) ──
// This order MUST be preserved across refactorings.
// Only AddExtensionLayer (layers 1-2) are per-connection; layers 3-15 are stateless.
//
// 1. AddExtensionLayer<RemoteAddr> — per-connection peer address
// 2. AddExtensionLayer<SocketAddr> — per-connection raw socket addr (TrustedProxy)
// 3. TrustedProxyLayer — conditional, parses X-Forwarded-For
// 4. SetRequestIdLayer — generates X-Request-ID
// 5. AdminChunkedContentLengthCompatLayer — admin API compat
// 6. CatchPanicLayer — panic → 500
// 7. ReadinessGateLayer — blocks until ready
// 8. KeystoneAuthLayer — X-Auth-Token validation
// 9. TraceLayer — request/response tracing + metrics
// 10. PropagateRequestIdLayer — X-Request-ID → response
// 11. PathCategoryInjectionLayer — injects path category for compression
// 12. CompressionLayer — response compression (whitelist, path-aware)
// 13. ObjectAttributesEtagFixLayer — ETag fix for GetObjectAttributes
// 14. ConditionalCorsLayer — S3 API CORS
// 15. RedirectLayer — console redirect (conditional)
// ─────────────────────────────────────────────────────────────
let hybrid_service = ServiceBuilder::new() let hybrid_service = ServiceBuilder::new()
// NOTE: Both extension types are intentionally inserted to maintain compatibility: // NOTE: Both extension types are intentionally inserted to maintain compatibility:
// 1. `Option<RemoteAddr>` - Used by existing admin/storage handlers throughout the codebase // 1. `Option<RemoteAddr>` - Used by existing admin/storage handlers throughout the codebase
@@ -618,11 +616,8 @@ fn process_connection(
.option_layer(remote_addr.map(|ra| AddExtensionLayer::new(ra.0))) .option_layer(remote_addr.map(|ra| AddExtensionLayer::new(ra.0)))
// Add TrustedProxyLayer to handle X-Forwarded-For and other proxy headers // 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 // This should be placed before TraceLayer so that logs reflect the real client IP
.option_layer(if rustfs_trusted_proxies::is_enabled() { // Pre-computed in ConnectionContext to avoid per-connection is_enabled() check.
Some(rustfs_trusted_proxies::layer().clone()) .option_layer(trusted_proxy_layer)
} else {
None
})
.layer(SetRequestIdLayer::x_request_id(MakeRequestUuid)) .layer(SetRequestIdLayer::x_request_id(MakeRequestUuid))
.layer(AdminChunkedContentLengthCompatLayer) .layer(AdminChunkedContentLengthCompatLayer)
.layer(CatchPanicLayer::new()) .layer(CatchPanicLayer::new())
@@ -632,10 +627,8 @@ fn process_connection(
// Add Keystone authentication middleware // Add Keystone authentication middleware
// This validates X-Auth-Token headers and stores credentials in task-local storage // This validates X-Auth-Token headers and stores credentials in task-local storage
// Must be placed AFTER ReadinessGateLayer but BEFORE business logic // Must be placed AFTER ReadinessGateLayer but BEFORE business logic
.layer({ // Pre-computed in ConnectionContext to avoid per-connection OnceLock read.
let keystone_auth = auth_keystone::get_keystone_auth(); .layer(KeystoneAuthLayer::new(keystone_auth))
KeystoneAuthLayer::new(keystone_auth)
})
.layer( .layer(
TraceLayer::new_for_http() TraceLayer::new_for_http()
.make_span_with(|request: &HttpRequest<_>| { .make_span_with(|request: &HttpRequest<_>| {
@@ -702,13 +695,27 @@ fn process_connection(
debug!("http response generated in {:?}", latency) debug!("http response generated in {:?}", latency)
}) })
.on_body_chunk(|chunk: &Bytes, latency: Duration, span: &Span| { .on_body_chunk(|chunk: &Bytes, latency: Duration, span: &Span| {
let _enter = span.enter(); #[cfg(feature = "tracing-chunk-debug")]
histogram!("rustfs.request.body.len").record(chunk.len() as f64); {
debug!("http body sending {} bytes in {:?}", chunk.len(), latency); let _enter = span.enter();
histogram!("rustfs.request.body.len").record(chunk.len() as f64);
debug!("http body sending {} bytes in {:?}", chunk.len(), latency);
}
#[cfg(not(feature = "tracing-chunk-debug"))]
{
let _ = (chunk, latency, span);
}
}) })
.on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, span: &Span| { .on_eos(|_trailers: Option<&HeaderMap>, stream_duration: Duration, span: &Span| {
let _enter = span.enter(); #[cfg(feature = "tracing-chunk-debug")]
debug!("http stream closed after {:?}", stream_duration) {
let _enter = span.enter();
debug!("http stream closed after {:?}", stream_duration);
}
#[cfg(not(feature = "tracing-chunk-debug"))]
{
let _ = (_trailers, stream_duration, span);
}
}) })
.on_failure(|_error, latency: Duration, span: &Span| { .on_failure(|_error, latency: Duration, span: &Span| {
let _enter = span.enter(); let _enter = span.enter();
@@ -719,7 +726,8 @@ fn process_connection(
.layer(PropagateRequestIdLayer::x_request_id()) .layer(PropagateRequestIdLayer::x_request_id())
// Compress responses based on whitelist configuration // Compress responses based on whitelist configuration
// Only compresses when enabled and matches configured extensions/MIME types // Only compresses when enabled and matches configured extensions/MIME types
.layer(CompressionLayer::new().compress_when(CompressionPredicate::new(compression_config))) .layer(PathCategoryInjectionLayer)
.layer(CompressionLayer::new().compress_when(PathAwareCompressionPredicate::new(compression_config)))
.layer(ObjectAttributesEtagFixLayer) .layer(ObjectAttributesEtagFixLayer)
// Conditional CORS layer: only applies to S3 API requests (not Admin, not Console) // Conditional CORS layer: only applies to S3 API requests (not Admin, not Console)
// Admin has its own CORS handling in router.rs // Admin has its own CORS handling in router.rs
@@ -733,12 +741,13 @@ fn process_connection(
let hybrid_service = TowerToHyperService::new(hybrid_service); let hybrid_service = TowerToHyperService::new(hybrid_service);
// Decide whether to handle HTTPS or HTTP connections based on the existence of TLS Acceptor // Decide whether to handle HTTPS or HTTP connections based on the existence of TLS Acceptor
if let Some(acceptor) = tls_acceptor { if let Some(holder) = tls_acceptor {
debug!("TLS handshake start"); debug!("TLS handshake start");
let peer_addr = socket let peer_addr = socket
.peer_addr() .peer_addr()
.ok() .ok()
.map_or_else(|| "unknown".to_string(), |addr| addr.to_string()); .map_or_else(|| "unknown".to_string(), |addr| addr.to_string());
let acceptor = holder.get();
match acceptor.accept(socket).await { match acceptor.accept(socket).await {
Ok(tls_socket) => { Ok(tls_socket) => {
debug!("TLS handshake successful"); debug!("TLS handshake successful");
@@ -749,32 +758,26 @@ fn process_connection(
} }
} }
Err(err) => { Err(err) => {
// Detailed analysis of the reasons why the TLS handshake fails
let err_str = err.to_string(); let err_str = err.to_string();
let mut key_failure_type_str: &str = "UNKNOWN"; let kind = TlsHandshakeFailureKind::classify(&err_str);
if err_str.contains("unexpected EOF") || err_str.contains("handshake eof") { match kind {
warn!(peer_addr = %peer_addr, "TLS handshake failed. If this client needs HTTP, it should connect to the HTTP port instead"); TlsHandshakeFailureKind::UnexpectedEof => {
key_failure_type_str = "UNEXPECTED_EOF"; warn!(peer_addr = %peer_addr, "TLS handshake failed (unexpected EOF). If this client needs HTTP, it should connect to the HTTP port instead");
} else if err_str.contains("protocol version") { }
error!( TlsHandshakeFailureKind::ProtocolVersion => {
peer_addr = %peer_addr, error!(peer_addr = %peer_addr, "TLS handshake failed (protocol version mismatch): {}", err);
"TLS handshake failed due to protocol version mismatch: {}", err }
); TlsHandshakeFailureKind::Certificate => {
key_failure_type_str = "PROTOCOL_VERSION"; error!(peer_addr = %peer_addr, "TLS handshake failed (certificate issue): {}", err);
} else if err_str.contains("certificate") { }
error!( TlsHandshakeFailureKind::Alert => {
peer_addr = %peer_addr, error!(peer_addr = %peer_addr, "TLS handshake failed (alert): {}", err);
"TLS handshake failed due to certificate issues: {}", err }
); TlsHandshakeFailureKind::Unknown => {
key_failure_type_str = "CERTIFICATE"; error!(peer_addr = %peer_addr, "TLS handshake failed: {}", err);
} else { }
error!(
peer_addr = %peer_addr,
"TLS handshake failed: {}", err
);
} }
counter!("rustfs_tls_handshake_failures", &[("key_failure_type", key_failure_type_str)]).increment(1); counter!("rustfs_tls_handshake_failures", &[("failure_type", kind.as_str())]).increment(1);
// Record detailed diagnostic information
debug!( debug!(
peer_addr = %peer_addr, peer_addr = %peer_addr,
error_type = %std::any::type_name_of_val(&err), error_type = %std::any::type_name_of_val(&err),
@@ -914,6 +917,68 @@ mod tests {
use http::HeaderMap; use http::HeaderMap;
use opentelemetry::propagation::Extractor; use opentelemetry::propagation::Extractor;
/// Baseline constants — reference the authoritative config defaults.
/// If a config default changes, tests automatically follow.
mod baseline {
use rustfs_config::{
DEFAULT_H2_INITIAL_CONN_WINDOW_SIZE, DEFAULT_H2_INITIAL_STREAM_WINDOW_SIZE, DEFAULT_H2_MAX_FRAME_SIZE,
DEFAULT_H2_MAX_HEADER_LIST_SIZE, DEFAULT_HTTP1_HEADER_READ_TIMEOUT, DEFAULT_HTTP1_MAX_BUF_SIZE,
};
/// Number of middleware layers in the canonical stack order (see http.rs).
/// Layers 1-2 are per-connection (AddExtension), 3-15 are stateless.
pub const MIDDLEWARE_LAYER_COUNT: usize = 15;
/// Current HTTP/2 defaults (from rustfs_config).
pub const H2_INITIAL_STREAM_WINDOW_SIZE: u32 = DEFAULT_H2_INITIAL_STREAM_WINDOW_SIZE;
pub const H2_INITIAL_CONN_WINDOW_SIZE: u32 = DEFAULT_H2_INITIAL_CONN_WINDOW_SIZE;
pub const H2_MAX_FRAME_SIZE: u32 = DEFAULT_H2_MAX_FRAME_SIZE;
pub const H2_MAX_HEADER_LIST_SIZE: u32 = DEFAULT_H2_MAX_HEADER_LIST_SIZE;
/// Current HTTP/1.1 defaults (from rustfs_config).
pub const HTTP1_HEADER_READ_TIMEOUT_SECS: u64 = DEFAULT_HTTP1_HEADER_READ_TIMEOUT;
pub const HTTP1_MAX_BUF_SIZE: usize = DEFAULT_HTTP1_MAX_BUF_SIZE;
/// Post-accept socket syscalls after T03 optimization.
/// Linux: 1 (TCP_QUICKACK only). Other platforms: 0.
#[cfg(target_os = "linux")]
pub const POST_ACCEPT_SYSCALL_COUNT_LINUX: usize = 1;
#[cfg(not(target_os = "linux"))]
pub const POST_ACCEPT_SYSCALL_COUNT_OTHER: usize = 0;
}
#[test]
fn test_baseline_h2_constants() {
use rustfs_config::{
DEFAULT_H2_INITIAL_CONN_WINDOW_SIZE, DEFAULT_H2_INITIAL_STREAM_WINDOW_SIZE, DEFAULT_H2_MAX_FRAME_SIZE,
DEFAULT_H2_MAX_HEADER_LIST_SIZE,
};
assert_eq!(baseline::H2_INITIAL_STREAM_WINDOW_SIZE, DEFAULT_H2_INITIAL_STREAM_WINDOW_SIZE);
assert_eq!(baseline::H2_INITIAL_CONN_WINDOW_SIZE, DEFAULT_H2_INITIAL_CONN_WINDOW_SIZE);
assert_eq!(baseline::H2_MAX_FRAME_SIZE, DEFAULT_H2_MAX_FRAME_SIZE);
assert_eq!(baseline::H2_MAX_HEADER_LIST_SIZE, DEFAULT_H2_MAX_HEADER_LIST_SIZE);
}
#[test]
fn test_baseline_http1_constants() {
use rustfs_config::{DEFAULT_HTTP1_HEADER_READ_TIMEOUT, DEFAULT_HTTP1_MAX_BUF_SIZE};
assert_eq!(baseline::HTTP1_HEADER_READ_TIMEOUT_SECS, DEFAULT_HTTP1_HEADER_READ_TIMEOUT);
assert_eq!(baseline::HTTP1_MAX_BUF_SIZE, DEFAULT_HTTP1_MAX_BUF_SIZE);
}
#[test]
fn test_baseline_middleware_count() {
assert_eq!(baseline::MIDDLEWARE_LAYER_COUNT, 15);
}
#[test]
fn test_baseline_post_accept_syscall_count() {
#[cfg(target_os = "linux")]
assert_eq!(baseline::POST_ACCEPT_SYSCALL_COUNT_LINUX, 1);
#[cfg(not(target_os = "linux"))]
assert_eq!(baseline::POST_ACCEPT_SYSCALL_COUNT_OTHER, 0);
}
#[test] #[test]
fn test_headermap_carrier_new() { fn test_headermap_carrier_new() {
let headers = HeaderMap::new(); let headers = HeaderMap::new();
rustfs/src/server/mod.rs
+1 -2
View File
@@ -13,7 +13,6 @@
// limitations under the License. // limitations under the License.
mod audit; mod audit;
mod cert;
mod compress; mod compress;
pub mod cors; pub mod cors;
mod event; mod event;
@@ -24,9 +23,9 @@ mod prefix;
mod readiness; mod readiness;
mod runtime; mod runtime;
mod service_state; mod service_state;
pub(crate) mod tls_material;
pub(crate) use audit::{start_audit_system, stop_audit_system}; pub(crate) use audit::{start_audit_system, stop_audit_system};
pub(crate) use cert::init_cert;
pub(crate) use event::{init_event_notifier, shutdown_event_notifier}; pub(crate) use event::{init_event_notifier, shutdown_event_notifier};
pub(crate) use http::start_http_server; pub(crate) use http::start_http_server;
pub(crate) use prefix::*; pub(crate) use prefix::*;
rustfs/src/server/tls_material.rs
+510
View File
@@ -0,0 +1,510 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Unified TLS Material Snapshot
//!
//! Provides a single loading point for all TLS materials, eliminating duplicate
//! directory scanning and PEM parsing between outbound and inbound paths.
//!
//! Usage:
//! 1. Call `TlsMaterialSnapshot::load(tls_path)` once at startup.
//! 2. Call `snapshot.apply_outbound()` to set global root CAs and mTLS identity.
//! 3. Call `snapshot.build_tls_acceptor(tls_path)` to build the server TLS acceptor.
use rustfs_common::{MtlsIdentityPem, set_global_mtls_identity, set_global_root_cert};
use rustfs_config::{
DEFAULT_TLS_RELOAD_ENABLE, DEFAULT_TLS_RELOAD_INTERVAL, DEFAULT_TRUST_LEAF_CERT_AS_CA, DEFAULT_TRUST_SYSTEM_CA,
ENV_MTLS_CLIENT_CERT, ENV_MTLS_CLIENT_KEY, ENV_TLS_RELOAD_ENABLE, ENV_TLS_RELOAD_INTERVAL, ENV_TRUST_LEAF_CERT_AS_CA,
ENV_TRUST_SYSTEM_CA, RUSTFS_CA_CERT, RUSTFS_CLIENT_CERT_FILENAME, RUSTFS_CLIENT_KEY_FILENAME, RUSTFS_PUBLIC_CERT,
RUSTFS_TLS_CERT, RUSTFS_TLS_KEY,
};
use rustfs_utils::{get_env_bool, get_env_opt_str};
use rustls::pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::RwLock;
use std::time::Duration;
use tokio_rustls::TlsAcceptor;
use tracing::{debug, info, warn};
/// System CA certificate search paths (platform-specific).
const SYSTEM_CA_PATHS: &[&str] = &[
"/etc/ssl/certs/ca-certificates.crt", // Debian/Ubuntu/Alpine
"/etc/pki/tls/certs/ca-bundle.crt", // Fedora/RHEL/CentOS
"/etc/ssl/ca-bundle.pem", // OpenSUSE
"/etc/pki/tls/cacert.pem", // OpenELEC
"/etc/ssl/cert.pem", // macOS/FreeBSD
"/usr/local/etc/openssl/cert.pem", // macOS/Homebrew OpenSSL
"/usr/local/share/certs/ca-root-nss.crt", // FreeBSD
"/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem", // RHEL
"/usr/share/pki/ca-trust-legacy/ca-bundle.legacy.crt", // RHEL legacy
];
/// Outbound TLS material for client connections (inter-node RPC).
#[derive(Debug, Clone)]
pub struct OutboundTlsMaterial {
/// Concatenated PEM-encoded root CA certificates.
pub root_ca_pem: Vec<u8>,
/// Optional mTLS client identity.
pub mtls_identity: Option<MtlsIdentityPem>,
}
/// Complete TLS material snapshot loaded once at startup.
#[derive(Debug)]
pub struct TlsMaterialSnapshot {
/// Material for outbound client connections.
pub outbound: OutboundTlsMaterial,
/// Whether any server certificates were found.
pub has_server_certs: bool,
}
impl TlsMaterialSnapshot {
/// Load all TLS materials from the given directory.
///
/// This is the single entry point that replaces both the old
/// `cert.rs::init_cert()` and `http.rs::setup_tls_acceptor()` loading logic.
pub async fn load(tls_path: &str) -> Result<Self, TlsMaterialError> {
if tls_path.is_empty() {
info!("No TLS path configured; skipping TLS material loading");
return Ok(Self::empty());
}
let tls_dir = PathBuf::from(tls_path);
// Load outbound material (root CAs + mTLS identity)
let outbound = load_outbound_material(&tls_dir).await?;
// Check if server certs exist (actual loading happens in build_tls_acceptor)
let has_server_certs = has_server_certificates(tls_path).await;
Ok(Self {
outbound,
has_server_certs,
})
}
/// Apply outbound material to global state (root CAs, mTLS identity).
pub async fn apply_outbound(&self) {
if !self.outbound.root_ca_pem.is_empty() {
set_global_root_cert(self.outbound.root_ca_pem.clone()).await;
info!("Configured custom root certificates for inter-node communication");
}
set_global_mtls_identity(self.outbound.mtls_identity.clone()).await;
}
/// Build a `TlsAcceptorHolder` from the loaded snapshot.
///
/// This is the single place that constructs the server `ServerConfig`,
/// handling both multi-cert (SNI resolver) and single-cert fallback.
/// Returns `None` if no TLS certificates are available.
pub async fn build_tls_acceptor(&self, tls_path: &str) -> Result<Option<Arc<TlsAcceptorHolder>>, TlsMaterialError> {
if tls_path.is_empty() || !self.has_server_certs {
return Ok(None);
}
let mtls_verifier = rustfs_utils::build_webpki_client_verifier(tls_path)
.map_err(|e| TlsMaterialError::Io(format!("build mTLS verifier: {e}")))?;
// Try multi-cert (SNI) first
match rustfs_utils::load_all_certs_from_directory(tls_path) {
Ok(cert_key_pairs) if !cert_key_pairs.is_empty() => match rustfs_utils::create_multi_cert_resolver(cert_key_pairs) {
Ok(resolver) => {
let config = build_server_config(ServerCertSource::Resolver(Arc::new(resolver)), mtls_verifier)?;
info!("Created TLS acceptor with SNI resolver");
let acceptor = Arc::new(TlsAcceptor::from(Arc::new(config)));
return Ok(Some(Arc::new(TlsAcceptorHolder::new(acceptor))));
}
Err(e) => warn!("Failed to build multi-cert resolver: {}, falling back to single-cert", e),
},
Ok(_) => debug!("No valid multi-cert directory structure found"),
Err(_) => debug!("load_all_certs_from_directory failed, trying single-cert fallback"),
}
// Fallback: single cert
let key_path = format!("{tls_path}/{RUSTFS_TLS_KEY}");
let cert_path = format!("{tls_path}/{RUSTFS_TLS_CERT}");
if tokio::try_join!(tokio::fs::metadata(&key_path), tokio::fs::metadata(&cert_path)).is_ok() {
let certs = rustfs_utils::load_certs(&cert_path).map_err(|e| TlsMaterialError::Io(format!("load certs: {e}")))?;
let key = rustfs_utils::load_private_key(&key_path).map_err(|e| TlsMaterialError::Io(format!("load key: {e}")))?;
let config = build_server_config(ServerCertSource::SingleCert { certs, key }, mtls_verifier)?;
info!("Created TLS acceptor with single certificate");
let acceptor = Arc::new(TlsAcceptor::from(Arc::new(config)));
return Ok(Some(Arc::new(TlsAcceptorHolder::new(acceptor))));
}
debug!("No valid TLS certificates found, starting with HTTP");
Ok(None)
}
fn empty() -> Self {
Self {
outbound: OutboundTlsMaterial {
root_ca_pem: Vec::new(),
mtls_identity: None,
},
has_server_certs: false,
}
}
}
// ── Server Config Construction ──
/// Certificate source for building a `ServerConfig`.
enum ServerCertSource {
/// Pre-built SNI resolver from multi-cert directory.
Resolver(Arc<dyn rustls::server::ResolvesServerCert + Send + Sync>),
/// Single certificate/key pair.
SingleCert {
certs: Vec<CertificateDer<'static>>,
key: PrivateKeyDer<'static>,
},
}
/// Build a `ServerConfig` with standardized ALPN, session cache, and key log settings.
///
/// This is the single place for `ServerConfig` construction, used by both
/// initial startup and hot-reload.
fn build_server_config(
cert_source: ServerCertSource,
mtls_verifier: Option<Arc<dyn rustls::server::danger::ClientCertVerifier>>,
) -> Result<rustls::ServerConfig, TlsMaterialError> {
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)
} else {
rustls::ServerConfig::builder()
.with_no_client_auth()
.with_cert_resolver(resolver)
}
}
ServerCertSource::SingleCert { certs, key } => {
if let Some(verifier) = mtls_verifier {
rustls::ServerConfig::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()
.with_no_client_auth()
.with_single_cert(certs, key)
.map_err(|e| TlsMaterialError::Io(format!("configure single cert: {e}")))?
}
}
};
config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec(), b"http/1.0".to_vec()];
config.session_storage = rustls::server::ServerSessionMemoryCache::new(10000);
if rustfs_utils::tls_key_log() {
config.key_log = Arc::new(rustls::KeyLogFile::new());
}
Ok(config)
}
// ── Outbound Material Loading ──
/// Load root CA certificates and mTLS identity for outbound connections.
async fn load_outbound_material(tls_dir: &Path) -> Result<OutboundTlsMaterial, TlsMaterialError> {
let mut root_ca_pem = Vec::new();
// 1. Optional: load leaf certs as root CAs
if get_env_bool(ENV_TRUST_LEAF_CERT_AS_CA, DEFAULT_TRUST_LEAF_CERT_AS_CA)
&& load_cert_file_by_name(tls_dir, RUSTFS_TLS_CERT, &mut root_ca_pem).await
{
info!("Loaded leaf certificate(s) as root CA as per RUSTFS_TRUST_LEAF_CERT_AS_CA");
}
// 2. Load public.crt and ca.crt
load_cert_file(&tls_dir.join(RUSTFS_PUBLIC_CERT), &mut root_ca_pem, "CA certificate").await;
load_cert_file(&tls_dir.join(RUSTFS_CA_CERT), &mut root_ca_pem, "CA certificate").await;
// 3. Optional: load system root CAs
if get_env_bool(ENV_TRUST_SYSTEM_CA, DEFAULT_TRUST_SYSTEM_CA) {
let mut system_loaded = false;
for path in SYSTEM_CA_PATHS {
if load_cert_file(Path::new(path), &mut root_ca_pem, "system root certificates").await {
system_loaded = true;
info!("Loaded system root certificates from {}", path);
break;
}
}
if !system_loaded {
debug!("Could not find system root certificates in common locations.");
}
} else {
info!("Loading system root certificates disabled via RUSTFS_TRUST_SYSTEM_CA");
}
// 4. Load optional mTLS identity
let mtls_identity = load_mtls_identity(tls_dir).await?;
Ok(OutboundTlsMaterial {
root_ca_pem,
mtls_identity,
})
}
/// Quick check whether server certificate files exist in the TLS directory.
async fn has_server_certificates(tls_path: &str) -> bool {
if tokio::fs::metadata(tls_path).await.is_err() {
return false;
}
// Check for multi-cert directory structure OR single cert files
if rustfs_utils::load_all_certs_from_directory(tls_path).is_ok_and(|p| !p.is_empty()) {
return true;
}
let key_path = format!("{tls_path}/{RUSTFS_TLS_KEY}");
let cert_path = format!("{tls_path}/{RUSTFS_TLS_CERT}");
tokio::try_join!(tokio::fs::metadata(&key_path), tokio::fs::metadata(&cert_path)).is_ok()
}
/// Load mTLS client identity from the TLS directory.
async fn load_mtls_identity(tls_dir: &Path) -> Result<Option<MtlsIdentityPem>, TlsMaterialError> {
let client_cert_path = match get_env_opt_str(ENV_MTLS_CLIENT_CERT) {
Some(p) => PathBuf::from(p),
None => tls_dir.join(RUSTFS_CLIENT_CERT_FILENAME),
};
let client_key_path = match get_env_opt_str(ENV_MTLS_CLIENT_KEY) {
Some(p) => PathBuf::from(p),
None => tls_dir.join(RUSTFS_CLIENT_KEY_FILENAME),
};
if !client_cert_path.exists() || !client_key_path.exists() {
info!(
"mTLS client identity not configured (missing {:?} and/or {:?}); proceeding with server-only TLS",
client_cert_path, client_key_path
);
return Ok(None);
}
let cert_pem = tokio::fs::read(&client_cert_path)
.await
.map_err(|e| TlsMaterialError::Io(format!("read client cert {client_cert_path:?}: {e}")))?;
let key_pem = tokio::fs::read(&client_key_path)
.await
.map_err(|e| TlsMaterialError::Io(format!("read client key {client_key_path:?}: {e}")))?;
// Validate parse-ability
let mut reader = std::io::Cursor::new(&cert_pem);
if CertificateDer::pem_reader_iter(&mut reader).next().is_none() {
return Err(TlsMaterialError::Parse("no valid certificate in client cert PEM".into()));
}
let mut reader = std::io::Cursor::new(&key_pem);
PrivateKeyDer::from_pem_reader(&mut reader).map_err(|e| TlsMaterialError::Parse(format!("invalid client key PEM: {e}")))?;
info!("Loaded mTLS client identity cert={:?} key={:?}", client_cert_path, client_key_path);
Ok(Some(MtlsIdentityPem { cert_pem, key_pem }))
}
/// Load a single certificate file and append PEM data.
/// Returns true if the file was successfully loaded.
async fn load_cert_file(path: &Path, pem_data: &mut Vec<u8>, desc: &str) -> bool {
if tokio::fs::metadata(path).await.is_err() {
debug!("{} file not found at {:?}", desc, path);
return false;
}
match tokio::fs::read(path).await {
Ok(data) => {
pem_data.extend_from_slice(&data);
pem_data.push(b'\n');
info!("Loaded {} from {:?}", desc, path);
true
}
Err(e) => {
debug!("Failed to read {} from {:?}: {}", desc, path, e);
false
}
}
}
/// Search for and load certificate files matching `cert_name` in the directory
/// and one level of subdirectories.
/// Returns `true` if at least one matching file was loaded.
async fn load_cert_file_by_name(dir: &Path, cert_name: &str, pem_data: &mut Vec<u8>) -> bool {
let Ok(mut rd) = tokio::fs::read_dir(dir).await else {
debug!("Certificate directory not found: {}", dir.display());
return false;
};
let mut loaded = false;
while let Ok(Some(entry)) = rd.next_entry().await {
let Ok(ft) = entry.file_type().await else { continue };
if ft.is_file() {
let fname = entry.file_name().to_string_lossy().to_string();
if fname == cert_name && load_cert_file(&entry.path(), pem_data, "certificate").await {
loaded = true;
}
} else if ft.is_dir() {
// Only check direct subdirectories (one level deep)
if let Ok(mut sub_rd) = tokio::fs::read_dir(&entry.path()).await {
while let Ok(Some(sub_entry)) = sub_rd.next_entry().await {
if let Ok(sub_ft) = sub_entry.file_type().await
&& sub_ft.is_file()
{
let fname = sub_entry.file_name().to_string_lossy().to_string();
if fname == cert_name && load_cert_file(&sub_entry.path(), pem_data, "certificate").await {
loaded = true;
}
}
}
}
}
}
loaded
}
/// Errors that can occur during TLS material loading.
#[derive(Debug)]
pub enum TlsMaterialError {
/// I/O error (file read, directory access).
Io(String),
/// PEM parsing error.
Parse(String),
}
impl std::fmt::Display for TlsMaterialError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
TlsMaterialError::Io(msg) => write!(f, "TLS material I/O error: {msg}"),
TlsMaterialError::Parse(msg) => write!(f, "TLS material parse error: {msg}"),
}
}
}
impl std::error::Error for TlsMaterialError {}
// ── TLS Handshake Error Classification ──
/// Structured classification of TLS handshake failures.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum TlsHandshakeFailureKind {
UnexpectedEof,
ProtocolVersion,
Certificate,
Alert,
Unknown,
}
impl TlsHandshakeFailureKind {
/// Classify a TLS accept error into a structured failure kind.
pub(crate) fn classify(err_msg: &str) -> Self {
if err_msg.contains("unexpected EOF") || err_msg.contains("handshake eof") {
Self::UnexpectedEof
} else if err_msg.contains("protocol version") {
Self::ProtocolVersion
} else if err_msg.contains("certificate") || err_msg.contains("invalid peer certificate") {
Self::Certificate
} else if err_msg.contains("alert") {
Self::Alert
} else {
Self::Unknown
}
}
/// Metric label string for Prometheus.
pub(crate) fn as_str(self) -> &'static str {
match self {
Self::UnexpectedEof => "UNEXPECTED_EOF",
Self::ProtocolVersion => "PROTOCOL_VERSION",
Self::Certificate => "CERTIFICATE",
Self::Alert => "ALERT",
Self::Unknown => "UNKNOWN",
}
}
}
// ── TLS Acceptor Holder (for hot reload) ──
/// Holds the current TLS acceptor and supports atomic swap for certificate rotation.
///
/// Uses `RwLock` so that multiple readers (per-connection `get()` calls)
/// do not block each other. The write lock is held only briefly during swap.
pub(crate) struct TlsAcceptorHolder {
current: RwLock<Arc<TlsAcceptor>>,
}
impl TlsAcceptorHolder {
pub(crate) fn new(acceptor: Arc<TlsAcceptor>) -> Self {
Self {
current: RwLock::new(acceptor),
}
}
/// Get the current TLS acceptor for handling a new connection.
#[inline]
pub(crate) fn get(&self) -> Arc<TlsAcceptor> {
match self.current.read() {
Ok(guard) => guard.clone(),
Err(poisoned) => poisoned.into_inner().clone(),
}
}
/// Atomically replace the TLS acceptor with a new one.
fn swap(&self, new_holder: &TlsAcceptorHolder) {
let new_acceptor = new_holder.get();
match self.current.write() {
Ok(mut guard) => *guard = new_acceptor,
Err(poisoned) => {
let mut guard = poisoned.into_inner();
*guard = new_acceptor;
}
}
}
}
/// Spawn a background task that periodically checks for TLS certificate changes.
pub(crate) fn spawn_reload_loop(tls_path: String, holder: Arc<TlsAcceptorHolder>) {
let enabled = get_env_bool(ENV_TLS_RELOAD_ENABLE, DEFAULT_TLS_RELOAD_ENABLE);
if !enabled {
debug!("TLS certificate hot reload is disabled (set {}=1 to enable)", ENV_TLS_RELOAD_ENABLE);
return;
}
let interval_secs = rustfs_utils::get_env_u64(ENV_TLS_RELOAD_INTERVAL, DEFAULT_TLS_RELOAD_INTERVAL).max(5);
info!("TLS certificate hot reload enabled, checking every {}s", interval_secs);
tokio::spawn(async move {
let mut interval = tokio::time::interval(Duration::from_secs(interval_secs));
loop {
interval.tick().await;
match TlsMaterialSnapshot::load(&tls_path).await {
Ok(snapshot) => {
// Always refresh outbound material (root CAs, mTLS identity) on reload.
snapshot.apply_outbound().await;
match snapshot.build_tls_acceptor(&tls_path).await {
Ok(Some(new_holder)) => {
info!("TLS certificates reloaded successfully");
holder.swap(&new_holder);
}
Ok(None) => debug!("TLS reload: no server certificates found in directory, skipping"),
Err(e) => warn!("TLS certificate reload failed (will retry): {}", e),
}
}
Err(e) => {
warn!("TLS material reload failed (will retry): {}", e);
}
}
}
});
}