feat(replication): support custom TLS for bucket targets (#3825)

* feat(replication): support insecure https bucket targets * feat(replication): support custom ca bucket targets * test(replication): satisfy e2e clippy for TLS helpers * fix(replication): avoid native root panic for custom trust stores * test(replication): decouple private IP target test from TLS roots * test(replication): use target TLS client in private IP unit test
2026-08-08 22:33:22 +00:00 · 2026-06-24 20:23:20 +08:00
parent fcd0c9ec0f
commit 623fc801f1
8 changed files with 1056 additions and 16 deletions
@@ -24,6 +24,7 @@ use crate::bucket::versioning_sys::BucketVersioningSys;
 use crate::runtime_sources;
 use aws_credential_types::Credentials as SdkCredentials;
 use aws_sdk_s3::config::Region as SdkRegion;
+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,11 +38,20 @@ use aws_sdk_s3::types::{
 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 http::{HeaderMap, HeaderName, HeaderValue, StatusCode};
+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 http::{HeaderMap, HeaderName, HeaderValue, StatusCode, Uri};
+use hyper_util::client::legacy::Client as HyperClient;
+use hyper_util::rt::{TokioExecutor, TokioTimer};
 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_filemeta::{ReplicationStatusType, ReplicationType};
-use rustfs_utils::egress::validate_outbound_url;
+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_STORAGE_CLASS, AMZ_WEBSITE_REDIRECT_LOCATION, is_amz_header, is_minio_header,
@@ -51,6 +61,7 @@ use rustfs_utils::http::{
    SUFFIX_FORCE_DELETE, SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_ETAG, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK,
    SUFFIX_SOURCE_REPLICATION_REQUEST, SUFFIX_SOURCE_VERSION_ID, insert_header,
 };
+use rustls_pki_types::pem::PemObject;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::error::Error;
@@ -63,6 +74,7 @@ use std::time::{Duration, Instant};
 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;
@@ -652,7 +664,7 @@ impl BucketTargetSys {
            access_key: credentials.access_key.clone(),
            error: format!("invalid target endpoint: {err}"),
        })?;
-        validate_outbound_url(&parsed_endpoint).map_err(|err| BucketTargetError::RemoteTargetConnectionErr {
+        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}"),
@@ -668,8 +680,14 @@ impl BucketTargetSys {
            config_builder = config_builder.force_path_style(true);
        }

-        if target.secure
-            && let Some(http_client) = build_aws_s3_http_client_from_tls_path().await
+        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);
        }
