From 64705d7589afb0e37de58d5e1483ed6e618e4685 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 28 Aug 2026 19:49:27 +0800 Subject: [PATCH] refactor(ecstore): migrate aliyun/azure warm backends to shared S3 constructor (#6774) --- .../ecstore/src/services/tier/warm_backend.rs | 33 ++++-- .../src/services/tier/warm_backend_aliyun.rs | 104 ++++-------------- .../src/services/tier/warm_backend_azure.rs | 104 ++++-------------- scripts/error-other-format-baseline.txt | 3 +- 4 files changed, 74 insertions(+), 170 deletions(-) diff --git a/crates/ecstore/src/services/tier/warm_backend.rs b/crates/ecstore/src/services/tier/warm_backend.rs index 731a0a321..cc1392b27 100644 --- a/crates/ecstore/src/services/tier/warm_backend.rs +++ b/crates/ecstore/src/services/tier/warm_backend.rs @@ -43,6 +43,7 @@ use rustfs_s3_client::{ api_put_object::{AdvancedPutOptions, PutObjectOptions}, transition_api::{ReadCloser, ReaderImpl}, }; +use rustfs_utils::egress::validate_outbound_url; use rustfs_utils::http::headers::{ CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, EXPIRES, HeaderExt as _, }; @@ -249,6 +250,13 @@ pub(crate) struct S3CompatibleWarmBackendParams<'a> { /// Tag handed to [`TransitionClient::new`] so per-provider client behavior /// and metrics stay attributable. pub provider_tag: &'a str, + /// SSRF guard run against the parsed endpoint once it's known to have a + /// host. Almost every provider passes [`rustfs_utils::egress::validate_outbound_url`] + /// unchanged; RustFS passes its own wrapper that adds a debug-only, + /// env-gated loopback exception for its e2e tier tests (see + /// rustfs/rustfs#6773) — the shared constructor stays the single call + /// site either way, so no provider can silently end up unvalidated. + pub validate_endpoint: fn(&url::Url) -> Result<(), rustfs_utils::egress::OutboundUrlError>, } /// Build the [`WarmBackendS3`] shared by the S3-compatible warm backend providers. @@ -256,10 +264,6 @@ pub(crate) struct S3CompatibleWarmBackendParams<'a> { /// Credential, bucket, and endpoint validation run in this order because the /// existing provider constructors report the first failure they hit, and their /// error texts are user-visible through the tier admin API. -#[allow( - dead_code, - reason = "expand step of the shared warm-backend extraction; the per-provider migrate step adds the production callers (backlog#2040)" -)] pub(crate) async fn new_s3_compatible_warm_backend( params: S3CompatibleWarmBackendParams<'_>, ) -> Result { @@ -298,6 +302,10 @@ pub(crate) async fn new_s3_compatible_warm_backend( let host = u .host_str() .ok_or_else(|| std::io::Error::other("Invalid endpoint URL: missing host"))?; + // Runs after the host-presence check above (not immediately after Url::parse) so a + // host-less endpoint still reports this constructor's own "missing host" text instead of + // validate_endpoint's differently-worded rejection for the same input. + (params.validate_endpoint)(&u).map_err(|err| std::io::Error::other(format!("tier endpoint is not allowed: {err}")))?; let client = TransitionClient::new(&format!("{}:{}", host, u.port().unwrap_or(default_port)), opts, params.provider_tag).await?; @@ -317,10 +325,6 @@ pub(crate) async fn new_s3_compatible_warm_backend( /// /// `object_size == -1` means "length unknown", so the caller is charged the /// worst case of a full [`MAX_MULTIPART_PUT_OBJECT_SIZE`] object. -#[allow( - dead_code, - reason = "expand step of the shared warm-backend extraction; the per-provider migrate step adds the production callers (backlog#2040)" -)] pub(crate) fn optimal_part_size(object_size: i64, min_part_size: i64) -> Result { let mut object_size = object_size; if object_size == -1 { @@ -936,6 +940,7 @@ mod tests { region: "us-east-1", bucket_lookup: BucketLookupType::BucketLookupDNS, provider_tag: "aliyun", + validate_endpoint: validate_outbound_url, } } @@ -984,6 +989,18 @@ mod tests { assert_eq!(err.to_string(), "Invalid endpoint URL: missing host"); } + /// Every migrated provider that uses `validate_outbound_url` directly (all + /// but RustFS, which injects its own debug-only, env-gated wrapper — see + /// rustfs/rustfs#6773) goes through this one construction path, so the + /// SSRF guard only needs to be pinned here rather than once per provider + /// file (see backlog#2040's migrate steps and rustfs/rustfs#6764). + #[tokio::test] + async fn s3_compatible_backend_rejects_a_loopback_endpoint_before_any_network_setup() { + let err = init_error(s3_compatible_params("https://127.0.0.1:9000"), "a loopback endpoint must be rejected").await; + + assert!(err.to_string().contains("not allowed"), "unexpected error: {err}"); + } + #[tokio::test] async fn s3_compatible_backend_carries_provider_options_to_the_transition_client() { let backend = new_s3_compatible_warm_backend(s3_compatible_params("http://tier.example.com:9000")) diff --git a/crates/ecstore/src/services/tier/warm_backend_aliyun.rs b/crates/ecstore/src/services/tier/warm_backend_aliyun.rs index 8ebe17658..e848f4f29 100644 --- a/crates/ecstore/src/services/tier/warm_backend_aliyun.rs +++ b/crates/ecstore/src/services/tier/warm_backend_aliyun.rs @@ -19,77 +19,38 @@ #![allow(clippy::all)] use std::collections::HashMap; -use std::sync::Arc; use crate::services::tier::{ tier_config::TierAliyun, - warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options}, + warm_backend::{ + S3CompatibleWarmBackendParams, WarmBackend, WarmBackendGetOpts, build_transition_put_options, + new_s3_compatible_warm_backend, optimal_part_size, + }, warm_backend_s3::WarmBackendS3, }; -use rustfs_s3_client::{ - admin_handler_utils::AdminError, - api_put_object::PutObjectOptions, - credentials::{Credentials, SignatureType, Static, Value}, - transition_api::{BucketLookupType, Options, ReadCloser, ReaderImpl, TransitionClient, TransitionCore}, -}; +use rustfs_s3_client::transition_api::{BucketLookupType, ReadCloser, ReaderImpl}; use rustfs_utils::egress::validate_outbound_url; -use tracing::warn; -const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5; -const MAX_PARTS_COUNT: i64 = 10000; -const _MAX_PART_SIZE: i64 = 1024 * 1024 * 1024 * 5; const MIN_PART_SIZE: i64 = 1024 * 1024 * 128; pub struct WarmBackendAliyun(WarmBackendS3); impl WarmBackendAliyun { pub async fn new(conf: &TierAliyun, tier: &str) -> Result { - if conf.access_key == "" || conf.secret_key == "" { - return Err(std::io::Error::other("both access and secret keys are required")); - } - - if conf.bucket == "" { - return Err(std::io::Error::other("no bucket name was provided")); - } - - let u = match url::Url::parse(&conf.endpoint) { - Ok(u) => u, - Err(e) => { - return Err(std::io::Error::other(e.to_string())); - } - }; - validate_outbound_url(&u).map_err(|err| std::io::Error::other(format!("tier endpoint is not allowed: {err}")))?; - - let creds = Credentials::new(Static(Value { - access_key_id: conf.access_key.clone(), - secret_access_key: conf.secret_key.clone(), - session_token: "".to_string(), - signer_type: SignatureType::SignatureV4, - ..Default::default() - })); - let opts = Options { - creds, - secure: u.scheme() == "https", - region: conf.region.clone(), - bucket_lookup: BucketLookupType::BucketLookupDNS, - ..Default::default() - }; - let scheme = u.scheme(); - let default_port = if scheme == "https" { 443 } else { 80 }; - let host = u - .host_str() - .ok_or_else(|| std::io::Error::other("Invalid endpoint URL: missing host"))?; - let client = TransitionClient::new(&format!("{}:{}", host, u.port().unwrap_or(default_port)), opts, "aliyun").await?; - - let client = Arc::new(client); - let core = TransitionCore(Arc::clone(&client)); - Ok(Self(WarmBackendS3 { - client, - core, - bucket: conf.bucket.clone(), - prefix: conf.prefix.strip_suffix("/").unwrap_or(&conf.prefix).to_owned(), - storage_class: "".to_string(), - })) + Ok(Self( + new_s3_compatible_warm_backend(S3CompatibleWarmBackendParams { + endpoint: &conf.endpoint, + access_key: &conf.access_key, + secret_key: &conf.secret_key, + bucket: &conf.bucket, + prefix: &conf.prefix, + region: &conf.region, + bucket_lookup: BucketLookupType::BucketLookupDNS, + provider_tag: "aliyun", + validate_endpoint: validate_outbound_url, + }) + .await?, + )) } } @@ -102,7 +63,7 @@ impl WarmBackend for WarmBackendAliyun { length: i64, meta: HashMap, ) -> Result { - let part_size = optimal_part_size(length)?; + let part_size = optimal_part_size(length, MIN_PART_SIZE)?; let client = self.0.client.clone(); let res = client .put_object(&self.0.bucket, &self.0.get_dest(object), r, length, &{ @@ -133,32 +94,15 @@ impl WarmBackend for WarmBackendAliyun { } } -fn optimal_part_size(object_size: i64) -> Result { - let mut object_size = object_size; - if object_size == -1 { - object_size = MAX_MULTIPART_PUT_OBJECT_SIZE; - } - - if object_size > MAX_MULTIPART_PUT_OBJECT_SIZE { - return Err(std::io::Error::other("entity too large")); - } - - let configured_part_size = MIN_PART_SIZE; - let mut part_size_flt = object_size as f64 / MAX_PARTS_COUNT as f64; - part_size_flt = (part_size_flt as f64 / configured_part_size as f64).ceil() * configured_part_size as f64; - - let part_size = part_size_flt as i64; - if part_size == 0 { - return Ok(MIN_PART_SIZE); - } - Ok(part_size) -} - #[cfg(test)] mod tests { use super::*; use crate::services::tier::tier_config::TierAliyun; + /// The SSRF guard itself is exercised once, generically, in + /// `warm_backend::tests` (see backlog#2040/backlog#2041 and + /// rustfs/rustfs#6764) — this test only pins that this provider's + /// production constructor really is wired through that shared path. #[tokio::test] async fn new_rejects_loopback_endpoint_before_network_setup() { let conf = TierAliyun { diff --git a/crates/ecstore/src/services/tier/warm_backend_azure.rs b/crates/ecstore/src/services/tier/warm_backend_azure.rs index 997492ffb..7b3723eac 100644 --- a/crates/ecstore/src/services/tier/warm_backend_azure.rs +++ b/crates/ecstore/src/services/tier/warm_backend_azure.rs @@ -19,77 +19,38 @@ #![allow(clippy::all)] use std::collections::HashMap; -use std::sync::Arc; use crate::services::tier::{ tier_config::TierAzure, - warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options}, + warm_backend::{ + S3CompatibleWarmBackendParams, WarmBackend, WarmBackendGetOpts, build_transition_put_options, + new_s3_compatible_warm_backend, optimal_part_size, + }, warm_backend_s3::WarmBackendS3, }; -use rustfs_s3_client::{ - admin_handler_utils::AdminError, - api_put_object::PutObjectOptions, - credentials::{Credentials, SignatureType, Static, Value}, - transition_api::{BucketLookupType, Options, ReadCloser, ReaderImpl, TransitionClient, TransitionCore}, -}; +use rustfs_s3_client::transition_api::{BucketLookupType, ReadCloser, ReaderImpl}; use rustfs_utils::egress::validate_outbound_url; -use tracing::warn; -const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5; -const MAX_PARTS_COUNT: i64 = 10000; -const _MAX_PART_SIZE: i64 = 1024 * 1024 * 1024 * 5; const MIN_PART_SIZE: i64 = 1024 * 1024 * 128; pub struct WarmBackendAzure(WarmBackendS3); impl WarmBackendAzure { pub async fn new(conf: &TierAzure, tier: &str) -> Result { - if conf.access_key == "" || conf.secret_key == "" { - return Err(std::io::Error::other("both access and secret keys are required")); - } - - if conf.bucket == "" { - return Err(std::io::Error::other("no bucket name was provided")); - } - - let u = match url::Url::parse(&conf.endpoint) { - Ok(u) => u, - Err(e) => { - return Err(std::io::Error::other(e.to_string())); - } - }; - validate_outbound_url(&u).map_err(|err| std::io::Error::other(format!("tier endpoint is not allowed: {err}")))?; - - let creds = Credentials::new(Static(Value { - access_key_id: conf.access_key.clone(), - secret_access_key: conf.secret_key.clone(), - session_token: "".to_string(), - signer_type: SignatureType::SignatureV4, - ..Default::default() - })); - let opts = Options { - creds, - secure: u.scheme() == "https", - region: conf.region.clone(), - bucket_lookup: BucketLookupType::BucketLookupDNS, - ..Default::default() - }; - let scheme = u.scheme(); - let default_port = if scheme == "https" { 443 } else { 80 }; - let host = u - .host_str() - .ok_or_else(|| std::io::Error::other("Invalid endpoint URL: missing host"))?; - let client = TransitionClient::new(&format!("{}:{}", host, u.port().unwrap_or(default_port)), opts, "azure").await?; - - let client = Arc::new(client); - let core = TransitionCore(Arc::clone(&client)); - Ok(Self(WarmBackendS3 { - client, - core, - bucket: conf.bucket.clone(), - prefix: conf.prefix.strip_suffix("/").unwrap_or(&conf.prefix).to_owned(), - storage_class: "".to_string(), - })) + Ok(Self( + new_s3_compatible_warm_backend(S3CompatibleWarmBackendParams { + endpoint: &conf.endpoint, + access_key: &conf.access_key, + secret_key: &conf.secret_key, + bucket: &conf.bucket, + prefix: &conf.prefix, + region: &conf.region, + bucket_lookup: BucketLookupType::BucketLookupDNS, + provider_tag: "azure", + validate_endpoint: validate_outbound_url, + }) + .await?, + )) } } @@ -102,7 +63,7 @@ impl WarmBackend for WarmBackendAzure { length: i64, meta: HashMap, ) -> Result { - let part_size = optimal_part_size(length)?; + let part_size = optimal_part_size(length, MIN_PART_SIZE)?; let client = self.0.client.clone(); let res = client .put_object(&self.0.bucket, &self.0.get_dest(object), r, length, &{ @@ -133,32 +94,15 @@ impl WarmBackend for WarmBackendAzure { } } -fn optimal_part_size(object_size: i64) -> Result { - let mut object_size = object_size; - if object_size == -1 { - object_size = MAX_MULTIPART_PUT_OBJECT_SIZE; - } - - if object_size > MAX_MULTIPART_PUT_OBJECT_SIZE { - return Err(std::io::Error::other("entity too large")); - } - - let configured_part_size = MIN_PART_SIZE; - let mut part_size_flt = object_size as f64 / MAX_PARTS_COUNT as f64; - part_size_flt = (part_size_flt as f64 / configured_part_size as f64).ceil() * configured_part_size as f64; - - let part_size = part_size_flt as i64; - if part_size == 0 { - return Ok(MIN_PART_SIZE); - } - Ok(part_size) -} - #[cfg(test)] mod tests { use super::*; use crate::services::tier::tier_config::TierAzure; + /// The SSRF guard itself is exercised once, generically, in + /// `warm_backend::tests` (see backlog#2040/backlog#2041 and + /// rustfs/rustfs#6764) — this test only pins that this provider's + /// production constructor really is wired through that shared path. #[tokio::test] async fn new_rejects_loopback_endpoint_before_network_setup() { let conf = TierAzure { diff --git a/scripts/error-other-format-baseline.txt b/scripts/error-other-format-baseline.txt index 328e855c3..3ef54c8ce 100644 --- a/scripts/error-other-format-baseline.txt +++ b/scripts/error-other-format-baseline.txt @@ -54,8 +54,7 @@ 19|crates/ecstore/src/services/rebalance/worker.rs 33|crates/ecstore/src/services/tier/tier.rs 1|crates/ecstore/src/services/tier/tier_config.rs -1|crates/ecstore/src/services/tier/warm_backend_aliyun.rs -1|crates/ecstore/src/services/tier/warm_backend_azure.rs +1|crates/ecstore/src/services/tier/warm_backend.rs 2|crates/ecstore/src/services/tier/warm_backend_gcs.rs 1|crates/ecstore/src/services/tier/warm_backend_huaweicloud.rs 1|crates/ecstore/src/services/tier/warm_backend_minio.rs