fran/rustfs
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor(ecstore): shared remote S3 client builder and ODM source client (#7067)

Browse Source 
* 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/<version> 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
This commit is contained in:
Zhengchao An
2026-09-02 21:59:11 +08:00
committed by GitHub
parent 99f85ca2b1
commit 7e1f261e38
8 changed files with 2295 additions and 471 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/api/mod.rs
+17
View File
@@ -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,
crates/ecstore/src/bucket/bucket_target_sys.rs
+125 -469
View File
@@ -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 = "<redacted>";
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<SdkCredentials> {
self.resolve_at(SystemTime::now()).ok()
}
}
fn remote_target_sdk_credentials(
credentials: &Credentials,
account_id: &str,
now: SystemTime,
) -> Result<SdkCredentials, &'static str> {
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<SdkError<HeadObjectError>>;
pub type GetObjectSdkError = Box<SdkError<GetObjectError>>;
pub type GetObjectTaggingSdkError = Box<SdkError<GetObjectTaggingError>>;
@@ -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<rustls::client::danger::ServerCertVerified, rustls::Error> {
Ok(rustls::client::danger::ServerCertVerified::assertion())
}
fn verify_tls12_signature(
&self,
_message: &[u8],
_cert: &rustls_pki_types::CertificateDer<'_>,
_dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
}
fn verify_tls13_signature(
&self,
_message: &[u8],
_cert: &rustls_pki_types::CertificateDer<'_>,
_dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
}
fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
rustls::crypto::aws_lc_rs::default_provider()
.signature_verification_algorithms
.supported_schemes()
}
}
#[derive(Clone)]
struct TargetHyperHttpConnector<C> {
client: HyperClient<C, SdkBody>,
}
impl<C> fmt::Debug for TargetHyperHttpConnector<C> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("TargetHyperHttpConnector")
.field("client", &"** hyper client **")
.finish()
}
}
impl<C> SmithyHttpConnector for TargetHyperHttpConnector<C>
where
C: Clone + Send + Sync + 'static,
C: Service<Uri>,
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<BoxError>,
{
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::<Result<Vec<_>, _>>()
.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<Item = Vec<u8>>) -> (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<SharedHttpClient, BucketTargetError> {
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<Vec<u8>> {
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<Vec<u8>> {
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<SharedHttpClient, BucketTargetError> {
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<Option<SharedHttpClient>, 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<aws_sdk_s3::config::SharedHttpClient> {
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");
crates/ecstore/src/bucket/mod.rs
+2
View File
@@ -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;
crates/ecstore/src/bucket/on_demand_migration/mod.rs
+18
View File
@@ -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;
crates/ecstore/src/bucket/on_demand_migration/source_client.rs
+1343
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/bucket/remote_s3_client.rs
+789
View File
@@ -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 = "<redacted>";
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<String>,
pub expiration: Option<SystemTime>,
/// 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<RemoteCredentials>,
/// 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<String>,
pub connect_timeout: Option<Duration>,
pub read_timeout: Option<Duration>,
/// 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<SdkCredentials> {
self.resolve_at(SystemTime::now()).ok()
}
}
pub(crate) fn remote_sdk_credentials(credentials: &RemoteCredentials, now: SystemTime) -> Result<SdkCredentials, &'static str> {
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<aws_sdk_s3::config::Builder, RemoteS3ClientError> {
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<S3Client, RemoteS3ClientError> {
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<rustls::client::danger::ServerCertVerified, rustls::Error> {
Ok(rustls::client::danger::ServerCertVerified::assertion())
}
fn verify_tls12_signature(
&self,
_message: &[u8],
_cert: &rustls_pki_types::CertificateDer<'_>,
_dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
}
fn verify_tls13_signature(
&self,
_message: &[u8],
_cert: &rustls_pki_types::CertificateDer<'_>,
_dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
}
fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
rustls::crypto::aws_lc_rs::default_provider()
.signature_verification_algorithms
.supported_schemes()
}
}
#[derive(Clone)]
struct TargetHyperHttpConnector<C> {
client: HyperClient<C, SdkBody>,
}
impl<C> fmt::Debug for TargetHyperHttpConnector<C> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("TargetHyperHttpConnector")
.field("client", &"** hyper client **")
.finish()
}
}
impl<C> SmithyHttpConnector for TargetHyperHttpConnector<C>
where
C: Clone + Send + Sync + 'static,
C: Service<Uri>,
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<BoxError>,
{
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::<Result<Vec<_>, _>>()
.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<Item = Vec<u8>>,
) -> (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<SharedHttpClient, RemoteS3ClientError> {
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<Vec<u8>> {
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<Vec<u8>> {
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<SharedHttpClient, RemoteS3ClientError> {
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<Option<SharedHttpClient>, 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<SharedHttpClient> {
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<Mutex<Vec<Vec<(String, String)>>>>;
#[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"
);
}
}
docs/operations/outbound-connection-policy.md
+1 -1
View File
@@ -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` |
scripts/error-other-format-baseline.txt
-1
View File
@@ -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