From 876f60c1f415469ab9a0d686dea20703375aa5c6 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 28 Aug 2026 12:53:52 +0800 Subject: [PATCH] fix(ci): restore tier e2e and locked builds (#6773) * fix(tier): restore loopback e2e coverage safely * fix(build): sync scanner dev dependency lock --- Cargo.lock | 1 + crates/e2e_test/src/reliant/tiering.rs | 70 ++++++++++--------- .../src/services/tier/warm_backend_rustfs.rs | 33 ++++++++- 3 files changed, 69 insertions(+), 35 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 464291ae9..af2a80f9e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10611,6 +10611,7 @@ dependencies = [ "s3s", "serde", "serde_json", + "serial_test", "sha2 0.11.0", "temp-env", "tempfile", diff --git a/crates/e2e_test/src/reliant/tiering.rs b/crates/e2e_test/src/reliant/tiering.rs index 0557dfac2..216072883 100644 --- a/crates/e2e_test/src/reliant/tiering.rs +++ b/crates/e2e_test/src/reliant/tiering.rs @@ -23,9 +23,9 @@ //! //! There are no containers, no external S3 backend and no `awscurl`: the //! `AddTier` admin call is signed in-process with `rustfs_signer`, exactly like -//! the other admin-API e2e suites in this crate. The RustFS warm backend has no -//! loopback/SSRF restriction (that guard is replication-only), so `hot` can tier -//! to `cold` over `http://127.0.0.1:`. +//! the other admin-API e2e suites in this crate. The source server uses the +//! explicit test-only loopback opt-in to tier to `cold` over +//! `http://127.0.0.1:` while production keeps the SSRF guard enabled. //! //! The hermetic tests drive the transition and restore paths and pin the //! chains required by ilm-7 and the restore follow-up: @@ -96,6 +96,7 @@ const MANUAL_ACTIVE_CANCEL_OBJECTS: usize = 512; const MANUAL_RESTART_CANCEL_OBJECTS: usize = 512; const MANUAL_ACTIVE_CANCEL_RUNNING_TIMEOUT: StdDuration = StdDuration::from_secs(15); const MANUAL_TRANSITION_CANCEL_BARRIER_ENV: &str = "RUSTFS_E2E_MANUAL_TRANSITION_CANCEL_BARRIER"; +const ALLOW_LOOPBACK_TIER_ENDPOINT_ENV: (&str, &str) = ("RUSTFS_TIER_RUSTFS_ALLOW_LOOPBACK_ENDPOINT", "true"); const MANUAL_ASYNC_CONFLICT_TERMINAL_TIMEOUT: StdDuration = StdDuration::from_secs(90); const MANUAL_RESTART_RECOVERY_TIMEOUT: StdDuration = StdDuration::from_secs(80); const OBJECT_KEY: &str = "tier/鲁A12345/report.bin"; @@ -112,6 +113,20 @@ const HDR_SOURCE_REPLICATION_REQUEST: &str = "x-rustfs-source-replication-reques const HDR_SOURCE_MTIME: &str = "x-rustfs-source-mtime"; const TIER_MUTATION_RECOVERY_CHANGED: &str = "Remote tier mutation recovery changed before publish"; +async fn start_tier_source(hot: &mut RustFSTestEnvironment, extra_env: &[(&str, &str)]) -> TestResult { + let mut env = Vec::with_capacity(extra_env.len() + 1); + env.push(ALLOW_LOOPBACK_TIER_ENDPOINT_ENV); + env.extend_from_slice(extra_env); + hot.start_rustfs_server_with_env(vec![], &env).await +} + +async fn restart_tier_source(hot: &mut RustFSTestEnvironment, extra_env: &[(&str, &str)]) -> TestResult { + let mut env = Vec::with_capacity(extra_env.len() + 1); + env.push(ALLOW_LOOPBACK_TIER_ENDPOINT_ENV); + env.extend_from_slice(extra_env); + hot.restart_server_preserving_data(vec![], &env).await +} + /// 5 MiB — the S3 minimum size for a non-final multipart part; the object's only /// internal part boundary sits at this offset. const PART0_SIZE: usize = 5 * 1024 * 1024; @@ -860,8 +875,7 @@ async fn test_hermetic_transition_main_path() -> TestResult { // Hot/source server. A 1s scanner cycle is a backstop; transition is // primarily driven immediately by the multipart completion path. let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_CYCLE", "1")]) - .await?; + start_tier_source(&mut hot, &[("RUSTFS_SCANNER_CYCLE", "1")]).await?; let hot_client = hot.create_s3_client(); // Wire the RustFS remote tier (real connectivity probe, no force). @@ -959,8 +973,7 @@ async fn test_hermetic_transition_restore_failure_expiry_and_retry() -> TestResu cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_CYCLE", "1"), ("RUSTFS_ILM_DEBUG_DAY_SECS", "5")]) - .await?; + start_tier_source(&mut hot, &[("RUSTFS_SCANNER_CYCLE", "1"), ("RUSTFS_ILM_DEBUG_DAY_SECS", "5")]).await?; let hot_client = hot.create_s3_client(); add_rustfs_tier(&hot, &cold).await?; @@ -1088,8 +1101,7 @@ async fn test_manual_transition_run_black_box_semantics() -> TestResult { cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) - .await?; + start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?; let hot_client = hot.create_s3_client(); add_rustfs_tier(&hot, &cold).await?; let due_mtime = OffsetDateTime::now_utc() - time::Duration::hours(25); @@ -1194,8 +1206,7 @@ async fn test_manual_transition_async_job_status_polling() -> TestResult { cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) - .await?; + start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?; let hot_client = hot.create_s3_client(); add_rustfs_tier(&hot, &cold).await?; @@ -1293,8 +1304,7 @@ async fn test_manual_transition_async_limit_reports_terminal_partial() -> TestRe cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) - .await?; + start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?; let hot_client = hot.create_s3_client(); add_rustfs_tier(&hot, &cold).await?; @@ -1455,8 +1465,8 @@ async fn test_manual_transition_async_scope_conflicts_report_active_job() -> Tes cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env( - vec![], + start_tier_source( + &mut hot, &[ ("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600"), @@ -1565,8 +1575,7 @@ async fn test_manual_transition_async_different_buckets_admit_concurrently() -> cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) - .await?; + start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?; let hot_client = hot.create_s3_client(); add_rustfs_tier(&hot, &cold).await?; @@ -1686,8 +1695,7 @@ async fn test_manual_transition_async_tier_failure_reports_terminal_partial() -> cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) - .await?; + start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?; let hot_client = hot.create_s3_client(); add_rustfs_tier(&hot, &cold).await?; @@ -1779,8 +1787,7 @@ async fn test_manual_transition_async_worker_failure_reports_terminal_partial() cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) - .await?; + start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?; let hot_client = hot.create_s3_client(); add_rustfs_tier(&hot, &cold).await?; cold.stop_server(); @@ -1872,8 +1879,8 @@ async fn test_manual_transition_async_active_cancel_reports_terminal_cancelled() cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env( - vec![], + start_tier_source( + &mut hot, &[ ("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600"), @@ -1978,7 +1985,7 @@ async fn test_manual_transition_async_cancel_after_process_restart_recovers_term ("RUSTFS_TRANSITION_QUEUE_CAPACITY", "512"), ]; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &restart_env).await?; + start_tier_source(&mut hot, &restart_env).await?; let hot_client = hot.create_s3_client(); add_rustfs_tier(&hot, &cold).await?; @@ -2006,7 +2013,7 @@ async fn test_manual_transition_async_cancel_after_process_restart_recovers_term .ok_or("async response must include status_endpoint")?; assert_eq!(accepted.cancel_endpoint.as_deref(), Some(status_endpoint)); - hot.restart_server_preserving_data(vec![], &restart_env).await?; + restart_tier_source(&mut hot, &restart_env).await?; let restarted = manual_transition_job_status(&hot, status_endpoint).await?; assert_eq!(restarted.job_id, job_id); @@ -2130,8 +2137,7 @@ async fn test_manual_transition_run_contract_no_status_cancel_fields() -> TestRe cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) - .await?; + start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?; let hot_client = hot.create_s3_client(); add_rustfs_tier(&hot, &cold).await?; @@ -2169,8 +2175,7 @@ async fn test_manual_transition_run_continuation_token_resumes_without_raw_marke cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) - .await?; + start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?; let hot_client = hot.create_s3_client(); add_rustfs_tier(&hot, &cold).await?; @@ -2210,8 +2215,7 @@ async fn test_manual_transition_run_continuation_token_resumes_without_raw_marke "continuation token must not expose the raw object prefix: {continuation}" ); - hot.restart_server_preserving_data(vec![], &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]) - .await?; + restart_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?; let second = manual_transition_run_with_max_and_continuation( &hot, @@ -2244,8 +2248,8 @@ async fn test_manual_transition_run_queue_pressure_partial() -> TestResult { cold_client.create_bucket().bucket(TIER_BUCKET).send().await?; let mut hot = RustFSTestEnvironment::new().await?; - hot.start_rustfs_server_with_env( - vec![], + start_tier_source( + &mut hot, &[ ("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600"), diff --git a/crates/ecstore/src/services/tier/warm_backend_rustfs.rs b/crates/ecstore/src/services/tier/warm_backend_rustfs.rs index e11377ea6..1f196718b 100644 --- a/crates/ecstore/src/services/tier/warm_backend_rustfs.rs +++ b/crates/ecstore/src/services/tier/warm_backend_rustfs.rs @@ -32,12 +32,32 @@ use rustfs_s3_client::{ credentials::{Credentials, SignatureType, Static, Value}, transition_api::{Options, ReadCloser, ReaderImpl, TransitionClient, TransitionCore}, }; -use rustfs_utils::egress::validate_outbound_url; +use rustfs_utils::egress::{OutboundUrlError, validate_outbound_url}; 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; +// Debug-only opt-in for single-host test/dev setups; release builds always reject loopback. +const ALLOW_LOOPBACK_TIER_ENDPOINT_ENV: &str = "RUSTFS_TIER_RUSTFS_ALLOW_LOOPBACK_ENDPOINT"; + +fn validate_rustfs_tier_endpoint(url: &url::Url) -> Result<(), OutboundUrlError> { + let allow_loopback = cfg!(debug_assertions) + && std::env::var(ALLOW_LOOPBACK_TIER_ENDPOINT_ENV) + .map(|value| value == "1" || value.eq_ignore_ascii_case("true")) + .unwrap_or(false); + validate_rustfs_tier_endpoint_inner(url, allow_loopback) +} + +fn validate_rustfs_tier_endpoint_inner(url: &url::Url, allow_loopback: bool) -> Result<(), OutboundUrlError> { + match validate_outbound_url(url) { + Err(OutboundUrlError::ForbiddenHost { + reason: "loopback address" | "loopback host", + .. + }) if allow_loopback => Ok(()), + result => result, + } +} pub struct WarmBackendRustFS(WarmBackendS3); @@ -55,7 +75,7 @@ impl WarmBackendRustFS { Ok(u) => u, Err(e) => return Err(std::io::Error::other(e)), }; - validate_outbound_url(&u).map_err(|err| std::io::Error::other(format!("tier endpoint is not allowed: {err}")))?; + validate_rustfs_tier_endpoint(&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(), @@ -208,4 +228,13 @@ mod tests { Err(err) => assert!(err.to_string().contains("not allowed")), } } + + #[test] + fn loopback_opt_in_does_not_allow_other_private_endpoints() { + let loopback = url::Url::parse("https://127.0.0.1:9000").unwrap(); + assert!(validate_rustfs_tier_endpoint_inner(&loopback, true).is_ok()); + + let private = url::Url::parse("https://10.0.0.1:9000").unwrap(); + assert!(validate_rustfs_tier_endpoint_inner(&private, true).is_err()); + } }