From 7e1f261e38b2822495b703cbfa59358bc8d8ecf2 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 2 Sep 2026 21:59:11 +0800 Subject: [PATCH] refactor(ecstore): shared remote S3 client builder and ODM source client (#7067) * 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 * feat(ecstore): add on-demand migration SourceClient Add bucket/on_demand_migration/source_client.rs on top of the shared remote S3 builder: HEAD, ranged streaming GET, ListObjectsV2 with source-prefix mapping, GetObjectTagging and an admin probe. Every request carries the x-rustfs-/x-minio-source-proxy-request anti-loop markers and a RustFS-OnDemandMigration/ User-Agent suffix; SSE-C source objects are rejected as unsupported. SourceError classifies SDK failures (not found, access denied, throttled, timeout, connect, server error) with retryability and a stable metrics label. Debug output redacts credentials. Refs rustfs/backlog#2149 * docs(operations): point outbound policy at shared remote S3 client builder --- crates/ecstore/src/api/mod.rs | 17 + .../ecstore/src/bucket/bucket_target_sys.rs | 594 ++------ crates/ecstore/src/bucket/mod.rs | 2 + .../src/bucket/on_demand_migration/mod.rs | 18 + .../on_demand_migration/source_client.rs | 1343 +++++++++++++++++ crates/ecstore/src/bucket/remote_s3_client.rs | 789 ++++++++++ docs/operations/outbound-connection-policy.md | 2 +- scripts/error-other-format-baseline.txt | 1 - 8 files changed, 2295 insertions(+), 471 deletions(-) create mode 100644 crates/ecstore/src/bucket/on_demand_migration/mod.rs create mode 100644 crates/ecstore/src/bucket/on_demand_migration/source_client.rs 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..a43b3bda5 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -166,6 +166,16 @@ pub mod bucket { pub use crate::bucket::migration::{LegacyBlobDecryptFn, try_migrate_bucket_metadata, try_migrate_iam_config}; } + pub mod on_demand_migration { + pub mod source_client { + pub use crate::bucket::on_demand_migration::source_client::{ + SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceObject, SourcePage, SourceProbe, + SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value, + resolve_path_style, + }; + } + } + pub mod object_lock { pub use crate::bucket::object_lock::{ObjectLockApi, ObjectLockStatusExt}; @@ -199,6 +209,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..48fce4e02 100644 --- a/crates/ecstore/src/bucket/mod.rs +++ b/crates/ecstore/src/bucket/mod.rs @@ -26,8 +26,10 @@ mod metadata_test; pub mod migration; mod msgp_decode; pub mod object_lock; +pub mod on_demand_migration; 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/on_demand_migration/mod.rs b/crates/ecstore/src/bucket/on_demand_migration/mod.rs new file mode 100644 index 000000000..167edacc1 --- /dev/null +++ b/crates/ecstore/src/bucket/on_demand_migration/mod.rs @@ -0,0 +1,18 @@ +// 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. + +//! On-demand migration (ODM): serve and back-fill objects from an external +//! S3-compatible source bucket. + +pub mod source_client; diff --git a/crates/ecstore/src/bucket/on_demand_migration/source_client.rs b/crates/ecstore/src/bucket/on_demand_migration/source_client.rs new file mode 100644 index 000000000..4e23f3973 --- /dev/null +++ b/crates/ecstore/src/bucket/on_demand_migration/source_client.rs @@ -0,0 +1,1343 @@ +// 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. + +//! Outbound client for an on-demand migration source bucket. +//! +//! `SourceClient` wraps an `aws_sdk_s3::Client` built through the shared +//! remote builder and exposes the read-only surface the migration path +//! needs (HEAD, ranged streaming GET, ListObjectsV2, GetObjectTagging, a +//! probe for admin validation). Every request carries the +//! `source-proxy-request` anti-loop marker in both the `x-rustfs-` and +//! `x-minio-` prefixes so a RustFS/MinIO source answers locally instead of +//! proxying the miss back, and the SDK `User-Agent` is suffixed with +//! `RustFS-OnDemandMigration/` for source-side log attribution. +//! Client-supplied `If-*`, `Authorization`, `Host` and SSE-C headers are never +//! forwarded: v1 rejects SSE-C source objects outright. + +use crate::bucket::remote_s3_client::{ + PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, build_remote_s3_config, +}; +use crate::storage_api_contracts::range::HTTPRangeSpec; +use aws_sdk_s3::Client as S3Client; +use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; +use aws_sdk_s3::operation::get_object::GetObjectOutput; +use aws_sdk_s3::operation::head_object::HeadObjectOutput; +use aws_sdk_s3::primitives::{ByteStream, DateTime as SdkDateTime}; +use aws_sdk_s3::types::{Object as SdkObject, ServerSideEncryption}; +use aws_smithy_runtime_api::box_error::BoxError; +use aws_smithy_runtime_api::client::interceptors::Intercept; +use aws_smithy_runtime_api::client::interceptors::context::BeforeTransmitInterceptorContextMut; +use aws_smithy_runtime_api::client::orchestrator::HttpResponse; +use aws_smithy_runtime_api::client::runtime_components::RuntimeComponents; +use aws_smithy_types::config_bag::ConfigBag; +use aws_smithy_types::error::display::DisplayErrorContext; +use http::HeaderMap; +use rustfs_utils::http::{SUFFIX_SOURCE_PROXY_REQUEST, insert_header}; +use std::collections::HashMap; +use std::fmt; +use std::num::NonZeroU64; +use std::time::{Duration, SystemTime}; +use url::Url; + +/// Appended to the SDK `User-Agent` on every source request. +pub const USER_AGENT_SUFFIX: &str = concat!("RustFS-OnDemandMigration/", env!("CARGO_PKG_VERSION")); + +/// Source provider family; drives the `PathStyle::Auto` decision. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum SourceProvider { + Aws, + Gcs, + R2, + Minio, + Rustfs, + /// Generic S3-compatible service. + #[default] + S3, +} + +impl SourceProvider { + pub fn from_label(label: &str) -> Option { + match label.trim().to_ascii_lowercase().as_str() { + "aws" => Some(Self::Aws), + "gcs" => Some(Self::Gcs), + "r2" => Some(Self::R2), + "minio" => Some(Self::Minio), + "rustfs" => Some(Self::Rustfs), + "s3" => Some(Self::S3), + _ => None, + } + } + + pub fn label(self) -> &'static str { + match self { + Self::Aws => "aws", + Self::Gcs => "gcs", + Self::R2 => "r2", + Self::Minio => "minio", + Self::Rustfs => "rustfs", + Self::S3 => "s3", + } + } + + fn prefers_virtual_host(self) -> bool { + matches!(self, Self::Aws | Self::Gcs | Self::R2) + } +} + +/// Resolves `PathStyle::Auto` for a source: IP-literal or `localhost` +/// endpoints cannot carry a bucket subdomain and always use path-style; +/// otherwise AWS/GCS/R2 use virtual-host addressing and MinIO/RustFS/generic +/// S3 use path-style. Explicit choices pass through unchanged. +pub fn resolve_path_style(path_style: PathStyle, provider: SourceProvider, endpoint_host: &str) -> PathStyle { + match path_style { + PathStyle::Auto => { + if host_is_ip_or_localhost(endpoint_host) || !provider.prefers_virtual_host() { + PathStyle::Path + } else { + PathStyle::VirtualHost + } + } + explicit => explicit, + } +} + +fn host_is_ip_or_localhost(host: &str) -> bool { + let bare = host.trim_start_matches('[').trim_end_matches(']'); + bare.eq_ignore_ascii_case("localhost") || bare.parse::().is_ok() +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct SourceTimeouts { + pub connect: Duration, + pub read: Duration, +} + +impl Default for SourceTimeouts { + fn default() -> Self { + Self { + connect: Duration::from_secs(10), + read: Duration::from_secs(60), + } + } +} + +/// Plain description of a source bucket; ODM-05 converts the persisted +/// bucket configuration into this shape. +#[derive(Clone, Debug)] +pub struct SourceClientSpec { + /// `scheme://host[:port]` with no path, query or userinfo. + pub endpoint: String, + pub region: String, + pub bucket: String, + /// Prepended to every local key when addressing the source; `None` or + /// empty means the local namespace maps 1:1 onto the source bucket. + pub source_prefix: Option, + pub provider: SourceProvider, + pub path_style: PathStyle, + pub credentials: Option, + pub skip_tls_verify: bool, + pub ca_cert_pem: Option, + pub timeouts: SourceTimeouts, + /// Bytes per second the pull pipeline may consume from this source; + /// `None` means unlimited. Enforced by the consumer, not by this client. + pub bandwidth_limit: Option, +} + +impl SourceClientSpec { + fn endpoint_spec(&self) -> Result { + let url = Url::parse(self.endpoint.trim()).map_err(|err| RemoteS3ClientError::InvalidEndpoint(err.to_string()))?; + let secure = match url.scheme() { + "https" => true, + "http" => false, + other => { + return Err(RemoteS3ClientError::InvalidEndpoint(format!( + "unsupported scheme {other}; expected http or https" + ))); + } + }; + let Some(host) = url.host_str() else { + return Err(RemoteS3ClientError::InvalidEndpoint("endpoint has no host".to_string())); + }; + if !url.username().is_empty() || url.password().is_some() { + return Err(RemoteS3ClientError::InvalidEndpoint("endpoint must not carry userinfo".to_string())); + } + if !matches!(url.path(), "" | "/") || url.query().is_some() || url.fragment().is_some() { + return Err(RemoteS3ClientError::InvalidEndpoint( + "endpoint must be an origin without path, query or fragment".to_string(), + )); + } + let endpoint = match url.port() { + Some(port) => format!("{host}:{port}"), + None => host.to_string(), + }; + + Ok(RemoteS3EndpointSpec { + endpoint, + secure, + region: self.region.clone(), + path_style: resolve_path_style(self.path_style, self.provider, host), + credentials: self.credentials.clone(), + skip_tls_verify: self.skip_tls_verify, + ca_cert_pem: self.ca_cert_pem.clone(), + connect_timeout: Some(self.timeouts.connect), + read_timeout: Some(self.timeouts.read), + user_agent_suffix: USER_AGENT_SUFFIX, + }) + } +} + +/// Failure classes of a source request. Variants carry a rendered message +/// rather than the SDK error so callers stay independent of the operation +/// error types; `class_label` is stable for metrics. +#[derive(Debug, thiserror::Error)] +pub enum SourceError { + #[error("source object not found")] + NotFound, + #[error("source denied access")] + AccessDenied, + #[error("source throttled the request")] + Throttled, + #[error("source request timed out")] + Timeout, + #[error("failed to connect to source: {0}")] + Connect(String), + #[error("source returned server error {0}")] + ServerError(u16), + #[error("unsupported source object: {0}")] + Unsupported(String), + #[error("source request failed: {0}")] + Other(String), +} + +impl SourceError { + /// Transient classes a caller may retry (subject to its own budget). + pub fn is_retryable(&self) -> bool { + matches!( + self, + SourceError::Throttled | SourceError::Timeout | SourceError::Connect(_) | SourceError::ServerError(_) + ) + } + + pub fn class_label(&self) -> &'static str { + match self { + SourceError::NotFound => "not_found", + SourceError::AccessDenied => "access_denied", + SourceError::Throttled => "throttled", + SourceError::Timeout => "timeout", + SourceError::Connect(_) => "connect", + SourceError::ServerError(_) => "server_error", + SourceError::Unsupported(_) => "unsupported", + SourceError::Other(_) => "other", + } + } +} + +const THROTTLE_CODES: &[&str] = &[ + "SlowDown", + "Throttling", + "ThrottlingException", + "RequestLimitExceeded", + "TooManyRequests", + "RequestThrottled", +]; +const NOT_FOUND_CODES: &[&str] = &["NoSuchKey", "NotFound", "NoSuchBucket", "NoSuchVersion"]; +const ACCESS_DENIED_CODES: &[&str] = &[ + "AccessDenied", + "InvalidAccessKeyId", + "SignatureDoesNotMatch", + "AllAccessDisabled", + "ExpiredToken", + "InvalidToken", +]; + +fn classify_status(status: u16, code: Option<&str>, message: String) -> SourceError { + if let Some(code) = code { + if THROTTLE_CODES.contains(&code) { + return SourceError::Throttled; + } + if NOT_FOUND_CODES.contains(&code) { + return SourceError::NotFound; + } + if ACCESS_DENIED_CODES.contains(&code) { + return SourceError::AccessDenied; + } + } + match status { + 404 => SourceError::NotFound, + 401 | 403 => SourceError::AccessDenied, + 429 | 503 => SourceError::Throttled, + 500..=599 => SourceError::ServerError(status), + _ => SourceError::Other(message), + } +} + +fn classify_sdk_error(err: SdkError) -> SourceError +where + E: ProvideErrorMetadata + std::error::Error + Send + Sync + 'static, +{ + let message = format!("{}", DisplayErrorContext(&err)); + match &err { + SdkError::TimeoutError(_) => SourceError::Timeout, + SdkError::DispatchFailure(failure) => { + if failure.is_timeout() { + SourceError::Timeout + } else if failure.is_io() { + SourceError::Connect(message) + } else { + SourceError::Other(message) + } + } + SdkError::ConstructionFailure(_) => SourceError::Other(message), + SdkError::ResponseError(response) => classify_status(response.raw().status().as_u16(), None, message), + SdkError::ServiceError(service) => classify_status(service.raw().status().as_u16(), err.code(), message), + _ => SourceError::Other(message), + } +} + +/// Server-side encryption the source reports for an object. Recognized only: +/// the write-back path stores plaintext bytes the source already decrypted. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum SourceSse { + S3, + Kms { key_id: Option }, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct SourceHead { + /// ETag with surrounding quotes stripped. + pub etag: Option, + /// `Content-Length` of the response: the object size for HEAD and + /// unranged GET, the range length for a ranged GET. + pub size: u64, + pub last_modified: Option, + pub content_type: Option, + pub content_encoding: Option, + pub content_disposition: Option, + pub content_language: Option, + pub cache_control: Option, + pub expires: Option, + /// `x-amz-meta-*` values keyed without the prefix, matching the stored + /// user-metadata shape. + pub user_metadata: HashMap, + pub version_id: Option, + pub storage_class: Option, + pub sse: Option, + pub is_multipart_etag: bool, +} + +/// Per-operation fields shared by HEAD and GET outputs. +struct HeadParts { + etag: Option, + content_length: Option, + last_modified: Option, + content_type: Option, + content_encoding: Option, + content_disposition: Option, + content_language: Option, + cache_control: Option, + expires: Option, + metadata: Option>, + version_id: Option, + storage_class: Option, + server_side_encryption: Option, + ssekms_key_id: Option, + sse_customer_algorithm: Option, +} + +fn normalize_etag(etag: Option) -> Option { + etag.map(|etag| etag.trim().trim_matches('"').to_string()) + .filter(|etag| !etag.is_empty()) +} + +/// Multipart ETags end in `-`; single-part ETags are bare MD5. +pub fn is_multipart_etag(etag: &str) -> bool { + etag.rsplit_once('-') + .is_some_and(|(_, parts)| !parts.is_empty() && parts.bytes().all(|b| b.is_ascii_digit())) +} + +fn system_time(value: Option) -> Option { + value.and_then(|value| SystemTime::try_from(value).ok()) +} + +fn source_head(parts: HeadParts) -> Result { + if parts.sse_customer_algorithm.is_some() { + return Err(SourceError::Unsupported( + "source object is encrypted with SSE-C; customer-key sources are not supported".to_string(), + )); + } + let size = parts + .content_length + .and_then(|length| u64::try_from(length).ok()) + .ok_or_else(|| SourceError::Other("source response has no valid content-length".to_string()))?; + let etag = normalize_etag(parts.etag); + let is_multipart_etag = etag.as_deref().is_some_and(is_multipart_etag); + let sse = parts.server_side_encryption.map(|sse| match sse { + ServerSideEncryption::Aes256 => SourceSse::S3, + _ => SourceSse::Kms { + key_id: parts.ssekms_key_id, + }, + }); + + Ok(SourceHead { + etag, + size, + last_modified: system_time(parts.last_modified), + content_type: parts.content_type, + content_encoding: parts.content_encoding, + content_disposition: parts.content_disposition, + content_language: parts.content_language, + cache_control: parts.cache_control, + expires: parts.expires, + user_metadata: parts.metadata.unwrap_or_default(), + version_id: parts.version_id, + storage_class: parts.storage_class, + sse, + is_multipart_etag, + }) +} + +fn source_head_from_head_output(output: HeadObjectOutput) -> Result { + source_head(HeadParts { + etag: output.e_tag, + content_length: output.content_length, + last_modified: output.last_modified, + content_type: output.content_type, + content_encoding: output.content_encoding, + content_disposition: output.content_disposition, + content_language: output.content_language, + cache_control: output.cache_control, + expires: output.expires_string, + metadata: output.metadata, + version_id: output.version_id, + storage_class: output.storage_class.map(|class| class.as_str().to_string()), + server_side_encryption: output.server_side_encryption, + ssekms_key_id: output.ssekms_key_id, + sse_customer_algorithm: output.sse_customer_algorithm, + }) +} + +/// Ranged/unranged GET response: `head` describes the returned bytes. +pub struct SourceGet { + pub head: SourceHead, + pub body: ByteStream, + /// `Content-Range` of a ranged response (`bytes a-b/total`). + pub content_range: Option, +} + +impl fmt::Debug for SourceGet { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("SourceGet") + .field("head", &self.head) + .field("content_range", &self.content_range) + .finish_non_exhaustive() + } +} + +fn source_get_from_output(output: GetObjectOutput) -> Result { + let content_range = output.content_range; + let body = output.body; + let head = source_head(HeadParts { + etag: output.e_tag, + content_length: output.content_length, + last_modified: output.last_modified, + content_type: output.content_type, + content_encoding: output.content_encoding, + content_disposition: output.content_disposition, + content_language: output.content_language, + cache_control: output.cache_control, + expires: output.expires_string, + metadata: output.metadata, + version_id: output.version_id, + storage_class: output.storage_class.map(|class| class.as_str().to_string()), + server_side_encryption: output.server_side_encryption, + ssekms_key_id: output.ssekms_key_id, + sse_customer_algorithm: output.sse_customer_algorithm, + })?; + Ok(SourceGet { + head, + body, + content_range, + }) +} + +/// Renders an `HTTPRangeSpec` as the `Range` header value sent to the source. +pub fn range_header_value(range: &HTTPRangeSpec) -> Result { + if range.is_suffix_length { + let suffix = range.start.unsigned_abs(); + if suffix == 0 { + return Err(SourceError::Other("invalid range: zero suffix length".to_string())); + } + return Ok(format!("bytes=-{suffix}")); + } + if range.start < 0 { + return Err(SourceError::Other("invalid range: negative start".to_string())); + } + match range.end { + -1 => Ok(format!("bytes={}-", range.start)), + end if end >= range.start => Ok(format!("bytes={}-{end}", range.start)), + _ => Err(SourceError::Other("invalid range: end precedes start".to_string())), + } +} + +/// One listing entry, keyed in the local namespace. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SourceObject { + pub key: String, + pub etag: Option, + pub size: u64, + pub last_modified: Option, + pub storage_class: Option, + pub is_multipart_etag: bool, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct SourcePage { + pub objects: Vec, + pub is_truncated: bool, + pub next_continuation_token: Option, +} + +/// Result of [`SourceClient::probe`]: the bucket answered HEAD and a +/// one-key listing succeeded. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SourceProbe { + pub sample_object: Option, + pub has_more_objects: bool, +} + +/// Adds the `source-proxy-request` anti-loop markers before signing so they +/// join the SigV4 canonical request (same shape as the replication proxy). +#[derive(Debug)] +struct SourceProxyMarkerInterceptor { + headers: HeaderMap, +} + +impl SourceProxyMarkerInterceptor { + fn new() -> Self { + let mut headers = HeaderMap::new(); + insert_header(&mut headers, SUFFIX_SOURCE_PROXY_REQUEST, "true"); + Self { headers } + } +} + +impl Intercept for SourceProxyMarkerInterceptor { + fn name(&self) -> &'static str { + "RustfsSourceProxyMarker" + } + + fn modify_before_signing( + &self, + context: &mut BeforeTransmitInterceptorContextMut<'_>, + _runtime_components: &RuntimeComponents, + _cfg: &mut ConfigBag, + ) -> Result<(), BoxError> { + let request_headers = context.request_mut().headers_mut(); + for (name, value) in &self.headers { + request_headers.try_insert(name.clone(), value.clone())?; + } + Ok(()) + } +} + +pub struct SourceClient { + client: S3Client, + endpoint: String, + bucket: String, + source_prefix: Option, + timeouts: SourceTimeouts, + bandwidth_limit: Option, +} + +impl fmt::Debug for SourceClient { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("SourceClient") + .field("endpoint", &self.endpoint) + .field("bucket", &self.bucket) + .field("source_prefix", &self.source_prefix) + .field("timeouts", &self.timeouts) + .field("bandwidth_limit", &self.bandwidth_limit) + .finish_non_exhaustive() + } +} + +impl SourceClient { + pub async fn new(spec: &SourceClientSpec) -> Result { + let endpoint = spec.endpoint_spec()?; + let config = build_remote_s3_config(&endpoint).await?; + Ok(Self::from_config_builder(config, endpoint.endpoint_url(), spec)) + } + + fn from_config_builder(config: aws_sdk_s3::config::Builder, endpoint: String, spec: &SourceClientSpec) -> Self { + let client = S3Client::from_conf(config.interceptor(SourceProxyMarkerInterceptor::new()).build()); + Self { + client, + endpoint, + bucket: spec.bucket.clone(), + source_prefix: spec.source_prefix.clone().filter(|prefix| !prefix.is_empty()), + timeouts: spec.timeouts, + bandwidth_limit: spec.bandwidth_limit, + } + } + + pub fn bucket(&self) -> &str { + &self.bucket + } + + pub fn source_prefix(&self) -> Option<&str> { + self.source_prefix.as_deref() + } + + pub fn timeouts(&self) -> SourceTimeouts { + self.timeouts + } + + pub fn bandwidth_limit(&self) -> Option { + self.bandwidth_limit + } + + /// Source-side key for a local key. + pub fn source_key(&self, local_key: &str) -> String { + match &self.source_prefix { + Some(prefix) => format!("{prefix}{local_key}"), + None => local_key.to_string(), + } + } + + /// Local key for a source key; `None` when the key lies outside the + /// configured prefix. + pub fn local_key<'a>(&self, source_key: &'a str) -> Option<&'a str> { + match &self.source_prefix { + Some(prefix) => source_key.strip_prefix(prefix.as_str()), + None => Some(source_key), + } + } + + pub async fn head_object(&self, key: &str) -> Result { + let output = self + .client + .head_object() + .bucket(&self.bucket) + .key(self.source_key(key)) + .send() + .await + .map_err(classify_sdk_error)?; + source_head_from_head_output(output) + } + + /// Streams the object; `range` is passed through as an HTTP `Range` + /// header and omitted entirely when `None`. + pub async fn get_object(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result { + let range = range.map(range_header_value).transpose()?; + let output = self + .client + .get_object() + .bucket(&self.bucket) + .key(self.source_key(key)) + .set_range(range) + .send() + .await + .map_err(classify_sdk_error)?; + source_get_from_output(output) + } + + /// Lists one page under the local `prefix`. Keys are returned in the + /// local namespace; entries outside `source_prefix` are skipped. + pub async fn list_objects_v2( + &self, + prefix: Option<&str>, + continuation_token: Option<&str>, + max_keys: i32, + ) -> Result { + let output = self + .client + .list_objects_v2() + .bucket(&self.bucket) + .prefix(self.source_key(prefix.unwrap_or_default())) + .set_continuation_token(continuation_token.map(str::to_string)) + .max_keys(max_keys) + .send() + .await + .map_err(classify_sdk_error)?; + + let is_truncated = output.is_truncated.unwrap_or(false); + let next_continuation_token = output.next_continuation_token; + if is_truncated && next_continuation_token.is_none() { + return Err(SourceError::Other( + "source reported a truncated listing without a continuation token".to_string(), + )); + } + let objects = output + .contents + .unwrap_or_default() + .into_iter() + .filter_map(|object| self.source_object(object)) + .collect(); + + Ok(SourcePage { + objects, + is_truncated, + next_continuation_token, + }) + } + + fn source_object(&self, object: SdkObject) -> Option { + let key = self.local_key(object.key.as_deref()?)?.to_string(); + let etag = normalize_etag(object.e_tag); + let is_multipart_etag = etag.as_deref().is_some_and(is_multipart_etag); + Some(SourceObject { + key, + etag, + size: object.size.and_then(|size| u64::try_from(size).ok()).unwrap_or(0), + last_modified: system_time(object.last_modified), + storage_class: object.storage_class.map(|class| class.as_str().to_string()), + is_multipart_etag, + }) + } + + pub async fn get_object_tagging(&self, key: &str) -> Result, SourceError> { + let output = self + .client + .get_object_tagging() + .bucket(&self.bucket) + .key(self.source_key(key)) + .send() + .await + .map_err(classify_sdk_error)?; + Ok(output.tag_set.into_iter().map(|tag| (tag.key, tag.value)).collect()) + } + + /// Admin validation: HeadBucket plus a one-key listing under the prefix. + pub async fn probe(&self) -> Result { + self.client + .head_bucket() + .bucket(&self.bucket) + .send() + .await + .map_err(classify_sdk_error)?; + let page = self.list_objects_v2(None, None, 1).await?; + Ok(SourceProbe { + sample_object: page.objects.into_iter().next(), + has_more_objects: page.is_truncated, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aws_sdk_s3::config::retry::RetryConfig; + use aws_smithy_runtime_api::client::http::{HttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn}; + use aws_smithy_runtime_api::client::orchestrator::HttpRequest; + use aws_smithy_runtime_api::client::result::ConnectorError; + use aws_smithy_runtime_api::http::StatusCode as SmithyStatusCode; + use aws_smithy_types::body::SdkBody; + use proptest::prelude::*; + use std::collections::VecDeque; + use std::sync::{Arc, Mutex}; + + #[derive(Clone, Debug)] + struct RecordedRequest { + method: String, + uri: String, + headers: Vec<(String, String)>, + } + + impl RecordedRequest { + fn header(&self, name: &str) -> Option<&str> { + self.headers + .iter() + .find(|(k, _)| k.eq_ignore_ascii_case(name)) + .map(|(_, v)| v.as_str()) + } + } + + #[derive(Clone, Debug)] + enum Scripted { + Response { + status: u16, + headers: Vec<(&'static str, String)>, + body: Vec, + }, + Io, + Timeout, + } + + fn ok(headers: Vec<(&'static str, String)>, body: &str) -> Scripted { + Scripted::Response { + status: 200, + headers, + body: body.as_bytes().to_vec(), + } + } + + fn status(status: u16, body: &str) -> Scripted { + Scripted::Response { + status, + headers: Vec::new(), + body: body.as_bytes().to_vec(), + } + } + + type Recorded = Arc>>; + + #[derive(Clone, Debug)] + struct ScriptedConnector { + requests: Recorded, + responses: Arc>>, + } + + impl HttpConnector for ScriptedConnector { + fn call(&self, request: HttpRequest) -> HttpConnectorFuture { + self.requests + .lock() + .expect("recorded request lock should not be poisoned") + .push(RecordedRequest { + method: request.method().to_string(), + uri: request.uri().to_string(), + headers: request + .headers() + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + }); + let next = self + .responses + .lock() + .expect("scripted response lock should not be poisoned") + .pop_front() + .expect("test script must provide a response for every request"); + match next { + Scripted::Response { status, headers, body } => { + let mut response = HttpResponse::new( + SmithyStatusCode::try_from(status).expect("scripted status should be valid"), + SdkBody::from(body), + ); + for (name, value) in headers { + response.headers_mut().insert(name, value); + } + HttpConnectorFuture::ready(Ok(response)) + } + Scripted::Io => HttpConnectorFuture::ready(Err(ConnectorError::io("connection refused".into()))), + Scripted::Timeout => HttpConnectorFuture::ready(Err(ConnectorError::timeout("connect timed out".into()))), + } + } + } + + fn spec(source_prefix: Option<&str>) -> SourceClientSpec { + SourceClientSpec { + endpoint: "https://source.example.com".to_string(), + region: "us-east-1".to_string(), + bucket: "source-bucket".to_string(), + source_prefix: source_prefix.map(str::to_string), + provider: SourceProvider::Minio, + path_style: PathStyle::Auto, + credentials: Some(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(), + }), + skip_tls_verify: false, + ca_cert_pem: None, + timeouts: SourceTimeouts::default(), + bandwidth_limit: NonZeroU64::new(1_000_000), + } + } + + async fn scripted_client(spec: &SourceClientSpec, responses: Vec) -> (SourceClient, Recorded) { + let requests: Recorded = Arc::new(Mutex::new(Vec::new())); + let connector = SharedHttpConnector::new(ScriptedConnector { + requests: Arc::clone(&requests), + responses: Arc::new(Mutex::new(responses.into_iter().collect())), + }); + let http_client = http_client_fn(move |_settings, _components| connector.clone()); + let endpoint = spec.endpoint_spec().expect("test spec endpoint should parse"); + let config = build_remote_s3_config(&endpoint) + .await + .expect("test spec should build") + .http_client(http_client) + .retry_config(RetryConfig::disabled()); + (SourceClient::from_config_builder(config, endpoint.endpoint_url(), spec), requests) + } + + fn recorded(requests: &Recorded) -> Vec { + requests.lock().expect("recorded request lock should not be poisoned").clone() + } + + fn assert_outbound_markers(request: &RecordedRequest) { + assert_eq!( + request.header("x-rustfs-source-proxy-request"), + Some("true"), + "{} {} must carry the rustfs anti-loop marker", + request.method, + request.uri + ); + assert_eq!( + request.header("x-minio-source-proxy-request"), + Some("true"), + "{} {} must carry the minio anti-loop marker", + request.method, + request.uri + ); + let user_agent = request.header("user-agent").expect("SDK request must carry a user-agent"); + assert!( + user_agent.ends_with(&format!(" {USER_AGENT_SUFFIX}")), + "user-agent {user_agent} must end with the migration suffix" + ); + assert!(request.header("authorization").is_some(), "request must be signed"); + assert!(request.header("x-amz-security-token").is_some(), "session token must be signed in"); + } + + fn head_headers() -> Vec<(&'static str, String)> { + vec![ + ("etag", "\"d41d8cd98f00b204e9800998ecf8427e-3\"".to_string()), + ("content-length", "1234".to_string()), + ("last-modified", "Wed, 21 Oct 2015 07:28:00 GMT".to_string()), + ("content-type", "text/plain".to_string()), + ("content-encoding", "gzip".to_string()), + ("content-disposition", "attachment; filename=\"a.txt\"".to_string()), + ("content-language", "en".to_string()), + ("cache-control", "max-age=60".to_string()), + ("expires", "Thu, 01 Jan 2026 00:00:00 GMT".to_string()), + ("x-amz-meta-owner", "alice".to_string()), + ("x-amz-meta-tier", "hot".to_string()), + ("x-amz-version-id", "v1".to_string()), + ("x-amz-storage-class", "STANDARD_IA".to_string()), + ("x-amz-server-side-encryption", "aws:kms".to_string()), + ("x-amz-server-side-encryption-aws-kms-key-id", "key-1".to_string()), + ] + } + + #[tokio::test] + async fn head_object_maps_source_head_fields() { + let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(head_headers(), "")]).await; + let head = client.head_object("dir/obj.txt").await.expect("HEAD should map"); + + let requests = recorded(&requests); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].method, "HEAD"); + assert!( + requests[0] + .uri + .starts_with("https://source.example.com/source-bucket/data/dir/obj.txt"), + "path-style URI with prefix expected, got {}", + requests[0].uri + ); + assert_outbound_markers(&requests[0]); + + assert_eq!(head.etag.as_deref(), Some("d41d8cd98f00b204e9800998ecf8427e-3")); + assert!(head.is_multipart_etag); + assert_eq!(head.size, 1234); + assert_eq!(head.last_modified, Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_445_412_480))); + assert_eq!(head.content_type.as_deref(), Some("text/plain")); + assert_eq!(head.content_encoding.as_deref(), Some("gzip")); + assert_eq!(head.content_disposition.as_deref(), Some("attachment; filename=\"a.txt\"")); + assert_eq!(head.content_language.as_deref(), Some("en")); + assert_eq!(head.cache_control.as_deref(), Some("max-age=60")); + assert_eq!(head.expires.as_deref(), Some("Thu, 01 Jan 2026 00:00:00 GMT")); + assert_eq!( + head.user_metadata, + HashMap::from([ + ("owner".to_string(), "alice".to_string()), + ("tier".to_string(), "hot".to_string()) + ]) + ); + assert_eq!(head.version_id.as_deref(), Some("v1")); + assert_eq!(head.storage_class.as_deref(), Some("STANDARD_IA")); + assert_eq!( + head.sse, + Some(SourceSse::Kms { + key_id: Some("key-1".to_string()) + }) + ); + } + + #[tokio::test] + async fn head_object_recognizes_sse_s3_and_single_part_etag() { + let headers = vec![ + ("etag", "\"d41d8cd98f00b204e9800998ecf8427e\"".to_string()), + ("content-length", "0".to_string()), + ("x-amz-server-side-encryption", "AES256".to_string()), + ]; + let (client, _) = scripted_client(&spec(None), vec![ok(headers, "")]).await; + let head = client.head_object("obj").await.expect("HEAD should map"); + assert_eq!(head.sse, Some(SourceSse::S3)); + assert!(!head.is_multipart_etag); + assert_eq!(head.size, 0); + assert!(head.user_metadata.is_empty()); + } + + #[tokio::test] + async fn head_object_rejects_sse_c_source_objects() { + let headers = vec![ + ("etag", "\"abc\"".to_string()), + ("content-length", "10".to_string()), + ("x-amz-server-side-encryption-customer-algorithm", "AES256".to_string()), + ]; + let (client, _) = scripted_client(&spec(None), vec![ok(headers, "")]).await; + let err = client + .head_object("obj") + .await + .expect_err("SSE-C source objects are unsupported"); + assert!(matches!(err, SourceError::Unsupported(_)), "{err:?}"); + assert_eq!(err.class_label(), "unsupported"); + assert!(!err.is_retryable()); + } + + #[tokio::test] + async fn get_object_passes_range_through_and_streams_body() { + let headers = vec![ + ("etag", "\"abc\"".to_string()), + ("content-length", "5".to_string()), + ("content-range", "bytes 10-14/100".to_string()), + ]; + let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(headers, "hello")]).await; + let range = HTTPRangeSpec { + is_suffix_length: false, + start: 10, + end: 14, + }; + let get = client + .get_object("obj", Some(&range)) + .await + .expect("ranged GET should succeed"); + + let requests = recorded(&requests); + assert_eq!(requests[0].method, "GET"); + assert_eq!(requests[0].header("range"), Some("bytes=10-14")); + assert_outbound_markers(&requests[0]); + + assert_eq!(get.content_range.as_deref(), Some("bytes 10-14/100")); + assert_eq!(get.head.size, 5); + let body = get.body.collect().await.expect("body should stream").into_bytes(); + assert_eq!(body.as_ref(), b"hello"); + } + + #[tokio::test] + async fn get_object_without_range_sends_no_range_header() { + let headers = vec![("etag", "\"abc\"".to_string()), ("content-length", "5".to_string())]; + let (client, requests) = scripted_client(&spec(None), vec![ok(headers, "hello")]).await; + let get = client.get_object("obj", None).await.expect("GET should succeed"); + let requests = recorded(&requests); + assert!(requests[0].header("range").is_none(), "unranged GET must not send Range"); + assert!(get.content_range.is_none()); + } + + #[test] + fn range_header_value_covers_open_and_suffix_forms() { + let render = |is_suffix_length, start, end| { + range_header_value(&HTTPRangeSpec { + is_suffix_length, + start, + end, + }) + }; + assert_eq!(render(false, 0, 99).expect("closed range"), "bytes=0-99"); + assert_eq!(render(false, 5, -1).expect("open range"), "bytes=5-"); + assert_eq!(render(true, 10, -1).expect("suffix range"), "bytes=-10"); + assert_eq!(render(true, -10, -1).expect("negative suffix range"), "bytes=-10"); + assert!(render(true, 0, -1).is_err()); + assert!(render(false, -1, 5).is_err()); + assert!(render(false, 10, 5).is_err()); + } + + const LIST_PAGE_ONE: &str = r#" + + source-bucket + data/photos/ + 2 + true + token-1 + + data/photos/a.jpg + 2015-10-21T07:28:00.000Z + "aaaa-2" + 42 + STANDARD + + + outside/b.jpg + "bbbb" + 7 + +"#; + + const LIST_PAGE_TWO: &str = r#" + + source-bucket + false + + data/photos/c.jpg + "cccc" + 1 + +"#; + + const LIST_TRUNCATED_WITHOUT_TOKEN: &str = r#" + + source-bucket + true +"#; + + #[tokio::test] + async fn list_objects_v2_pages_and_strips_source_prefix() { + let (client, requests) = + scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), LIST_PAGE_ONE), ok(Vec::new(), LIST_PAGE_TWO)]).await; + + let page = client + .list_objects_v2(Some("photos/"), None, 2) + .await + .expect("first page should list"); + assert!(page.is_truncated); + assert_eq!(page.next_continuation_token.as_deref(), Some("token-1")); + assert_eq!( + page.objects, + vec![SourceObject { + key: "photos/a.jpg".to_string(), + etag: Some("aaaa-2".to_string()), + size: 42, + last_modified: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1_445_412_480)), + storage_class: Some("STANDARD".to_string()), + is_multipart_etag: true, + }], + "entries outside the source prefix are dropped" + ); + + let page = client + .list_objects_v2(Some("photos/"), page.next_continuation_token.as_deref(), 2) + .await + .expect("second page should list"); + assert!(!page.is_truncated); + assert!(page.next_continuation_token.is_none()); + assert_eq!(page.objects.len(), 1); + assert_eq!(page.objects[0].key, "photos/c.jpg"); + assert!(!page.objects[0].is_multipart_etag); + + let requests = recorded(&requests); + assert_eq!(requests.len(), 2); + for request in &requests { + assert_eq!(request.method, "GET"); + assert!(request.uri.contains("list-type=2"), "{}", request.uri); + assert!(request.uri.contains("prefix=data%2Fphotos%2F"), "{}", request.uri); + assert!(request.uri.contains("max-keys=2"), "{}", request.uri); + assert_outbound_markers(request); + } + assert!(!requests[0].uri.contains("continuation-token"), "{}", requests[0].uri); + assert!(requests[1].uri.contains("continuation-token=token-1"), "{}", requests[1].uri); + } + + #[tokio::test] + async fn list_objects_v2_rejects_truncated_page_without_token() { + let (client, _) = scripted_client(&spec(None), vec![ok(Vec::new(), LIST_TRUNCATED_WITHOUT_TOKEN)]).await; + let err = client + .list_objects_v2(None, None, 10) + .await + .expect_err("truncated page without token is corrupt"); + assert!(matches!(err, SourceError::Other(_)), "{err:?}"); + } + + const TAGGING_BODY: &str = r#" + + + envprod + teamstorage + +"#; + + #[tokio::test] + async fn get_object_tagging_and_probe_carry_markers_on_every_request() { + let (client, requests) = scripted_client( + &spec(Some("data/")), + vec![ + ok(Vec::new(), TAGGING_BODY), + ok(Vec::new(), ""), + ok(Vec::new(), LIST_PAGE_ONE), + ], + ) + .await; + + let tags = client.get_object_tagging("obj").await.expect("tagging should parse"); + assert_eq!( + tags, + HashMap::from([ + ("env".to_string(), "prod".to_string()), + ("team".to_string(), "storage".to_string()) + ]) + ); + + let probe = client.probe().await.expect("probe should succeed"); + assert!(probe.has_more_objects); + assert_eq!(probe.sample_object.as_ref().map(|object| object.key.as_str()), Some("photos/a.jpg")); + + let requests = recorded(&requests); + assert_eq!(requests.len(), 3); + assert!(requests[0].uri.contains("tagging"), "{}", requests[0].uri); + assert_eq!(requests[1].method, "HEAD"); + assert!(requests[2].uri.contains("max-keys=1"), "{}", requests[2].uri); + for request in &requests { + assert_outbound_markers(request); + } + } + + const SLOW_DOWN_BODY: &str = r#" +SlowDownPlease reduce your request rate."#; + const ACCESS_DENIED_BODY: &str = r#" +AccessDeniedAccess Denied"#; + + #[tokio::test] + async fn source_error_classification_covers_every_class() { + let cases: Vec<(Scripted, &str, bool)> = vec![ + (status(404, ""), "not_found", false), + (status(403, ACCESS_DENIED_BODY), "access_denied", false), + (status(401, ""), "access_denied", false), + (status(429, ""), "throttled", true), + (status(503, SLOW_DOWN_BODY), "throttled", true), + (status(500, ""), "server_error", true), + (status(502, ""), "server_error", true), + (Scripted::Io, "connect", true), + (Scripted::Timeout, "timeout", true), + ]; + for (scripted, expected_label, retryable) in cases { + let (client, _) = scripted_client(&spec(None), vec![scripted.clone()]).await; + let err = match client.get_object("obj", None).await { + Ok(_) => panic!("{scripted:?} must fail"), + Err(err) => err, + }; + assert_eq!(err.class_label(), expected_label, "{scripted:?} -> {err:?}"); + assert_eq!(err.is_retryable(), retryable, "{scripted:?} -> {err:?}"); + if let SourceError::ServerError(code) = &err { + assert!(matches!(scripted, Scripted::Response { status, .. } if status == *code)); + } + } + + // HEAD carries no error body, so the classification must work from the + // status alone as well. + let (client, _) = scripted_client(&spec(None), vec![status(404, "")]).await; + assert!(matches!(client.head_object("missing").await, Err(SourceError::NotFound))); + let (client, _) = scripted_client(&spec(None), vec![status(403, "")]).await; + assert!(matches!(client.head_object("secret").await, Err(SourceError::AccessDenied))); + } + + #[tokio::test] + async fn source_client_debug_redacts_credentials() { + let (client, _) = scripted_client(&spec(Some("data/")), Vec::new()).await; + let rendered = format!("{client:?}"); + assert!(rendered.contains("source-bucket")); + assert!(rendered.contains("data/")); + assert!(rendered.contains("https://source.example.com")); + assert!(!rendered.contains("very-secret")); + assert!(!rendered.contains("session-token")); + assert!(!rendered.contains("access"), "access key must not be rendered either: {rendered}"); + } + + #[test] + fn source_client_spec_endpoint_parsing() { + let mut s = spec(None); + let endpoint = s.endpoint_spec().expect("https origin should parse"); + assert_eq!(endpoint.endpoint, "source.example.com"); + assert!(endpoint.secure); + assert_eq!(endpoint.user_agent_suffix, USER_AGENT_SUFFIX); + assert_eq!(endpoint.connect_timeout, Some(Duration::from_secs(10))); + assert_eq!(endpoint.read_timeout, Some(Duration::from_secs(60))); + assert_eq!(endpoint.path_style, PathStyle::Path); + + s.endpoint = "http://[::1]:9000".to_string(); + let endpoint = s.endpoint_spec().expect("bracketed IPv6 origin should parse"); + assert_eq!(endpoint.endpoint, "[::1]:9000"); + assert!(!endpoint.secure); + + for bad in [ + "ftp://source.example.com", + "https://user:pw@source.example.com", + "https://source.example.com/bucket", + "https://source.example.com/?x=1", + "not a url", + ] { + s.endpoint = bad.to_string(); + assert!( + matches!(s.endpoint_spec(), Err(RemoteS3ClientError::InvalidEndpoint(_))), + "{bad} must be rejected" + ); + } + } + + #[test] + fn resolve_path_style_auto_follows_provider_and_host() { + use SourceProvider::*; + for provider in [Aws, Gcs, R2] { + assert_eq!( + resolve_path_style(PathStyle::Auto, provider, "s3.example.com"), + PathStyle::VirtualHost, + "{provider:?}" + ); + } + for provider in [Minio, Rustfs, S3] { + assert_eq!( + resolve_path_style(PathStyle::Auto, provider, "s3.example.com"), + PathStyle::Path, + "{provider:?}" + ); + } + for host in ["10.0.0.1", "[::1]", "localhost", "LOCALHOST"] { + assert_eq!(resolve_path_style(PathStyle::Auto, Aws, host), PathStyle::Path, "{host}"); + } + assert_eq!(resolve_path_style(PathStyle::VirtualHost, Minio, "10.0.0.1"), PathStyle::VirtualHost); + assert_eq!(resolve_path_style(PathStyle::Path, Aws, "s3.amazonaws.com"), PathStyle::Path); + assert_eq!(SourceProvider::from_label(" AWS "), Some(Aws)); + assert_eq!(SourceProvider::from_label("azure"), None); + } + + fn prefix_client(prefix: Option) -> SourceClient { + SourceClient { + client: S3Client::from_conf( + aws_sdk_s3::Config::builder() + .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest()) + .build(), + ), + endpoint: "https://source.example.com".to_string(), + bucket: "bucket".to_string(), + source_prefix: prefix.filter(|prefix| !prefix.is_empty()), + timeouts: SourceTimeouts::default(), + bandwidth_limit: None, + } + } + + proptest! { + #[test] + fn source_and_local_keys_round_trip(prefix in proptest::option::of("[a-z0-9/_-]{0,16}"), key in "[a-zA-Z0-9/._ -]{0,32}") { + let client = prefix_client(prefix.clone()); + let source_key = client.source_key(&key); + prop_assert_eq!(client.local_key(&source_key), Some(key.as_str())); + match prefix.as_deref().filter(|prefix| !prefix.is_empty()) { + Some(prefix) => { + prop_assert!(source_key.starts_with(prefix)); + prop_assert_eq!(&source_key[prefix.len()..], key.as_str()); + } + None => prop_assert_eq!(source_key.as_str(), key.as_str()), + } + } + + #[test] + fn local_key_rejects_keys_outside_prefix(prefix in "[a-z]{1,8}/", key in "[a-z]{1,8}/[a-z]{0,8}") { + let client = prefix_client(Some(prefix.clone())); + let source_key = format!("{prefix}{key}"); + let inside = client.local_key(&source_key); + prop_assert_eq!(inside, Some(key.as_str())); + if !key.starts_with(&prefix) { + prop_assert_eq!(client.local_key(&key), None); + } + } + } +} 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/docs/operations/outbound-connection-policy.md b/docs/operations/outbound-connection-policy.md index 968b8a87e..770c40e12 100644 --- a/docs/operations/outbound-connection-policy.md +++ b/docs/operations/outbound-connection-policy.md @@ -18,7 +18,7 @@ RustFS validates every operator-configured outbound destination to close a serve | Target configuration validation (startup and admin API) | Full policy | `crates/targets/src/config/common.rs` `validate_outbound_http_url`; `rustfs/src/admin/handlers/target_descriptor.rs` | | OIDC discovery, JWKS, and token requests | Full policy | A blocked provider logs `OIDC provider discovery blocked by outbound policy` naming the origin to allowlist (`crates/iam/src/oidc.rs`) | | Object Lambda targets | Full policy | `rustfs/src/admin/router.rs` `outbound_policy` | -| Bucket replication targets | Literal check, relaxed | Private addresses are always allowed; loopback only with `RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET=true` (`crates/ecstore/src/bucket/bucket_target_sys.rs` `validate_replication_target_endpoint`) | +| Bucket replication targets | Literal check, relaxed | Private addresses are always allowed; loopback only with `RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET=true` (`crates/ecstore/src/bucket/remote_s3_client.rs` `validate_remote_endpoint`, shared with on-demand migration sources) | | Site replication peers | Literal check | `rustfs/src/site_replication/mod.rs` | | Tiering warm backends (S3, MinIO, RustFS, Azure, GCS, Aliyun, Tencent, Huawei, R2) | Literal check | `crates/ecstore/src/services/tier/warm_backend.rs` `validate_endpoint`; the RustFS provider adds a debug-only, env-gated loopback exception for e2e tests | | Keystone `auth_url` | Literal check | `crates/keystone/src/config.rs` | 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