gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-01 09:48:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(ecstore): migrate aliyun/azure warm backends to shared S3 constructor (#6774)

Browse Source
This commit is contained in:
Zhengchao An
2026-08-28 19:49:27 +08:00
committed by GitHub
parent 206ef7d086
commit 64705d7589
4 changed files with 74 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/services/tier/warm_backend.rs
+25 -8
View File
@@ -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<WarmBackendS3, std::io::Error> {
@@ -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<i64, std::io::Error> {
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"))
crates/ecstore/src/services/tier/warm_backend_aliyun.rs
+24 -80
View File
@@ -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<Self, std::io::Error> {
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<String, String>,
) -> Result<String, std::io::Error> {
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<i64, std::io::Error> {
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 {
crates/ecstore/src/services/tier/warm_backend_azure.rs
+24 -80
View File
@@ -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<Self, std::io::Error> {
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<String, String>,
) -> Result<String, std::io::Error> {
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<i64, std::io::Error> {
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 {
scripts/error-other-format-baseline.txt
+1 -2
View File
@@ -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