fix(tier): switch outbound URL check to the operator-overridable policy

The initial fix (routing all warm-tier constructors through
validate_outbound_url) rejected the hermetic reliant::tiering e2e suite's
real hot->cold connection over 127.0.0.1, since two embedded RustFS
servers in that suite talk to each other over loopback by design.

validate_outbound_url has no override; OutboundPolicy (already used by
webhook targets and OIDC discovery URLs) enforces the identical default
restrictions but lets an operator allowlist one exact origin via
RUSTFS_OUTBOUND_ALLOW_ORIGINS -- metadata, link-local, and unspecified
addresses can never be allowlisted, so this does not reopen the SSRF
gap the previous commit closed. Switch every warm-tier constructor
(including S3, which folds Wasabi in via new_with_bucket_lookup) to
this policy through one shared crates/ecstore/src/services/tier/
warm_backend.rs::validate_tier_endpoint_url helper, replacing the nine
scattered validate_outbound_url call sites the previous commit added
and consolidating their error(format!) ratchet accounting into one
file.

Update the e2e suite to set RUSTFS_OUTBOUND_ALLOW_ORIGINS to the cold
node's real origin before starting/restarting hot, via a hot_env_for_tier
helper, and fix the resulting borrow-checker conflict in the one test
that stops cold mid-test by cloning its origin into an owned String
first. Also retarget a WarmBackendRustFS unit test that asserted on a
now-unreachable local host-missing message: the shared policy's
http(s)-only scheme check runs first and is now what actually rejects
that fixture's non-http endpoint.

Impact: operators with an existing self-hosted RustFS/MinIO/etc. tier
whose endpoint is a bare loopback/private/link-local IP literal (not a
hostname) need RUSTFS_OUTBOUND_ALLOW_ORIGINS=<origin> set and the
server restarted to keep that tier working after this change.

crates/ecstore/src/services/tier/warm_backend.rs

@@ -41,6 +41,7 @@ use rustfs_s3_client::{
     api_put_object::{AdvancedPutOptions, PutObjectOptions},
     transition_api::{ReadCloser, ReaderImpl},
 };
+use rustfs_utils::egress::OutboundPolicy;
 use rustfs_utils::http::headers::{
     CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, EXPIRES, HeaderExt as _,
 };
@@ -53,11 +54,31 @@ use std::collections::HashMap;
 use time::OffsetDateTime;
 use time::format_description::well_known::{Rfc2822, Rfc3339};
 use tracing::{info, warn};
+use url::Url;
 
 pub type WarmBackendImpl = Box<dyn WarmBackend + Send + Sync + 'static>;
 
 const PROBE_OBJECT: &str = "probeobject";
 
