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

@@ -10611,6 +10611,7 @@ dependencies = [
  "s3s",
  "serde",
  "serde_json",
+ "serial_test",
  "sha2 0.11.0",
  "temp-env",
  "tempfile",

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
+//! the other admin-API e2e suites in this crate. The source server uses the
-//! loopback/SSRF restriction (that guard is replication-only), so `hot` can tier
+//! explicit test-only loopback opt-in to tier to `cold` over
-//! to `cold` over `http://127.0.0.1:<port>`.
+//! `http://127.0.0.1:<port>` 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")])
+    start_tier_source(&mut hot, &[("RUSTFS_SCANNER_CYCLE", "1")]).await?;
-        .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")])
+    start_tier_source(&mut hot, &[("RUSTFS_SCANNER_CYCLE", "1"), ("RUSTFS_ILM_DEBUG_DAY_SECS", "5")]).await?;
-        .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")])
+    start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?;
-        .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")])
+    start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?;
-        .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")])
+    start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?;
-        .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(
+    start_tier_source(
-        vec![],
+        &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")])
+    start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?;
-        .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")])
+    start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?;
-        .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")])
+    start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?;
-        .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(
+    start_tier_source(
-        vec![],
+        &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")])
+    start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?;
-        .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")])
+    start_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?;
-        .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")])
+    restart_tier_source(&mut hot, &[("RUSTFS_SCANNER_ENABLED", "false"), ("RUSTFS_SCANNER_CYCLE", "3600")]).await?;
-        .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(
+    start_tier_source(
-        vec![],
+        &mut hot,
         &[
             ("RUSTFS_SCANNER_ENABLED", "false"),
             ("RUSTFS_SCANNER_CYCLE", "3600"),

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());
+    }
 }