From 6b8c1f0776fe62b6775a78f75dd8d7353131adf9 Mon Sep 17 00:00:00 2001 From: overtrue Date: Wed, 2 Sep 2026 21:44:57 +0800 Subject: [PATCH] refactor(ecstore): extract shared remote S3 client builder Move the aws_sdk_s3 client construction out of bucket_target_sys into bucket/remote_s3_client.rs: endpoint assembly, credential provider, path-style selection, custom CA / skip-TLS transports and the outbound SSRF gate now build from a neutral RemoteS3EndpointSpec so replication targets and the upcoming on-demand migration source client share one policy. Replication builds its client through From<&BucketTarget>; the gate keeps its relaxed semantics (private allowed, loopback only behind RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET) verbatim. The builder also gains optional connect/read timeouts and a User-Agent suffix interceptor, both unset for replication. Refs rustfs/backlog#2149 --- crates/ecstore/src/api/mod.rs | 7 + .../ecstore/src/bucket/bucket_target_sys.rs | 594 +++---------- crates/ecstore/src/bucket/mod.rs | 1 + crates/ecstore/src/bucket/remote_s3_client.rs | 789 ++++++++++++++++++ scripts/error-other-format-baseline.txt | 1 - 5 files changed, 922 insertions(+), 470 deletions(-) create mode 100644 crates/ecstore/src/bucket/remote_s3_client.rs diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 4d0bfc1bd..6d3d3bb60 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -199,6 +199,13 @@ pub mod bucket { } } + pub mod remote_s3_client { + pub use crate::bucket::remote_s3_client::{ + PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, build_remote_s3_client, + validate_remote_endpoint, + }; + } + pub mod replication { pub use crate::bucket::replication::replication_pool::{ DurableMrfBacklogSummary, DurableMrfBucketBacklog, DurableMrfTargetBacklog, MrfBacklogObservabilitySummary, diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 7daddcefd..9ed77d446 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -15,17 +15,13 @@ use crate::bucket::metadata::BucketMetadata; use crate::bucket::metadata_sys::get_bucket_targets_config; use crate::bucket::metadata_sys::get_replication_config; +use crate::bucket::remote_s3_client::{PathStyle, RemoteCredentials, RemoteS3EndpointSpec, build_remote_s3_client}; use crate::bucket::replication::{ReplicationStatusType, ReplicationTargetConfigBridge}; use crate::bucket::target::ARN; use crate::bucket::target::BucketTargetType; use crate::bucket::target::{self, BucketTarget, BucketTargets, Credentials}; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::runtime::sources as runtime_sources; -use aws_credential_types::Credentials as SdkCredentials; -use aws_credential_types::provider::{ProvideCredentials, error::CredentialsError, future}; -use aws_sdk_s3::config::Region as SdkRegion; -use aws_sdk_s3::config::RequestChecksumCalculation; -use aws_sdk_s3::config::SharedHttpClient; use aws_sdk_s3::error::ProvideErrorMetadata; use aws_sdk_s3::error::SdkError; use aws_sdk_s3::operation::complete_multipart_upload::CompleteMultipartUploadOutput; @@ -37,28 +33,17 @@ use aws_sdk_s3::operation::head_object::HeadObjectError; use aws_sdk_s3::operation::put_object_tagging::{PutObjectTaggingError, PutObjectTaggingOutput}; use aws_sdk_s3::operation::upload_part::UploadPartOutput; use aws_sdk_s3::primitives::ByteStream; +use aws_sdk_s3::types::BucketVersioningStatus; use aws_sdk_s3::types::Tagging as SdkTagging; use aws_sdk_s3::types::{ ChecksumMode, CompletedMultipartUpload, CompletedPart, ObjectLockLegalHoldStatus, ObjectLockRetentionMode, ServerSideEncryption, }; -use aws_sdk_s3::{Client as S3Client, Config as S3Config, operation::head_object::HeadObjectOutput}; -use aws_sdk_s3::{config::SharedCredentialsProvider, types::BucketVersioningStatus}; -use aws_smithy_http_client::{Builder as SmithyHttpClientBuilder, tls as smithy_tls}; -use aws_smithy_runtime_api::box_error::BoxError; -use aws_smithy_runtime_api::client::http::{ - HttpConnector as SmithyHttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn, -}; -use aws_smithy_runtime_api::client::orchestrator::{HttpRequest, HttpResponse}; -use aws_smithy_runtime_api::client::result::ConnectorError; -use aws_smithy_types::body::SdkBody; +use aws_sdk_s3::{Client as S3Client, operation::head_object::HeadObjectOutput}; +use aws_smithy_runtime_api::client::orchestrator::HttpRequest; use futures::{StreamExt, stream}; -use http::{HeaderMap, HeaderName, HeaderValue, StatusCode, Uri}; -use hyper_util::client::legacy::Client as HyperClient; -use hyper_util::rt::{TokioExecutor, TokioTimer}; +use http::{HeaderMap, HeaderName, HeaderValue, StatusCode}; use reqwest::Client as HttpClient; -use rustfs_config::{DEFAULT_TRUST_LEAF_CERT_AS_CA, ENV_TRUST_LEAF_CERT_AS_CA, RUSTFS_CA_CERT, RUSTFS_TLS_CERT}; -use rustfs_utils::egress::{OutboundUrlError, validate_outbound_url}; use rustfs_utils::http::{ AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_BYPASS_GOVERNANCE, AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_MODE, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, AMZ_OBJECT_TAGGING_LOWER, AMZ_STORAGE_CLASS, AMZ_WEBSITE_REDIRECT_LOCATION, is_amz_header, @@ -70,12 +55,10 @@ use rustfs_utils::http::{ SUFFIX_SOURCE_REPLICATION_RETENTION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_TAGGING_TIMESTAMP, SUFFIX_SOURCE_VERSION_ID, insert_header, }; -use rustls_pki_types::pem::PemObject; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::error::Error; use std::fmt; -use std::path::Path; use std::str::FromStr as _; use std::sync::Arc; use std::sync::OnceLock; @@ -84,7 +67,6 @@ use std::time::{Duration, Instant, SystemTime}; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tokio::sync::Mutex; use tokio::sync::RwLock; -use tower::Service; use tracing::error; use tracing::warn; use url::Url; @@ -92,72 +74,50 @@ use uuid::Uuid; const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16; const REDACTED_CREDENTIAL: &str = ""; -const EXPIRED_REMOTE_TARGET_CREDENTIALS: &str = "remote target credentials have expired"; -#[derive(Clone)] -struct RemoteTargetCredentialsProvider { - credentials: SdkCredentials, +fn remote_credentials(credentials: &Credentials, account_id: &str) -> RemoteCredentials { + RemoteCredentials { + access_key: credentials.access_key.clone(), + secret_key: credentials.secret_key.clone(), + session_token: credentials.effective_session_token().map(str::to_string), + expiration: credentials.effective_expiration().map(SystemTime::from), + account_id: account_id.to_string(), + } } -impl RemoteTargetCredentialsProvider { - fn resolve_at(&self, now: SystemTime) -> aws_credential_types::provider::Result { - if self.credentials.expiry().is_some_and(|expiration| expiration <= now) { - return Err(CredentialsError::provider_error(std::io::Error::other(EXPIRED_REMOTE_TARGET_CREDENTIALS))); +fn target_path_style(path: &str) -> PathStyle { + match path.trim().to_ascii_lowercase().as_str() { + // Explicit DNS/virtual-hosted-style requested by user. + "dns" | "off" | "false" => PathStyle::VirtualHost, + // Explicit path-style or legacy boolean-like values. + "path" | "on" | "true" => PathStyle::Path, + // `auto` and empty are defaulted to path-style for custom S3-compatible endpoints. + "auto" | "" => PathStyle::Auto, + // Unknown values: prefer compatibility with S3-compatible services. + _ => PathStyle::Path, + } +} + +impl From<&BucketTarget> for RemoteS3EndpointSpec { + fn from(target: &BucketTarget) -> Self { + RemoteS3EndpointSpec { + endpoint: target.endpoint.clone(), + secure: target.secure, + region: target.region.clone(), + path_style: target_path_style(&target.path), + credentials: target + .credentials + .as_ref() + .map(|credentials| remote_credentials(credentials, &target.reset_id)), + skip_tls_verify: target.skip_tls_verify, + ca_cert_pem: (!target.ca_cert_pem.trim().is_empty()).then(|| target.ca_cert_pem.clone()), + connect_timeout: None, + read_timeout: None, + user_agent_suffix: "", } - Ok(self.credentials.clone()) } } -impl fmt::Debug for RemoteTargetCredentialsProvider { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("RemoteTargetCredentialsProvider") - .field("temporary", &self.credentials.session_token().is_some()) - .field("expiration", &self.credentials.expiry()) - .finish() - } -} - -impl ProvideCredentials for RemoteTargetCredentialsProvider { - fn provide_credentials<'a>(&'a self) -> future::ProvideCredentials<'a> - where - Self: 'a, - { - future::ProvideCredentials::ready(self.resolve_at(SystemTime::now())) - } - - fn fallback_on_interrupt(&self) -> Option { - self.resolve_at(SystemTime::now()).ok() - } -} - -fn remote_target_sdk_credentials( - credentials: &Credentials, - account_id: &str, - now: SystemTime, -) -> Result { - let session_token = credentials.effective_session_token(); - let expiration = credentials.effective_expiration().map(SystemTime::from); - if expiration.is_some() && session_token.is_none() { - return Err("remote target credential expiration requires a session token"); - } - if expiration.is_some_and(|expiration| expiration <= now) { - return Err(EXPIRED_REMOTE_TARGET_CREDENTIALS); - } - - let mut builder = SdkCredentials::builder() - .access_key_id(credentials.access_key.clone()) - .secret_access_key(credentials.secret_key.clone()) - .account_id(account_id.to_string()) - .provider_name("bucket_target_sys"); - if let Some(session_token) = session_token { - builder = builder.session_token(session_token.to_string()); - } - if let Some(expiration) = expiration { - builder = builder.expiry(expiration); - } - Ok(builder.build()) -} - pub type HeadObjectSdkError = Box>; pub type GetObjectSdkError = Box>; pub type GetObjectTaggingSdkError = Box>; @@ -1058,57 +1018,17 @@ impl BucketTargetSys { }); }; - let creds = remote_target_sdk_credentials(credentials, &target.reset_id, SystemTime::now()).map_err(|error| { - BucketTargetError::RemoteTargetConnectionErr { + let spec = RemoteS3EndpointSpec::from(target); + let client = build_remote_s3_client(&spec) + .await + .map_err(|err| BucketTargetError::RemoteTargetConnectionErr { bucket: target.target_bucket.clone(), access_key: credentials.access_key.clone(), - error: error.to_string(), - } - })?; - - let endpoint = if target.secure { - format!("https://{}", target.endpoint) - } else { - format!("http://{}", target.endpoint) - }; - let parsed_endpoint = Url::parse(&endpoint).map_err(|err| BucketTargetError::RemoteTargetConnectionErr { - bucket: target.target_bucket.clone(), - access_key: credentials.access_key.clone(), - error: format!("invalid target endpoint: {err}"), - })?; - validate_replication_target_endpoint(&parsed_endpoint).map_err(|err| BucketTargetError::RemoteTargetConnectionErr { - bucket: target.target_bucket.clone(), - access_key: credentials.access_key.clone(), - error: format!("target endpoint is not allowed: {err}"), - })?; - - let mut config_builder = S3Config::builder() - .endpoint_url(endpoint.clone()) - .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider { credentials: creds })) - .region(SdkRegion::new(target.region.clone())) - .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest()) - .request_checksum_calculation(replication_request_checksum_calculation()); - - if should_force_path_style(target) { - config_builder = config_builder.force_path_style(true); - } - - if let Some(http_client) = - build_aws_s3_http_client_for_target(target) - .await - .map_err(|err| BucketTargetError::RemoteTargetConnectionErr { - bucket: target.target_bucket.clone(), - access_key: credentials.access_key.clone(), - error: err.to_string(), - })? - { - config_builder = config_builder.http_client(http_client); - } - - let config = config_builder.build(); + error: err.to_string(), + })?; Ok(TargetClient { - endpoint, + endpoint: spec.endpoint_url(), credentials: target.credentials.clone(), bucket: target.target_bucket.clone(), storage_class: target.storage_class.clone(), @@ -1118,7 +1038,7 @@ impl BucketTargetSys { secure: target.secure, health_check_duration: target.health_check_duration, replicate_sync: target.replication_sync, - client: Arc::new(S3Client::from_conf(config)), + client: Arc::new(client), }) } @@ -1281,327 +1201,6 @@ impl BucketTargetSys { } } -#[derive(Debug)] -struct AcceptAnyServerCertVerifier; - -impl rustls::client::danger::ServerCertVerifier for AcceptAnyServerCertVerifier { - fn verify_server_cert( - &self, - _end_entity: &rustls_pki_types::CertificateDer<'_>, - _intermediates: &[rustls_pki_types::CertificateDer<'_>], - _server_name: &rustls_pki_types::ServerName<'_>, - _ocsp_response: &[u8], - _now: rustls_pki_types::UnixTime, - ) -> Result { - Ok(rustls::client::danger::ServerCertVerified::assertion()) - } - - fn verify_tls12_signature( - &self, - _message: &[u8], - _cert: &rustls_pki_types::CertificateDer<'_>, - _dss: &rustls::DigitallySignedStruct, - ) -> Result { - Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) - } - - fn verify_tls13_signature( - &self, - _message: &[u8], - _cert: &rustls_pki_types::CertificateDer<'_>, - _dss: &rustls::DigitallySignedStruct, - ) -> Result { - Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) - } - - fn supported_verify_schemes(&self) -> Vec { - rustls::crypto::aws_lc_rs::default_provider() - .signature_verification_algorithms - .supported_schemes() - } -} - -#[derive(Clone)] -struct TargetHyperHttpConnector { - client: HyperClient, -} - -impl fmt::Debug for TargetHyperHttpConnector { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("TargetHyperHttpConnector") - .field("client", &"** hyper client **") - .finish() - } -} - -impl SmithyHttpConnector for TargetHyperHttpConnector -where - C: Clone + Send + Sync + 'static, - C: Service, - C::Response: - hyper::rt::Read + hyper::rt::Write + hyper_util::client::legacy::connect::Connection + Send + Sync + Unpin + 'static, - C::Future: Unpin + Send + 'static, - C::Error: Into, -{ - fn call(&self, request: HttpRequest) -> HttpConnectorFuture { - let request = match request.try_into_http1x() { - Ok(request) => request, - Err(err) => return HttpConnectorFuture::ready(Err(ConnectorError::user(err.into()))), - }; - - let mut client = self.client.clone(); - let fut = client.call(request); - HttpConnectorFuture::new(async move { - let response = fut - .await - .map_err(|err| ConnectorError::io(err.into()))? - .map(SdkBody::from_body_1_x); - HttpResponse::try_from(response).map_err(|err| ConnectorError::other(err.into(), None)) - }) - } -} - -fn ensure_rustls_crypto_provider() { - if rustls::crypto::CryptoProvider::get_default().is_none() { - let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); - } -} - -fn has_custom_ca_pem(target: &BucketTarget) -> bool { - !target.ca_cert_pem.trim().is_empty() -} - -/// Env opt-in that re-enables loopback replication targets. Loopback (`127.0.0.1`, -/// `::1`, `localhost`) is a classic SSRF vector and stays rejected by default, but -/// single-host multi-instance dev setups and the e2e harness legitimately replicate -/// over loopback. Never set this in production. -const ALLOW_LOOPBACK_REPLICATION_TARGET_ENV: &str = "RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET"; - -fn loopback_replication_targets_allowed() -> bool { - std::env::var(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV) - .map(|v| v.eq_ignore_ascii_case("true") || v == "1") - .unwrap_or(false) -} - -const REPLICATION_STREAMING_CHECKSUMS_ENV: &str = "RUSTFS_REPLICATION_STREAMING_CHECKSUMS"; - -/// Streaming trailer checksums make the SDK frame request bodies as -/// `aws-chunked`; a target that does not decode that framing stores the frames -/// verbatim, silently corrupting every replica while the transfer itself -/// succeeds (#6853). Plain signed payloads are the compatible default; the env -/// knob restores trailer checksums for fleets whose targets are all known to -/// decode them. -fn replication_request_checksum_calculation() -> RequestChecksumCalculation { - if std::env::var(REPLICATION_STREAMING_CHECKSUMS_ENV) - .map(|v| v.eq_ignore_ascii_case("true") || v == "1") - .unwrap_or(false) - { - RequestChecksumCalculation::WhenSupported - } else { - RequestChecksumCalculation::WhenRequired - } -} - -fn validate_replication_target_endpoint(url: &Url) -> Result<(), OutboundUrlError> { - validate_replication_target_endpoint_inner(url, loopback_replication_targets_allowed()) -} - -fn validate_replication_target_endpoint_inner(url: &Url, allow_loopback: bool) -> Result<(), OutboundUrlError> { - match validate_outbound_url(url) { - Ok(()) => Ok(()), - // Replication targets are trusted infrastructure the operator configures, and - // legitimately live on private networks, so private addresses are always allowed. - Err(OutboundUrlError::ForbiddenHost { - reason: "private address", - .. - }) => Ok(()), - // Loopback is far higher SSRF risk, so it is allowed only under the explicit, - // off-by-default opt-in above (single-host multi-instance / the e2e harness). - Err(OutboundUrlError::ForbiddenHost { - reason: "loopback address" | "loopback host", - .. - }) if allow_loopback => Ok(()), - Err(err) => Err(err), - } -} - -fn build_insecure_aws_s3_http_client() -> SharedHttpClient { - ensure_rustls_crypto_provider(); - - let tls_config = rustls::ClientConfig::builder() - .dangerous() - .with_custom_certificate_verifier(Arc::new(AcceptAnyServerCertVerifier)) - .with_no_client_auth(); - - let https = hyper_rustls::HttpsConnectorBuilder::new() - .with_tls_config(tls_config) - .https_or_http() - .enable_http1() - .enable_http2() - .build(); - let mut client_builder = HyperClient::builder(TokioExecutor::new()); - client_builder.pool_timer(TokioTimer::new()); - let client = client_builder.build(https); - let connector = SharedHttpConnector::new(TargetHyperHttpConnector { client }); - - http_client_fn(move |_settings, _components| connector.clone()) -} - -fn validate_ca_pem_bundle(ca_cert_pem: &[u8]) -> Result<(), String> { - let certs = rustls_pki_types::CertificateDer::pem_slice_iter(ca_cert_pem) - .collect::, _>>() - .map_err(|err| format!("invalid PEM encoding: {err}"))?; - - if certs.is_empty() { - return Err("no certificates found".to_string()); - } - - // Smithy's rustls adapter defers parsing custom certificates and assumes - // they are valid when the HTTPS connector is built. Validate every DER - // certificate first so malformed configuration is reported rather than - // reaching an `expect` in the dependency. - let mut validation_store = rustls::RootCertStore::empty(); - for cert in certs { - validation_store - .add(cert) - .map_err(|err| format!("invalid X.509 certificate: {err}"))?; - } - - Ok(()) -} - -fn validate_target_ca_pem(ca_cert_pem: &str) -> Result<(), BucketTargetError> { - validate_ca_pem_bundle(ca_cert_pem.as_bytes()) - .map_err(|err| BucketTargetError::Io(std::io::Error::other(format!("invalid target CA PEM: {err}")))) -} - -fn compose_replication_trust_store(certificate_bundles: impl IntoIterator>) -> (smithy_tls::TrustStore, usize) { - // `TrustStore::default()` keeps the platform-native roots enabled. Target - // and RUSTFS_TLS_PATH certificates extend that baseline instead of - // replacing it with a target-specific trust island. - let mut trust_store = smithy_tls::TrustStore::default(); - let mut custom_bundle_count = 0; - for pem in certificate_bundles { - trust_store.add_pem_certificate(pem); - custom_bundle_count += 1; - } - - (trust_store, custom_bundle_count) -} - -fn build_aws_s3_http_client_with_trust_store(trust_store: smithy_tls::TrustStore) -> Result { - let tls_context = smithy_tls::TlsContext::builder() - .with_trust_store(trust_store) - .build() - .map_err(|err| BucketTargetError::Io(std::io::Error::other(format!("invalid target CA PEM: {err}"))))?; - - Ok(SmithyHttpClientBuilder::new() - .tls_provider(smithy_tls::Provider::rustls(smithy_tls::rustls_provider::CryptoMode::AwsLc)) - .tls_context(tls_context) - .build_https()) -} - -async fn load_tls_path_ca_bundles(tls_dir: &Path, trust_leaf_cert_as_ca: bool) -> Vec> { - let mut certificate_bundles = Vec::new(); - - let ca_path = tls_dir.join(RUSTFS_CA_CERT); - match tokio::fs::read(&ca_path).await { - Ok(pem) => match validate_ca_pem_bundle(&pem) { - Ok(()) => certificate_bundles.push(pem), - Err(err) => warn!("ignoring invalid custom CA bundle {:?} for replication client: {}", ca_path, err), - }, - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => warn!("failed to read custom CA bundle {:?} for replication client: {}", ca_path, e), - } - - if trust_leaf_cert_as_ca { - let leaf_cert_path = tls_dir.join(RUSTFS_TLS_CERT); - match tokio::fs::read(&leaf_cert_path).await { - Ok(pem) => match validate_ca_pem_bundle(&pem) { - Ok(()) => certificate_bundles.push(pem), - Err(err) => warn!( - "ignoring invalid leaf certificate {:?} for replication client trust store: {}", - leaf_cert_path, err - ), - }, - Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} - Err(e) => warn!("failed to read leaf cert {:?} for replication client trust store: {}", leaf_cert_path, e), - } - } - - certificate_bundles -} - -async fn load_configured_tls_ca_bundles() -> Vec> { - let tls_path = rustfs_utils::get_env_str(rustfs_config::ENV_RUSTFS_TLS_PATH, rustfs_config::DEFAULT_RUSTFS_TLS_PATH); - if tls_path.is_empty() { - return Vec::new(); - } - - load_tls_path_ca_bundles( - Path::new(&tls_path), - rustfs_utils::get_env_bool(ENV_TRUST_LEAF_CERT_AS_CA, DEFAULT_TRUST_LEAF_CERT_AS_CA), - ) - .await -} - -async fn build_aws_s3_http_client_from_target_ca_pem(ca_cert_pem: &str) -> Result { - validate_target_ca_pem(ca_cert_pem)?; - - let mut certificate_bundles = load_configured_tls_ca_bundles().await; - certificate_bundles.push(ca_cert_pem.as_bytes().to_vec()); - let (trust_store, _) = compose_replication_trust_store(certificate_bundles); - - build_aws_s3_http_client_with_trust_store(trust_store) -} - -async fn build_aws_s3_http_client_for_target(target: &BucketTarget) -> Result, BucketTargetError> { - if !target.secure { - return Ok(None); - } - - if target.skip_tls_verify { - return Ok(Some(build_insecure_aws_s3_http_client())); - } - - if has_custom_ca_pem(target) { - return build_aws_s3_http_client_from_target_ca_pem(&target.ca_cert_pem) - .await - .map(Some); - } - - Ok(build_aws_s3_http_client_from_tls_path().await) -} - -async fn build_aws_s3_http_client_from_tls_path() -> Option { - let certificate_bundles = load_configured_tls_ca_bundles().await; - if certificate_bundles.is_empty() { - return None; - } - - let (trust_store, _) = compose_replication_trust_store(certificate_bundles); - match build_aws_s3_http_client_with_trust_store(trust_store) { - Ok(client) => Some(client), - Err(e) => { - warn!("failed to build AWS SDK TLS context for replication client: {}", e); - None - } - } -} - -fn should_force_path_style(target: &BucketTarget) -> bool { - match target.path.trim().to_ascii_lowercase().as_str() { - // Explicit DNS/virtual-hosted-style requested by user. - "dns" | "off" | "false" => false, - // Explicit path-style or legacy boolean-like values. - "path" | "on" | "true" => true, - // `auto` and empty are defaulted to path-style for custom S3-compatible endpoints. - "auto" | "" => true, - // Unknown values: prefer compatibility with S3-compatible services. - _ => true, - } -} - // generate ARN that is unique to this target type fn generate_arn(t: &BucketTarget, depl_id: &str) -> String { let uuid = if depl_id.is_empty() { @@ -2707,7 +2306,24 @@ impl Error for BucketTargetError {} #[cfg(test)] mod tests { use super::*; + use crate::bucket::remote_s3_client::{ + EXPIRED_REMOTE_TARGET_CREDENTIALS, RemoteTargetCredentialsProvider, build_aws_s3_http_client_for_spec, + build_aws_s3_http_client_from_target_ca_pem, build_aws_s3_http_client_with_trust_store, + build_insecure_aws_s3_http_client, compose_replication_trust_store, ensure_rustls_crypto_provider, + load_tls_path_ca_bundles, remote_sdk_credentials, replication_request_checksum_calculation, + validate_remote_endpoint_inner, validate_target_ca_pem, + }; + use aws_credential_types::Credentials as SdkCredentials; + use aws_sdk_s3::Config as S3Config; + use aws_sdk_s3::config::{Region as SdkRegion, RequestChecksumCalculation, SharedCredentialsProvider, SharedHttpClient}; + use aws_smithy_runtime_api::client::http::{ + HttpConnector as SmithyHttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn, + }; + use aws_smithy_runtime_api::client::orchestrator::HttpResponse; + use aws_smithy_types::body::SdkBody; use rcgen::generate_simple_self_signed; + use rustfs_config::{RUSTFS_CA_CERT, RUSTFS_TLS_CERT}; + use rustfs_utils::egress::OutboundUrlError; // The startup panic fix for hosts without a CA bundle (issue #6734) rests // on two properties: the health-check client constructor never panics, and @@ -2934,8 +2550,8 @@ mod tests { expiration: Some(jiff::Timestamp::try_from(expiration).expect("test expiration should convert")), }; - let sdk_credentials = - remote_target_sdk_credentials(&credentials, "account", now).expect("unexpired temporary credentials should build"); + let sdk_credentials = remote_sdk_credentials(&remote_credentials(&credentials, "account"), now) + .expect("unexpired temporary credentials should build"); assert_eq!(sdk_credentials.session_token(), Some("temporary-session-token")); assert_eq!(sdk_credentials.expiry(), Some(expiration)); @@ -2951,7 +2567,7 @@ mod tests { expiration: Some("0001-01-01T00:00:00Z".parse().expect("Go zero time should parse")), }; - let sdk_credentials = remote_target_sdk_credentials(&credentials, "", SystemTime::now()) + let sdk_credentials = remote_sdk_credentials(&remote_credentials(&credentials, ""), SystemTime::now()) .expect("Go zero expiration should remain compatible with static credentials"); assert!(sdk_credentials.session_token().is_none()); @@ -2969,14 +2585,14 @@ mod tests { }; assert_eq!( - remote_target_sdk_credentials(&credentials, "", SystemTime::UNIX_EPOCH + Duration::from_secs(1_000)) + remote_sdk_credentials(&remote_credentials(&credentials, ""), SystemTime::UNIX_EPOCH + Duration::from_secs(1_000)) .expect_err("expiration without a session token must fail"), "remote target credential expiration requires a session token" ); credentials.session_token = Some("temporary-session-token".to_string()); assert_eq!( - remote_target_sdk_credentials(&credentials, "", expiration) + remote_sdk_credentials(&remote_credentials(&credentials, ""), expiration) .expect_err("credentials expire at the exact expiration boundary"), EXPIRED_REMOTE_TARGET_CREDENTIALS ); @@ -3036,7 +2652,7 @@ mod tests { session_token: Some("temporary-session-token".to_string()), expiration: Some("2099-01-01T00:00:00Z".parse().expect("future expiration should parse")), }; - let sdk_credentials = remote_target_sdk_credentials(&credentials, "", SystemTime::now()) + let sdk_credentials = remote_sdk_credentials(&remote_credentials(&credentials, ""), SystemTime::now()) .expect("unexpired temporary credentials should build"); let client = S3Client::from_conf( S3Config::builder() @@ -3211,6 +2827,46 @@ mod tests { assert!(!replication_target_versioning_enabled(None)); } + #[test] + fn remote_endpoint_spec_from_target_keeps_legacy_path_style_and_trust_semantics() { + for (path, expected) in [ + ("dns", PathStyle::VirtualHost), + ("OFF", PathStyle::VirtualHost), + ("false", PathStyle::VirtualHost), + ("path", PathStyle::Path), + ("on", PathStyle::Path), + ("true", PathStyle::Path), + (" auto ", PathStyle::Auto), + ("", PathStyle::Auto), + ("something-else", PathStyle::Path), + ] { + assert_eq!(target_path_style(path), expected, "path={path:?}"); + } + + let spec = RemoteS3EndpointSpec::from(&BucketTarget { + endpoint: "192.168.1.10:9000".to_string(), + secure: true, + region: "us-east-1".to_string(), + ca_cert_pem: " ".to_string(), + reset_id: "reset-1".to_string(), + credentials: Some(Credentials { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + session_token: Some(" ".to_string()), + expiration: Some("0001-01-01T00:00:00Z".parse().expect("Go zero time should parse")), + }), + ..Default::default() + }); + assert_eq!(spec.endpoint_url(), "https://192.168.1.10:9000"); + assert!(spec.ca_cert_pem.is_none(), "whitespace-only CA PEM means unset"); + assert!(spec.connect_timeout.is_none() && spec.read_timeout.is_none()); + assert_eq!(spec.user_agent_suffix, ""); + let credentials = spec.credentials.expect("credentials carry over"); + assert_eq!(credentials.account_id, "reset-1"); + assert!(credentials.session_token.is_none(), "blank session token is absent"); + assert!(credentials.expiration.is_none(), "Go zero expiration is absent"); + } + fn parse_url(raw: &str) -> Url { Url::parse(raw).expect("test URL should parse") } @@ -3220,16 +2876,16 @@ mod tests { // Public hosts and private-network targets are allowed regardless of the // loopback opt-in — replication commonly runs across trusted private infra. for allow_loopback in [false, true] { - assert!(validate_replication_target_endpoint_inner(&parse_url("https://s3.example.com"), allow_loopback).is_ok()); - assert!(validate_replication_target_endpoint_inner(&parse_url("http://10.0.0.5:9000"), allow_loopback).is_ok()); - assert!(validate_replication_target_endpoint_inner(&parse_url("http://192.168.1.20"), allow_loopback).is_ok()); + assert!(validate_remote_endpoint_inner(&parse_url("https://s3.example.com"), allow_loopback).is_ok()); + assert!(validate_remote_endpoint_inner(&parse_url("http://10.0.0.5:9000"), allow_loopback).is_ok()); + assert!(validate_remote_endpoint_inner(&parse_url("http://192.168.1.20"), allow_loopback).is_ok()); } } #[test] fn replication_endpoint_rejects_loopback_without_opt_in() { // Default (production) behaviour: loopback IP and localhost host both rejected. - let err = validate_replication_target_endpoint_inner(&parse_url("http://127.0.0.1:9000"), false) + let err = validate_remote_endpoint_inner(&parse_url("http://127.0.0.1:9000"), false) .expect_err("loopback IP must be rejected by default"); assert!(matches!( err, @@ -3238,7 +2894,7 @@ mod tests { .. } )); - let err = validate_replication_target_endpoint_inner(&parse_url("http://localhost:9000"), false) + let err = validate_remote_endpoint_inner(&parse_url("http://localhost:9000"), false) .expect_err("localhost must be rejected by default"); assert!(matches!( err, @@ -3253,15 +2909,15 @@ mod tests { fn replication_endpoint_allows_loopback_with_opt_in() { // e2e harness / single-host multi-instance: opt-in re-enables loopback in // both IP (127.0.0.1, ::1) and hostname (localhost) forms. - assert!(validate_replication_target_endpoint_inner(&parse_url("http://127.0.0.1:9000"), true).is_ok()); - assert!(validate_replication_target_endpoint_inner(&parse_url("http://[::1]:9000"), true).is_ok()); - assert!(validate_replication_target_endpoint_inner(&parse_url("http://localhost:9000"), true).is_ok()); + assert!(validate_remote_endpoint_inner(&parse_url("http://127.0.0.1:9000"), true).is_ok()); + assert!(validate_remote_endpoint_inner(&parse_url("http://[::1]:9000"), true).is_ok()); + assert!(validate_remote_endpoint_inner(&parse_url("http://localhost:9000"), true).is_ok()); } #[test] fn replication_endpoint_opt_in_does_not_open_other_ssrf_targets() { // The loopback opt-in must not widen into link-local / metadata endpoints. - let err = validate_replication_target_endpoint_inner(&parse_url("http://169.254.169.254/latest/meta-data"), true) + let err = validate_remote_endpoint_inner(&parse_url("http://169.254.169.254/latest/meta-data"), true) .expect_err("metadata endpoint must stay rejected even with loopback opt-in"); assert!(matches!( err, @@ -3270,7 +2926,7 @@ mod tests { .. } )); - let err = validate_replication_target_endpoint_inner(&parse_url("http://[fe80::1]:9000"), true) + let err = validate_remote_endpoint_inner(&parse_url("http://[fe80::1]:9000"), true) .expect_err("link-local must stay rejected even with loopback opt-in"); assert!(matches!( err, @@ -4279,12 +3935,12 @@ mod tests { #[tokio::test] async fn skip_tls_verify_takes_priority_over_invalid_custom_ca_pem() { - let client = build_aws_s3_http_client_for_target(&BucketTarget { + let client = build_aws_s3_http_client_for_spec(&RemoteS3EndpointSpec::from(&BucketTarget { secure: true, skip_tls_verify: true, ca_cert_pem: "not a pem".to_string(), ..Default::default() - }) + })) .await .expect("skip verification should bypass custom CA parsing"); diff --git a/crates/ecstore/src/bucket/mod.rs b/crates/ecstore/src/bucket/mod.rs index 54306247a..925af5309 100644 --- a/crates/ecstore/src/bucket/mod.rs +++ b/crates/ecstore/src/bucket/mod.rs @@ -28,6 +28,7 @@ mod msgp_decode; pub mod object_lock; pub mod policy_sys; pub mod quota; +pub mod remote_s3_client; pub mod replication; pub mod tagging; pub mod target; diff --git a/crates/ecstore/src/bucket/remote_s3_client.rs b/crates/ecstore/src/bucket/remote_s3_client.rs new file mode 100644 index 000000000..58a311a04 --- /dev/null +++ b/crates/ecstore/src/bucket/remote_s3_client.rs @@ -0,0 +1,789 @@ +// 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. + +//! Shared builder for outbound `aws_sdk_s3::Client`s. +//! +//! Replication targets (`bucket_target_sys`) and the on-demand migration +//! source client build their remote clients from one neutral +//! [`RemoteS3EndpointSpec`]: endpoint assembly, credential handling, path-style +//! selection, custom CA / skip-TLS transports and the outbound SSRF gate all +//! live here so both callers share exactly one policy. The gate keeps the +//! relaxed replication semantics documented in +//! `docs/operations/outbound-connection-policy.md`: private addresses are +//! always allowed, loopback only behind `RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET`. + +use aws_credential_types::Credentials as SdkCredentials; +use aws_credential_types::provider::{ProvideCredentials, error::CredentialsError, future}; +use aws_sdk_s3::config::Region as SdkRegion; +use aws_sdk_s3::config::RequestChecksumCalculation; +use aws_sdk_s3::config::SharedCredentialsProvider; +use aws_sdk_s3::config::SharedHttpClient; +use aws_sdk_s3::{Client as S3Client, Config as S3Config}; +use aws_smithy_http_client::{Builder as SmithyHttpClientBuilder, tls as smithy_tls}; +use aws_smithy_runtime_api::box_error::BoxError; +use aws_smithy_runtime_api::client::http::{ + HttpConnector as SmithyHttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn, +}; +use aws_smithy_runtime_api::client::interceptors::Intercept; +use aws_smithy_runtime_api::client::interceptors::context::BeforeTransmitInterceptorContextMut; +use aws_smithy_runtime_api::client::orchestrator::{HttpRequest, HttpResponse}; +use aws_smithy_runtime_api::client::result::ConnectorError; +use aws_smithy_runtime_api::client::runtime_components::RuntimeComponents; +use aws_smithy_types::body::SdkBody; +use aws_smithy_types::config_bag::ConfigBag; +use aws_smithy_types::timeout::TimeoutConfig; +use http::Uri; +use hyper_util::client::legacy::Client as HyperClient; +use hyper_util::rt::{TokioExecutor, TokioTimer}; +use rustfs_config::{DEFAULT_TRUST_LEAF_CERT_AS_CA, ENV_TRUST_LEAF_CERT_AS_CA, RUSTFS_CA_CERT, RUSTFS_TLS_CERT}; +use rustfs_utils::egress::{OutboundUrlError, validate_outbound_url}; +use rustls_pki_types::pem::PemObject; +use std::fmt; +use std::path::Path; +use std::sync::Arc; +use std::time::{Duration, SystemTime}; +use tower::Service; +use tracing::warn; +use url::Url; + +const REDACTED_CREDENTIAL: &str = ""; +pub(crate) const EXPIRED_REMOTE_TARGET_CREDENTIALS: &str = "remote target credentials have expired"; + +/// Request addressing style for a remote S3-compatible endpoint. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum PathStyle { + /// Caller did not choose; the builder defaults to path-style because that + /// is what custom S3-compatible endpoints accept most reliably. + Auto, + /// `https://endpoint/bucket/key`. + Path, + /// `https://bucket.endpoint/key`. + VirtualHost, +} + +impl PathStyle { + /// Resolves the style to the SDK `force_path_style` flag. `Auto` keeps + /// the historical replication default (path-style). + pub fn force_path_style(self) -> bool { + !matches!(self, PathStyle::VirtualHost) + } +} + +/// Static or temporary credentials for a remote endpoint. `expiration` without +/// a `session_token` is rejected at build time: only STS-style temporary +/// credentials expire, so that combination is a corrupted configuration +/// rather than a static key. +#[derive(Clone)] +pub struct RemoteCredentials { + pub access_key: String, + pub secret_key: String, + pub session_token: Option, + pub expiration: Option, + /// SDK credential `account_id`; replication targets pass their reset id. + pub account_id: String, +} + +impl fmt::Debug for RemoteCredentials { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("RemoteCredentials") + .field("access_key", &self.access_key) + .field("secret_key", &REDACTED_CREDENTIAL) + .field("session_token", &self.session_token.as_ref().map(|_| REDACTED_CREDENTIAL)) + .field("expiration", &self.expiration) + .field("account_id", &self.account_id) + .finish() + } +} + +/// Neutral description of a remote S3 endpoint from which an +/// `aws_sdk_s3::Client` is built. +#[derive(Clone, Debug)] +pub struct RemoteS3EndpointSpec { + /// `host[:port]` without a scheme; `secure` selects `https` or `http`. + pub endpoint: String, + pub secure: bool, + pub region: String, + pub path_style: PathStyle, + pub credentials: Option, + /// Accept any server certificate. Takes priority over `ca_cert_pem`. + pub skip_tls_verify: bool, + /// Extra PEM bundle trusted alongside the platform roots and the + /// `RUSTFS_TLS_PATH` bundle. `None` and whitespace-only mean "not set". + pub ca_cert_pem: Option, + pub connect_timeout: Option, + pub read_timeout: Option, + /// Appended to the SDK `User-Agent` (space separated) so the remote side + /// can identify the caller; empty means no suffix. + pub user_agent_suffix: &'static str, +} + +impl RemoteS3EndpointSpec { + /// Full endpoint URL (`scheme://host[:port]`) as handed to the SDK. + pub fn endpoint_url(&self) -> String { + if self.secure { + format!("https://{}", self.endpoint) + } else { + format!("http://{}", self.endpoint) + } + } + + fn custom_ca_pem(&self) -> Option<&str> { + self.ca_cert_pem.as_deref().filter(|pem| !pem.trim().is_empty()) + } +} + +#[derive(Debug, thiserror::Error)] +pub enum RemoteS3ClientError { + #[error("remote endpoint requires credentials")] + MissingCredentials, + #[error("{0}")] + Credentials(&'static str), + #[error("invalid target endpoint: {0}")] + InvalidEndpoint(String), + #[error("target endpoint is not allowed: {0}")] + EndpointNotAllowed(#[source] OutboundUrlError), + #[error("invalid target CA PEM: {0}")] + InvalidCaPem(String), +} + +#[derive(Clone)] +pub(crate) struct RemoteTargetCredentialsProvider { + pub(crate) credentials: SdkCredentials, +} + +impl RemoteTargetCredentialsProvider { + pub(crate) fn resolve_at(&self, now: SystemTime) -> aws_credential_types::provider::Result { + if self.credentials.expiry().is_some_and(|expiration| expiration <= now) { + return Err(CredentialsError::provider_error(std::io::Error::other(EXPIRED_REMOTE_TARGET_CREDENTIALS))); + } + Ok(self.credentials.clone()) + } +} + +impl fmt::Debug for RemoteTargetCredentialsProvider { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("RemoteTargetCredentialsProvider") + .field("temporary", &self.credentials.session_token().is_some()) + .field("expiration", &self.credentials.expiry()) + .finish() + } +} + +impl ProvideCredentials for RemoteTargetCredentialsProvider { + fn provide_credentials<'a>(&'a self) -> future::ProvideCredentials<'a> + where + Self: 'a, + { + future::ProvideCredentials::ready(self.resolve_at(SystemTime::now())) + } + + fn fallback_on_interrupt(&self) -> Option { + self.resolve_at(SystemTime::now()).ok() + } +} + +pub(crate) fn remote_sdk_credentials(credentials: &RemoteCredentials, now: SystemTime) -> Result { + if credentials.expiration.is_some() && credentials.session_token.is_none() { + return Err("remote target credential expiration requires a session token"); + } + if credentials.expiration.is_some_and(|expiration| expiration <= now) { + return Err(EXPIRED_REMOTE_TARGET_CREDENTIALS); + } + + let mut builder = SdkCredentials::builder() + .access_key_id(credentials.access_key.clone()) + .secret_access_key(credentials.secret_key.clone()) + .account_id(credentials.account_id.clone()) + .provider_name("bucket_target_sys"); + if let Some(session_token) = &credentials.session_token { + builder = builder.session_token(session_token.clone()); + } + if let Some(expiration) = credentials.expiration { + builder = builder.expiry(expiration); + } + Ok(builder.build()) +} + +/// Appends a caller-identifying token to the SDK `User-Agent`. Runs after +/// signing: SigV4 excludes `user-agent` from the canonical request, so the +/// signature stays valid. +#[derive(Debug)] +struct UserAgentSuffixInterceptor { + suffix: &'static str, +} + +impl Intercept for UserAgentSuffixInterceptor { + fn name(&self) -> &'static str { + "RustfsUserAgentSuffix" + } + + fn modify_before_transmit( + &self, + context: &mut BeforeTransmitInterceptorContextMut<'_>, + _runtime_components: &RuntimeComponents, + _cfg: &mut ConfigBag, + ) -> Result<(), BoxError> { + let headers = context.request_mut().headers_mut(); + let user_agent = match headers.get(http::header::USER_AGENT.as_str()) { + Some(existing) => format!("{existing} {}", self.suffix), + None => self.suffix.to_string(), + }; + headers.try_insert(http::header::USER_AGENT.as_str(), user_agent)?; + Ok(()) + } +} + +/// Builds the SDK config for `spec` without finalizing it, so callers can add +/// interceptors or (in tests) swap the HTTP client before `build()`. +pub(crate) async fn build_remote_s3_config( + spec: &RemoteS3EndpointSpec, +) -> Result { + let Some(credentials) = &spec.credentials else { + return Err(RemoteS3ClientError::MissingCredentials); + }; + let creds = remote_sdk_credentials(credentials, SystemTime::now()).map_err(RemoteS3ClientError::Credentials)?; + + let endpoint = spec.endpoint_url(); + let parsed_endpoint = Url::parse(&endpoint).map_err(|err| RemoteS3ClientError::InvalidEndpoint(err.to_string()))?; + validate_remote_endpoint(&parsed_endpoint).map_err(RemoteS3ClientError::EndpointNotAllowed)?; + + let mut config_builder = S3Config::builder() + .endpoint_url(endpoint) + .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider { credentials: creds })) + .region(SdkRegion::new(spec.region.clone())) + .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest()) + .request_checksum_calculation(replication_request_checksum_calculation()); + + if spec.path_style.force_path_style() { + config_builder = config_builder.force_path_style(true); + } + + if let Some(http_client) = build_aws_s3_http_client_for_spec(spec).await? { + config_builder = config_builder.http_client(http_client); + } + + if spec.connect_timeout.is_some() || spec.read_timeout.is_some() { + let mut timeouts = TimeoutConfig::builder(); + if let Some(connect_timeout) = spec.connect_timeout { + timeouts = timeouts.connect_timeout(connect_timeout); + } + if let Some(read_timeout) = spec.read_timeout { + timeouts = timeouts.read_timeout(read_timeout); + } + config_builder = config_builder.timeout_config(timeouts.build()); + } + + if !spec.user_agent_suffix.is_empty() { + config_builder = config_builder.interceptor(UserAgentSuffixInterceptor { + suffix: spec.user_agent_suffix, + }); + } + + Ok(config_builder) +} + +/// Builds an `aws_sdk_s3::Client` for `spec`, applying the outbound endpoint +/// gate, credential validation and the TLS transport selection. +pub async fn build_remote_s3_client(spec: &RemoteS3EndpointSpec) -> Result { + Ok(S3Client::from_conf(build_remote_s3_config(spec).await?.build())) +} + +#[derive(Debug)] +struct AcceptAnyServerCertVerifier; + +impl rustls::client::danger::ServerCertVerifier for AcceptAnyServerCertVerifier { + fn verify_server_cert( + &self, + _end_entity: &rustls_pki_types::CertificateDer<'_>, + _intermediates: &[rustls_pki_types::CertificateDer<'_>], + _server_name: &rustls_pki_types::ServerName<'_>, + _ocsp_response: &[u8], + _now: rustls_pki_types::UnixTime, + ) -> Result { + Ok(rustls::client::danger::ServerCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + _message: &[u8], + _cert: &rustls_pki_types::CertificateDer<'_>, + _dss: &rustls::DigitallySignedStruct, + ) -> Result { + Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) + } + + fn verify_tls13_signature( + &self, + _message: &[u8], + _cert: &rustls_pki_types::CertificateDer<'_>, + _dss: &rustls::DigitallySignedStruct, + ) -> Result { + Ok(rustls::client::danger::HandshakeSignatureValid::assertion()) + } + + fn supported_verify_schemes(&self) -> Vec { + rustls::crypto::aws_lc_rs::default_provider() + .signature_verification_algorithms + .supported_schemes() + } +} + +#[derive(Clone)] +struct TargetHyperHttpConnector { + client: HyperClient, +} + +impl fmt::Debug for TargetHyperHttpConnector { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TargetHyperHttpConnector") + .field("client", &"** hyper client **") + .finish() + } +} + +impl SmithyHttpConnector for TargetHyperHttpConnector +where + C: Clone + Send + Sync + 'static, + C: Service, + C::Response: + hyper::rt::Read + hyper::rt::Write + hyper_util::client::legacy::connect::Connection + Send + Sync + Unpin + 'static, + C::Future: Unpin + Send + 'static, + C::Error: Into, +{ + fn call(&self, request: HttpRequest) -> HttpConnectorFuture { + let request = match request.try_into_http1x() { + Ok(request) => request, + Err(err) => return HttpConnectorFuture::ready(Err(ConnectorError::user(err.into()))), + }; + + let mut client = self.client.clone(); + let fut = client.call(request); + HttpConnectorFuture::new(async move { + let response = fut + .await + .map_err(|err| ConnectorError::io(err.into()))? + .map(SdkBody::from_body_1_x); + HttpResponse::try_from(response).map_err(|err| ConnectorError::other(err.into(), None)) + }) + } +} + +pub(crate) fn ensure_rustls_crypto_provider() { + if rustls::crypto::CryptoProvider::get_default().is_none() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + } +} + +/// Env opt-in that re-enables loopback replication targets. Loopback (`127.0.0.1`, +/// `::1`, `localhost`) is a classic SSRF vector and stays rejected by default, but +/// single-host multi-instance dev setups and the e2e harness legitimately replicate +/// over loopback. Never set this in production. +const ALLOW_LOOPBACK_REPLICATION_TARGET_ENV: &str = "RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET"; + +fn loopback_replication_targets_allowed() -> bool { + std::env::var(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV) + .map(|v| v.eq_ignore_ascii_case("true") || v == "1") + .unwrap_or(false) +} + +const REPLICATION_STREAMING_CHECKSUMS_ENV: &str = "RUSTFS_REPLICATION_STREAMING_CHECKSUMS"; + +/// Streaming trailer checksums make the SDK frame request bodies as +/// `aws-chunked`; a target that does not decode that framing stores the frames +/// verbatim, silently corrupting every replica while the transfer itself +/// succeeds (#6853). Plain signed payloads are the compatible default; the env +/// knob restores trailer checksums for fleets whose targets are all known to +/// decode them. +pub(crate) fn replication_request_checksum_calculation() -> RequestChecksumCalculation { + if std::env::var(REPLICATION_STREAMING_CHECKSUMS_ENV) + .map(|v| v.eq_ignore_ascii_case("true") || v == "1") + .unwrap_or(false) + { + RequestChecksumCalculation::WhenSupported + } else { + RequestChecksumCalculation::WhenRequired + } +} + +/// Outbound gate for operator-configured remote endpoints (replication +/// targets, on-demand migration sources). See +/// `docs/operations/outbound-connection-policy.md`. +pub fn validate_remote_endpoint(url: &Url) -> Result<(), OutboundUrlError> { + validate_remote_endpoint_inner(url, loopback_replication_targets_allowed()) +} + +pub(crate) fn validate_remote_endpoint_inner(url: &Url, allow_loopback: bool) -> Result<(), OutboundUrlError> { + match validate_outbound_url(url) { + Ok(()) => Ok(()), + // Replication targets are trusted infrastructure the operator configures, and + // legitimately live on private networks, so private addresses are always allowed. + Err(OutboundUrlError::ForbiddenHost { + reason: "private address", + .. + }) => Ok(()), + // Loopback is far higher SSRF risk, so it is allowed only under the explicit, + // off-by-default opt-in above (single-host multi-instance / the e2e harness). + Err(OutboundUrlError::ForbiddenHost { + reason: "loopback address" | "loopback host", + .. + }) if allow_loopback => Ok(()), + Err(err) => Err(err), + } +} + +pub(crate) fn build_insecure_aws_s3_http_client() -> SharedHttpClient { + ensure_rustls_crypto_provider(); + + let tls_config = rustls::ClientConfig::builder() + .dangerous() + .with_custom_certificate_verifier(Arc::new(AcceptAnyServerCertVerifier)) + .with_no_client_auth(); + + let https = hyper_rustls::HttpsConnectorBuilder::new() + .with_tls_config(tls_config) + .https_or_http() + .enable_http1() + .enable_http2() + .build(); + let mut client_builder = HyperClient::builder(TokioExecutor::new()); + client_builder.pool_timer(TokioTimer::new()); + let client = client_builder.build(https); + let connector = SharedHttpConnector::new(TargetHyperHttpConnector { client }); + + http_client_fn(move |_settings, _components| connector.clone()) +} + +fn validate_ca_pem_bundle(ca_cert_pem: &[u8]) -> Result<(), String> { + let certs = rustls_pki_types::CertificateDer::pem_slice_iter(ca_cert_pem) + .collect::, _>>() + .map_err(|err| format!("invalid PEM encoding: {err}"))?; + + if certs.is_empty() { + return Err("no certificates found".to_string()); + } + + // Smithy's rustls adapter defers parsing custom certificates and assumes + // they are valid when the HTTPS connector is built. Validate every DER + // certificate first so malformed configuration is reported rather than + // reaching an `expect` in the dependency. + let mut validation_store = rustls::RootCertStore::empty(); + for cert in certs { + validation_store + .add(cert) + .map_err(|err| format!("invalid X.509 certificate: {err}"))?; + } + + Ok(()) +} + +pub(crate) fn validate_target_ca_pem(ca_cert_pem: &str) -> Result<(), RemoteS3ClientError> { + validate_ca_pem_bundle(ca_cert_pem.as_bytes()).map_err(RemoteS3ClientError::InvalidCaPem) +} + +pub(crate) fn compose_replication_trust_store( + certificate_bundles: impl IntoIterator>, +) -> (smithy_tls::TrustStore, usize) { + // `TrustStore::default()` keeps the platform-native roots enabled. Target + // and RUSTFS_TLS_PATH certificates extend that baseline instead of + // replacing it with a target-specific trust island. + let mut trust_store = smithy_tls::TrustStore::default(); + let mut custom_bundle_count = 0; + for pem in certificate_bundles { + trust_store.add_pem_certificate(pem); + custom_bundle_count += 1; + } + + (trust_store, custom_bundle_count) +} + +pub(crate) fn build_aws_s3_http_client_with_trust_store( + trust_store: smithy_tls::TrustStore, +) -> Result { + let tls_context = smithy_tls::TlsContext::builder() + .with_trust_store(trust_store) + .build() + .map_err(|err| RemoteS3ClientError::InvalidCaPem(err.to_string()))?; + + Ok(SmithyHttpClientBuilder::new() + .tls_provider(smithy_tls::Provider::rustls(smithy_tls::rustls_provider::CryptoMode::AwsLc)) + .tls_context(tls_context) + .build_https()) +} + +pub(crate) async fn load_tls_path_ca_bundles(tls_dir: &Path, trust_leaf_cert_as_ca: bool) -> Vec> { + let mut certificate_bundles = Vec::new(); + + let ca_path = tls_dir.join(RUSTFS_CA_CERT); + match tokio::fs::read(&ca_path).await { + Ok(pem) => match validate_ca_pem_bundle(&pem) { + Ok(()) => certificate_bundles.push(pem), + Err(err) => warn!("ignoring invalid custom CA bundle {:?} for replication client: {}", ca_path, err), + }, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => warn!("failed to read custom CA bundle {:?} for replication client: {}", ca_path, e), + } + + if trust_leaf_cert_as_ca { + let leaf_cert_path = tls_dir.join(RUSTFS_TLS_CERT); + match tokio::fs::read(&leaf_cert_path).await { + Ok(pem) => match validate_ca_pem_bundle(&pem) { + Ok(()) => certificate_bundles.push(pem), + Err(err) => warn!( + "ignoring invalid leaf certificate {:?} for replication client trust store: {}", + leaf_cert_path, err + ), + }, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => warn!("failed to read leaf cert {:?} for replication client trust store: {}", leaf_cert_path, e), + } + } + + certificate_bundles +} + +async fn load_configured_tls_ca_bundles() -> Vec> { + let tls_path = rustfs_utils::get_env_str(rustfs_config::ENV_RUSTFS_TLS_PATH, rustfs_config::DEFAULT_RUSTFS_TLS_PATH); + if tls_path.is_empty() { + return Vec::new(); + } + + load_tls_path_ca_bundles( + Path::new(&tls_path), + rustfs_utils::get_env_bool(ENV_TRUST_LEAF_CERT_AS_CA, DEFAULT_TRUST_LEAF_CERT_AS_CA), + ) + .await +} + +pub(crate) async fn build_aws_s3_http_client_from_target_ca_pem( + ca_cert_pem: &str, +) -> Result { + validate_target_ca_pem(ca_cert_pem)?; + + let mut certificate_bundles = load_configured_tls_ca_bundles().await; + certificate_bundles.push(ca_cert_pem.as_bytes().to_vec()); + let (trust_store, _) = compose_replication_trust_store(certificate_bundles); + + build_aws_s3_http_client_with_trust_store(trust_store) +} + +/// Selects the HTTP client for `spec`: `None` keeps the SDK default (plain +/// HTTP, or HTTPS with platform roots when no custom trust is configured). +pub(crate) async fn build_aws_s3_http_client_for_spec( + spec: &RemoteS3EndpointSpec, +) -> Result, RemoteS3ClientError> { + if !spec.secure { + return Ok(None); + } + + if spec.skip_tls_verify { + return Ok(Some(build_insecure_aws_s3_http_client())); + } + + if let Some(ca_cert_pem) = spec.custom_ca_pem() { + return build_aws_s3_http_client_from_target_ca_pem(ca_cert_pem).await.map(Some); + } + + Ok(build_aws_s3_http_client_from_tls_path().await) +} + +async fn build_aws_s3_http_client_from_tls_path() -> Option { + let certificate_bundles = load_configured_tls_ca_bundles().await; + if certificate_bundles.is_empty() { + return None; + } + + let (trust_store, _) = compose_replication_trust_store(certificate_bundles); + match build_aws_s3_http_client_with_trust_store(trust_store) { + Ok(client) => Some(client), + Err(e) => { + warn!("failed to build AWS SDK TLS context for replication client: {}", e); + None + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aws_smithy_runtime_api::http::StatusCode as SmithyStatusCode; + use std::sync::Mutex; + + fn spec(endpoint: &str, secure: bool) -> RemoteS3EndpointSpec { + RemoteS3EndpointSpec { + endpoint: endpoint.to_string(), + secure, + region: "us-east-1".to_string(), + path_style: PathStyle::Auto, + credentials: Some(RemoteCredentials { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + session_token: None, + expiration: None, + account_id: String::new(), + }), + skip_tls_verify: false, + ca_cert_pem: None, + connect_timeout: None, + read_timeout: None, + user_agent_suffix: "", + } + } + + type RecordedHeaders = Arc>>>; + + #[derive(Clone, Debug)] + struct RecordingHeaderConnector { + request_headers: RecordedHeaders, + } + + impl SmithyHttpConnector for RecordingHeaderConnector { + fn call(&self, request: HttpRequest) -> HttpConnectorFuture { + self.request_headers + .lock() + .expect("recorded header lock should not be poisoned") + .push( + request + .headers() + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + ); + HttpConnectorFuture::ready(Ok(HttpResponse::new( + SmithyStatusCode::try_from(200_u16).expect("200 should be a valid response status"), + SdkBody::empty(), + ))) + } + } + + #[tokio::test] + async fn build_remote_s3_client_rejects_loopback_and_metadata_endpoints() { + // Default (no loopback opt-in): loopback in IPv4, IPv6 and hostname + // forms plus the metadata endpoint all return the typed gate error. + for endpoint in ["127.0.0.1:9000", "[::1]:9000", "localhost:9000", "169.254.169.254"] { + let err = build_remote_s3_client(&spec(endpoint, false)) + .await + .err() + .unwrap_or_else(|| panic!("{endpoint} must be rejected by the outbound gate")); + assert!( + matches!(err, RemoteS3ClientError::EndpointNotAllowed(OutboundUrlError::ForbiddenHost { .. })), + "{endpoint}: unexpected error {err:?}" + ); + assert!(err.to_string().contains("not allowed"), "{endpoint}: {err}"); + } + } + + #[tokio::test] + async fn build_remote_s3_client_allows_private_and_public_endpoints() { + for endpoint in ["10.0.0.1:9000", "192.168.1.20", "s3.example.com"] { + build_remote_s3_client(&spec(endpoint, false)) + .await + .unwrap_or_else(|err| panic!("{endpoint} should be allowed: {err}")); + } + } + + #[tokio::test] + async fn build_remote_s3_client_requires_credentials() { + let mut spec = spec("s3.example.com", true); + spec.credentials = None; + let err = build_remote_s3_client(&spec) + .await + .expect_err("missing credentials must be a typed error"); + assert!(matches!(err, RemoteS3ClientError::MissingCredentials)); + } + + #[tokio::test] + async fn build_remote_s3_client_rejects_expiration_without_session_token() { + let mut spec = spec("s3.example.com", true); + spec.credentials + .as_mut() + .expect("spec fixture carries credentials") + .expiration = Some(SystemTime::now() + Duration::from_secs(3_600)); + let err = build_remote_s3_client(&spec) + .await + .expect_err("expiration without session token must be rejected"); + assert_eq!(err.to_string(), "remote target credential expiration requires a session token"); + } + + #[tokio::test] + async fn build_remote_s3_client_rejects_invalid_custom_ca_pem() { + let mut spec = spec("192.168.1.10:9000", true); + spec.ca_cert_pem = Some("not a pem".to_string()); + let err = build_remote_s3_client(&spec) + .await + .expect_err("invalid custom CA PEM must be rejected"); + assert!(matches!(err, RemoteS3ClientError::InvalidCaPem(_))); + assert!(err.to_string().contains("invalid target CA PEM")); + } + + #[test] + fn path_style_auto_and_path_force_path_style() { + assert!(PathStyle::Auto.force_path_style()); + assert!(PathStyle::Path.force_path_style()); + assert!(!PathStyle::VirtualHost.force_path_style()); + } + + #[test] + fn remote_credentials_debug_redacts_secrets() { + let credentials = RemoteCredentials { + access_key: "access".to_string(), + secret_key: "very-secret".to_string(), + session_token: Some("session-token".to_string()), + expiration: None, + account_id: String::new(), + }; + let rendered = format!("{credentials:?}"); + assert!(rendered.contains("access")); + assert!(!rendered.contains("very-secret")); + assert!(!rendered.contains("session-token")); + } + + #[tokio::test] + async fn user_agent_suffix_is_appended_after_signing() { + let request_headers: RecordedHeaders = Arc::new(Mutex::new(Vec::new())); + let connector = SharedHttpConnector::new(RecordingHeaderConnector { + request_headers: Arc::clone(&request_headers), + }); + let http_client = http_client_fn(move |_settings, _components| connector.clone()); + + let mut spec = spec("s3.example.com", true); + spec.user_agent_suffix = "RustFS-Test/0.0"; + spec.connect_timeout = Some(Duration::from_secs(5)); + spec.read_timeout = Some(Duration::from_secs(5)); + let config = build_remote_s3_config(&spec) + .await + .expect("spec should build") + .http_client(http_client) + .build(); + S3Client::from_conf(config) + .head_bucket() + .bucket("bucket") + .send() + .await + .expect("recording connector should accept the request"); + + let recorded = request_headers.lock().expect("recorded header lock should not be poisoned"); + let headers = &recorded[0]; + let user_agent = headers + .iter() + .find(|(k, _)| k.eq_ignore_ascii_case("user-agent")) + .map(|(_, v)| v.as_str()) + .expect("SDK request must carry a user-agent"); + assert!(user_agent.ends_with(" RustFS-Test/0.0"), "user-agent was {user_agent}"); + assert!(user_agent.starts_with("aws-sdk-rust/"), "SDK identity must be preserved: {user_agent}"); + assert!( + headers.iter().any(|(k, _)| k.eq_ignore_ascii_case("authorization")), + "request must still be signed" + ); + } +} diff --git a/scripts/error-other-format-baseline.txt b/scripts/error-other-format-baseline.txt index e41af7bfd..4925fe18c 100644 --- a/scripts/error-other-format-baseline.txt +++ b/scripts/error-other-format-baseline.txt @@ -5,7 +5,6 @@ # scripts/check_error_other_format_ratchet.sh --update-baseline. A PR that # raises a count or adds a file is introducing a new quorum-bucketing hazard # and must carry an explicit exemption rationale in its description. -2|crates/ecstore/src/bucket/bucket_target_sys.rs 4|crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs 3|crates/ecstore/src/bucket/lifecycle/durable_namespace.rs 2|crates/ecstore/src/bucket/lifecycle/metadata_boundary.rs