+/// Validates a warm-tier `endpoint` URL against the shared outbound policy
+/// (crates/utils/src/egress.rs), the same fail-closed check already applied to
+/// webhook targets and OIDC discovery URLs. By default this rejects loopback,
+/// RFC1918/link-local, and known cloud metadata-service hosts; an operator can
+/// allowlist one exact self-hosted origin via `RUSTFS_OUTBOUND_ALLOW_ORIGINS`
+/// (metadata, link-local, and unspecified addresses can never be allowlisted).
+///
+/// Every `WarmBackendXxx::new` constructor must call this immediately after
+/// parsing `conf.endpoint` and before building any credentials or network
+/// client, so a tier endpoint can never reach the network unvalidated
+/// regardless of provider (rustfs/backlog#2039 adversarial-review finding).
+pub(crate) fn validate_tier_endpoint_url(url: &Url) -> Result<(), std::io::Error> {
+    let policy =
+        OutboundPolicy::from_env_cached().map_err(|err| std::io::Error::other(format!("invalid outbound policy: {err}")))?;
+    policy
+        .validate_url(url)
+        .map_err(|err| std::io::Error::other(format!("tier endpoint is not allowed: {err}")))
+}
+
 #[derive(Default)]
 pub struct WarmBackendGetOpts {
     pub start_offset: i64,

crates/ecstore/src/services/tier/warm_backend_aliyun.rs

@@ -23,7 +23,7 @@ use std::sync::Arc;
 
 use crate::services::tier::{
     tier_config::TierAliyun,
-    warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options},
+    warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options, validate_tier_endpoint_url},
     warm_backend_s3::WarmBackendS3,
 };
 use rustfs_s3_client::{
@@ -32,7 +32,6 @@ use rustfs_s3_client::{
     credentials::{Credentials, SignatureType, Static, Value},
     transition_api::{BucketLookupType, Options, ReadCloser, ReaderImpl, TransitionClient, TransitionCore},
 };
-use rustfs_utils::egress::validate_outbound_url;
 use tracing::warn;
 
 const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5;
@@ -58,7 +57,7 @@ impl WarmBackendAliyun {
                 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}")))?;
+        validate_tier_endpoint_url(&u)?;
 
         let creds = Credentials::new(Static(Value {
             access_key_id: conf.access_key.clone(),

crates/ecstore/src/services/tier/warm_backend_azure.rs

@@ -23,7 +23,7 @@ use std::sync::Arc;
 
 use crate::services::tier::{
     tier_config::TierAzure,
-    warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options},
+    warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options, validate_tier_endpoint_url},
     warm_backend_s3::WarmBackendS3,
 };
 use rustfs_s3_client::{
@@ -32,7 +32,6 @@ use rustfs_s3_client::{
     credentials::{Credentials, SignatureType, Static, Value},
     transition_api::{BucketLookupType, Options, ReadCloser, ReaderImpl, TransitionClient, TransitionCore},
 };
-use rustfs_utils::egress::validate_outbound_url;
 use tracing::warn;
 
 const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5;
@@ -58,7 +57,7 @@ impl WarmBackendAzure {
                 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}")))?;
+        validate_tier_endpoint_url(&u)?;
 
         let creds = Credentials::new(Static(Value {
             access_key_id: conf.access_key.clone(),

crates/ecstore/src/services/tier/warm_backend_gcs.rs

@@ -32,14 +32,13 @@ use std::convert::TryFrom;
 
 use crate::services::tier::{
     tier_config::TierGCS,
-    warm_backend::{WarmBackend, WarmBackendGetOpts},
+    warm_backend::{WarmBackend, WarmBackendGetOpts, validate_tier_endpoint_url},
 };
 use rustfs_s3_client::{
     admin_handler_utils::AdminError,
     api_put_object::PutObjectOptions,
     transition_api::{Options, ReadCloser, ReaderImpl},
 };
-use rustfs_utils::egress::validate_outbound_url;
 use tracing::warn;
 
 const _MAX_PART_SIZE: i64 = 1024 * 1024 * 1024 * 5;
@@ -76,8 +75,7 @@ impl WarmBackendGCS {
 
         if !conf.endpoint.is_empty() {
             let endpoint_url = url::Url::parse(&conf.endpoint).map_err(|e| std::io::Error::other(e.to_string()))?;
-            validate_outbound_url(&endpoint_url)
-                .map_err(|err| std::io::Error::other(format!("tier endpoint is not allowed: {err}")))?;
+            validate_tier_endpoint_url(&endpoint_url)?;
         }
 
         let authorized_user = serde_json::from_str(&conf.creds)?;

crates/ecstore/src/services/tier/warm_backend_huaweicloud.rs

@@ -23,7 +23,7 @@ use std::sync::Arc;
 
 use crate::services::tier::{
     tier_config::TierHuaweicloud,
-    warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options},
+    warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options, validate_tier_endpoint_url},
     warm_backend_s3::WarmBackendS3,
 };
 use rustfs_s3_client::{
@@ -32,7 +32,6 @@ use rustfs_s3_client::{
     credentials::{Credentials, SignatureType, Static, Value},
     transition_api::{BucketLookupType, Options, ReadCloser, ReaderImpl, TransitionClient, TransitionCore},
 };
-use rustfs_utils::egress::validate_outbound_url;
 use tracing::warn;
 
 const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5;
@@ -58,7 +57,7 @@ impl WarmBackendHuaweicloud {
                 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}")))?;
+        validate_tier_endpoint_url(&u)?;
 
         let creds = Credentials::new(Static(Value {
             access_key_id: conf.access_key.clone(),

crates/ecstore/src/services/tier/warm_backend_minio.rs

@@ -23,7 +23,9 @@ use std::sync::Arc;
 
 use crate::services::tier::{
     tier_config::TierMinIO,
-    warm_backend::{TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options},
+    warm_backend::{
+        TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options, validate_tier_endpoint_url,
+    },
     warm_backend_s3::WarmBackendS3,
 };
 use rustfs_s3_client::{
@@ -32,7 +34,6 @@ use rustfs_s3_client::{
     credentials::{Credentials, SignatureType, Static, Value},
     transition_api::{Options, ReadCloser, ReaderImpl, TransitionClient, TransitionCore},
 };
-use rustfs_utils::egress::validate_outbound_url;
 use tracing::warn;
 
 const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5;
@@ -58,7 +59,7 @@ impl WarmBackendMinIO {
                 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}")))?;
+        validate_tier_endpoint_url(&u)?;
 
         let creds = Credentials::new(Static(Value {
             access_key_id: conf.access_key.clone(),

crates/ecstore/src/services/tier/warm_backend_r2.rs

@@ -23,7 +23,9 @@ use std::sync::Arc;
 
 use crate::services::tier::{
     tier_config::TierR2,
-    warm_backend::{TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options},
+    warm_backend::{
+        TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options, validate_tier_endpoint_url,
+    },
     warm_backend_s3::WarmBackendS3,
 };
 use rustfs_s3_client::{
@@ -32,7 +34,6 @@ use rustfs_s3_client::{
     credentials::{Credentials, SignatureType, Static, Value},
     transition_api::{Options, ReadCloser, ReaderImpl, TransitionClient, TransitionCore},
 };
-use rustfs_utils::egress::validate_outbound_url;
 use tracing::warn;
 
 const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5;
@@ -58,7 +59,7 @@ impl WarmBackendR2 {
                 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}")))?;
+        validate_tier_endpoint_url(&u)?;
 
         let creds = Credentials::new(Static(Value {
             access_key_id: conf.access_key.clone(),

crates/ecstore/src/services/tier/warm_backend_rustfs.rs

@@ -23,7 +23,9 @@ use std::sync::Arc;
 
 use crate::services::tier::{
     tier_config::TierRustFS,
-    warm_backend::{TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options},
+    warm_backend::{
+        TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options, validate_tier_endpoint_url,
+    },
     warm_backend_s3::WarmBackendS3,
 };
 use rustfs_s3_client::{
@@ -32,7 +34,6 @@ use rustfs_s3_client::{
     credentials::{Credentials, SignatureType, Static, Value},
     transition_api::{Options, ReadCloser, ReaderImpl, TransitionClient, TransitionCore},
 };
-use rustfs_utils::egress::validate_outbound_url;
 
 const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5;
 const MAX_PARTS_COUNT: i64 = 10000;
@@ -55,7 +56,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_tier_endpoint_url(&u)?;
 
         let creds = Credentials::new(Static(Value {
             access_key_id: conf.access_key.clone(),
@@ -186,18 +187,25 @@ mod tests {
         }
     }
 
+    // `validate_tier_endpoint_url` now runs before this constructor's own
+    // `u.host_str()` check, and it only accepts a parsed URL once its scheme
+    // is http(s) and its host is non-empty (http/https can never parse with a
+    // missing host per the WHATWG URL spec), so that local host check is
+    // unreachable in practice. This regression instead pins the shared policy
+    // rejecting a non-http(s) endpoint scheme, which is the case that now
+    // actually exercises this path first.
     #[tokio::test]
-    async fn new_returns_error_when_endpoint_has_no_host() {
+    async fn new_rejects_endpoint_with_disallowed_scheme() {
         let conf = rustfs_tier("rustfs://");
 
         let outcome = AssertUnwindSafe(WarmBackendRustFS::new(&conf, "tier")).catch_unwind().await;
 
         let result = outcome.expect("initialization should return an error instead of panicking");
         let err = match result {
-            Ok(_) => panic!("endpoint without host must be rejected"),
+            Ok(_) => panic!("endpoint with a non-http(s) scheme must be rejected"),
             Err(err) => err,
         };
-        assert!(err.to_string().contains("host"), "expected host validation error, got: {err}");
+        assert!(err.to_string().contains("scheme"), "expected scheme validation error, got: {err}");
     }
 
     #[tokio::test]

crates/ecstore/src/services/tier/warm_backend_s3.rs

@@ -26,7 +26,7 @@ use crate::services::tier::{
     tier_config::TierS3,
     warm_backend::{
         TransitionCandidateIdentity, TransitionCandidateProbe, TransitionCandidateReconciler, WarmBackend, WarmBackendGetOpts,
-        build_transition_put_options,
+        build_transition_put_options, validate_tier_endpoint_url,
     },
 };
 use http::HeaderMap;
@@ -41,7 +41,6 @@ use rustfs_s3_client::{
     transition_api::{BucketLookupType, Options, TransitionClient, TransitionCore},
     transition_api::{ReadCloser, ReaderImpl},
 };
-use rustfs_utils::egress::validate_outbound_url;
 use rustfs_utils::path::SLASH_SEPARATOR;
 use s3s::dto::BucketVersioningStatus;
 
@@ -90,7 +89,7 @@ impl WarmBackendS3 {
                 return Err(std::io::Error::other(err.to_string()));
             }
         };
-        validate_outbound_url(&u).map_err(|err| std::io::Error::other(format!("tier endpoint is not allowed: {err}")))?;
+        validate_tier_endpoint_url(&u)?;
 
         if conf.aws_role_web_identity_token_file == "" && conf.aws_role_arn != ""
             || conf.aws_role_web_identity_token_file != "" && conf.aws_role_arn == ""

crates/ecstore/src/services/tier/warm_backend_tencent.rs

@@ -23,7 +23,7 @@ use std::sync::Arc;
 
 use crate::services::tier::{
     tier_config::TierTencent,
-    warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options},
+    warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options, validate_tier_endpoint_url},
     warm_backend_s3::WarmBackendS3,
 };
 use rustfs_s3_client::{
@@ -32,7 +32,6 @@ use rustfs_s3_client::{
     credentials::{Credentials, SignatureType, Static, Value},
     transition_api::{BucketLookupType, Options, ReadCloser, ReaderImpl, TransitionClient, TransitionCore},
 };
-use rustfs_utils::egress::validate_outbound_url;
 use tracing::warn;
 
 const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5;
@@ -58,7 +57,7 @@ impl WarmBackendTencent {
                 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}")))?;
+        validate_tier_endpoint_url(&u)?;
 
         let creds = Credentials::new(Static(Value {
             access_key_id: conf.access_key.clone(),