@@ -821,6 +839,170 @@ 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()
+}
+
+fn validate_replication_target_endpoint(url: &Url) -> Result<(), OutboundUrlError> {
+    match validate_outbound_url(url) {
+        Ok(()) => Ok(()),
+        Err(OutboundUrlError::ForbiddenHost {
+            reason: "private address",
+            ..
+        }) => 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 build_aws_s3_http_client_from_target_ca_pem(ca_cert_pem: &str) -> Result<SharedHttpClient, BucketTargetError> {
+    let certs = rustls_pki_types::CertificateDer::pem_slice_iter(ca_cert_pem.as_bytes())
+        .collect::<Result<Vec<_>, _>>()
+        .map_err(|err| BucketTargetError::Io(std::io::Error::other(format!("invalid target CA PEM: {err}"))))?;
+
+    if certs.is_empty() {
+        return Err(BucketTargetError::Io(std::io::Error::other(
+            "invalid target CA PEM: no certificates found",
+        )));
+    }
+
+    let mut trust_store = smithy_tls::TrustStore::empty();
+    trust_store.add_pem_certificate(ca_cert_pem.as_bytes());
+
+    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 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).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 tls_path = rustfs_utils::get_env_str(rustfs_config::ENV_RUSTFS_TLS_PATH, rustfs_config::DEFAULT_RUSTFS_TLS_PATH);
    if tls_path.is_empty() {
@@ -828,7 +1010,7 @@ async fn build_aws_s3_http_client_from_tls_path() -> Option<aws_sdk_s3::config::
    }

    let tls_dir = Path::new(&tls_path);
-    let mut trust_store = smithy_tls::TrustStore::default();
+    let mut trust_store = smithy_tls::TrustStore::empty();
    let mut has_custom_certs = false;

    let ca_path = tls_dir.join(RUSTFS_CA_CERT);
@@ -1606,6 +1788,7 @@ impl Error for BucketTargetError {}
 #[cfg(test)]
 mod tests {
    use super::*;
+    use rcgen::generate_simple_self_signed;

    #[test]
    fn build_remove_object_headers_includes_internal_version_id_for_replication_delete() {
@@ -1674,4 +1857,77 @@ mod tests {

        assert!(err.to_string().contains("not allowed"));
    }
+
+    #[tokio::test]
+    async fn get_remote_target_client_internal_allows_private_ip_endpoint() {
+        let sys = BucketTargetSys::default();
+        let client = sys
+            .get_remote_target_client_internal(&BucketTarget {
+                endpoint: "192.168.1.10:9000".to_string(),
+                secure: true,
+                skip_tls_verify: true,
+                target_bucket: "bucket".to_string(),
+                region: "us-east-1".to_string(),
+                credentials: Some(Credentials {
+                    access_key: "access".to_string(),
+                    secret_key: "secret".to_string(),
+                    session_token: None,
+                    expiration: None,
+                }),
+                ..Default::default()
+            })
+            .await
+            .expect("private IP endpoints should be allowed for replication targets");
+
+        assert_eq!(client.endpoint, "https://192.168.1.10:9000");
+    }
+
+    #[tokio::test]
+    async fn get_remote_target_client_internal_allows_custom_ca_pem() {
+        let sys = BucketTargetSys::default();
+        let cert = generate_simple_self_signed(vec!["192.168.1.10".to_string()]).expect("certificate should generate");
+        let client = sys
+            .get_remote_target_client_internal(&BucketTarget {
+                endpoint: "192.168.1.10:9000".to_string(),
+                secure: true,
+                target_bucket: "bucket".to_string(),
+                region: "us-east-1".to_string(),
+                ca_cert_pem: cert.cert.pem(),
+                credentials: Some(Credentials {
+                    access_key: "access".to_string(),
+                    secret_key: "secret".to_string(),
+                    session_token: None,
+                    expiration: None,
+                }),
+                ..Default::default()
+            })
+            .await
+            .expect("custom CA PEM should build a target client");
+
+        assert_eq!(client.endpoint, "https://192.168.1.10:9000");
+    }
+
+    #[tokio::test]
+    async fn get_remote_target_client_internal_rejects_invalid_custom_ca_pem() {
+        let sys = BucketTargetSys::default();
+        let err = sys
+            .get_remote_target_client_internal(&BucketTarget {
+                endpoint: "192.168.1.10:9000".to_string(),
+                secure: true,
+                target_bucket: "bucket".to_string(),
+                region: "us-east-1".to_string(),
+                ca_cert_pem: "not a pem".to_string(),
+                credentials: Some(Credentials {
+                    access_key: "access".to_string(),
+                    secret_key: "secret".to_string(),
+                    session_token: None,
+                    expiration: None,
+                }),
+                ..Default::default()
+            })
+            .await
+            .expect_err("invalid custom CA PEM should be rejected");
+
+        assert!(err.to_string().contains("invalid target CA PEM"));
+    }
 }
@@ -165,6 +165,10 @@ pub struct BucketTarget {
    pub replication_sync: bool,
    #[serde(default)]
    pub storage_class: String,
+    #[serde(rename = "skipTlsVerify", default)]
+    pub skip_tls_verify: bool,
+    #[serde(rename = "caCertPem", default)]
+    pub ca_cert_pem: String,
    #[serde(rename = "healthCheckDuration", with = "duration_seconds", default)]
    pub health_check_duration: Duration,
    #[serde(rename = "disableProxy", default)]
@@ -277,6 +281,8 @@ mod tests {
            "bandwidth_limit": 1000000,
            "replicationSync": true,
            "storage_class": "STANDARD",
+            "skipTlsVerify": true,
+            "caCertPem": "-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----\n",
            "healthCheckDuration": 30,
            "disableProxy": false,
            "resetBeforeDate": null,
@@ -314,6 +320,8 @@ mod tests {
        assert_eq!(target.bandwidth_limit, 1000000);
        assert!(target.replication_sync);
        assert_eq!(target.storage_class, "STANDARD");
+        assert!(target.skip_tls_verify);
+        assert_eq!(target.ca_cert_pem, "-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----\n");
        assert_eq!(target.health_check_duration, Duration::from_secs(30));
        assert!(!target.disable_proxy);
        assert_eq!(target.reset_id, "reset-123");
@@ -363,6 +371,8 @@ mod tests {
            bandwidth_limit: 500000,
            replication_sync: false,
            storage_class: "REDUCED_REDUNDANCY".to_string(),
+            skip_tls_verify: true,
+            ca_cert_pem: "-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----\n".to_string(),
            health_check_duration: Duration::from_secs(60),
            disable_proxy: true,
            reset_before_date: Some(OffsetDateTime::now_utc()),
@@ -396,6 +406,8 @@ mod tests {
        assert_eq!(original.region, deserialized.region);
        assert_eq!(original.bandwidth_limit, deserialized.bandwidth_limit);
        assert_eq!(original.replication_sync, deserialized.replication_sync);
+        assert_eq!(original.skip_tls_verify, deserialized.skip_tls_verify);
+        assert_eq!(original.ca_cert_pem, deserialized.ca_cert_pem);
        assert_eq!(original.health_check_duration, deserialized.health_check_duration);
        assert_eq!(original.online, deserialized.online);
        assert_eq!(original.edge, deserialized.edge);
@@ -478,6 +490,8 @@ mod tests {
                    "bandwidth_limit": 0,
                    "replicationSync": false,
                    "storage_class": "",
+                    "skipTlsVerify": false,
+                    "caCertPem": "",
                    "healthCheckDuration": 0,
                    "disableProxy": false,
                    "resetBeforeDate": null,
@@ -526,6 +540,8 @@ mod tests {
            "api": "s3v4",
            "type": "replication",
            "replicationSync": false,
+            "skipTlsVerify": true,
+            "caCertPem": "-----BEGIN CERTIFICATE-----\nMC4x\n-----END CERTIFICATE-----\n",
            "healthCheckDuration": 60,
            "disableProxy": false,
            "resetBeforeDate": "0001-01-01T00:00:00Z",
@@ -556,6 +572,8 @@ mod tests {
        assert_eq!(target.api, "s3v4");
        assert_eq!(target.target_type, BucketTargetType::ReplicationService);
        assert!(!target.replication_sync);
+        assert!(target.skip_tls_verify);
+        assert_eq!(target.ca_cert_pem, "-----BEGIN CERTIFICATE-----\nMC4x\n-----END CERTIFICATE-----\n");
        assert_eq!(target.health_check_duration, Duration::from_secs(60));
        assert!(!target.disable_proxy);
        assert!(!target.online);
@@ -604,6 +622,8 @@ mod tests {
                    "region": "",
                    "replicationSync": false,
                    "storage_class": "",
+                    "skipTlsVerify": true,
+                    "caCertPem": "-----BEGIN CERTIFICATE-----\nMC4x\n-----END CERTIFICATE-----\n",
                    "healthCheckDuration": 60,
                    "disableProxy": false,
                    "resetBeforeDate": "0001-01-01T00:00:00Z",
@@ -637,6 +657,8 @@ mod tests {
        assert_eq!(target.endpoint, "localhost:8000");
        assert_eq!(target.target_bucket, "test");
        assert_eq!(target.bandwidth_limit, 107374182400);
+        assert!(target.skip_tls_verify);
+        assert_eq!(target.ca_cert_pem, "-----BEGIN CERTIFICATE-----\nMC4x\n-----END CERTIFICATE-----\n");

        println!("✅ User provided JSON successfully deserialized to BucketTargets");
    }
@@ -670,6 +692,8 @@ mod tests {
        assert_eq!(target.bandwidth_limit, 0); // i64 default is 0
        assert!(!target.replication_sync); // bool default is false
        assert_eq!(target.storage_class, ""); // String default is empty
+        assert!(!target.skip_tls_verify); // bool default is false
+        assert_eq!(target.ca_cert_pem, ""); // String default is empty
        assert_eq!(target.health_check_duration, Duration::from_secs(0)); // Duration default
        assert!(!target.disable_proxy); // bool default is false
        assert!(target.reset_before_date.is_none()); // Option default is None
@@ -709,6 +733,8 @@ mod tests {
        assert_eq!(target.bandwidth_limit, 0);
        assert!(!target.replication_sync);
        assert_eq!(target.storage_class, "");
+        assert!(!target.skip_tls_verify);
+        assert_eq!(target.ca_cert_pem, "");
        assert_eq!(target.health_check_duration, Duration::from_secs(0));
        assert!(!target.disable_proxy);
        assert!(target.reset_before_date.is_none());
@@ -746,6 +772,8 @@ mod tests {
            "api": "s3v4",
            "type": "replication",
            "replicationSync": false,
+            "skipTlsVerify": true,
+            "caCertPem": "-----BEGIN CERTIFICATE-----\nMC4x\n-----END CERTIFICATE-----\n",
            "healthCheckDuration": 60,
            "disableProxy": false,
            "resetBeforeDate": "0001-01-01T00:00:00Z",
@@ -774,6 +802,8 @@ mod tests {
        assert_eq!(target.api, "s3v4");
        assert_eq!(target.target_type, BucketTargetType::ReplicationService);
        assert!(!target.replication_sync);
+        assert!(target.skip_tls_verify);
+        assert_eq!(target.ca_cert_pem, "-----BEGIN CERTIFICATE-----\nMC4x\n-----END CERTIFICATE-----\n");
        assert_eq!(target.health_check_duration, Duration::from_secs(60));
        assert!(!target.disable_proxy);
        assert!(!target.online);