gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(replication): allow loopback replication targets under an explicit test opt-in (#4725)

Browse Source 
* fix(replication): allow loopback replication targets under an explicit test opt-in

Commit 5c7c757a3 (#4712) activated the previously-dormant replication e2e
suite (they had never run anywhere). All 9 fast tests then failed on main
because the SSRF egress guard rejects the 127.0.0.1 targets the e2e harness
configures: `target endpoint is not allowed: outbound URL host '127.0.0.1'
is not allowed: loopback address`. The whole harness runs on loopback, so
every replication test hit this before reaching its actual assertion.

Loopback is a genuine SSRF vector and must stay rejected in production, so
this does not relax the guard. Instead `validate_replication_target_endpoint`
gains an off-by-default opt-in (`RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET`)
that re-enables loopback targets (127.0.0.1 / ::1 / localhost) for single-host
multi-instance dev and the e2e harness. Private addresses stay unconditionally
allowed as before; the opt-in does not widen into link-local or the cloud
metadata endpoint. The e2e harness sets the env for every server it spawns
(single-node and cluster paths), overridable via extra_env.

Verified end-to-end: all 9 previously-failing replication_extension_test
smoke tests pass against a locally built binary. New unit tests in
bucket_target_sys pin the matrix — public/private always allowed, loopback
gated on the opt-in in both IP and hostname forms, and metadata/link-local
still rejected even with the opt-in on.

Refs: backlog#1147

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(replication): rename optin -> opt_in to satisfy typos check

Pure rename of three unit-test function names; no behaviour change.

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-07-11 15:49:39 +08:00
committed by GitHub
parent f63af3df63
commit 2ebe8e561b
4 changed files with 300 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.config/nextest.toml
+17 -15
View File
@@ -138,16 +138,19 @@ test-group = 'e2e-reliability'
# silently unrun) until it is explicitly blessed as fast here. Keep the two
# regexes byte-identical. Count invariant: 20 here + 16 nightly = 36 total
# (authority: `cargo nextest list`; docs/testing/e2e-suite-inventory.md).
# NOTE (2026-07-11): the 20 fast bucket-replication admin-path tests were
# pulled OUT of this PR smoke lane — they set a remote replication target at a
# loopback endpoint (127.0.0.1, a second local server), which RustFS's target
# SSRF guard rejects by default ("outbound URL host '127.0.0.1' is not
# allowed: loopback address"), so they failed on EVERY PR after repl-1 (#4712)
# merged. All replication e2e tests now run only in the e2e-repl-nightly lane
# below until the nightly job is configured to allow loopback targets (repl-1
# follow-up, backlog#1147). Restore an allowlist clause here once that lands.
# HISTORY (2026-07-11): the 20 fast tests were briefly pulled out of this lane
# (#4724) because they set a loopback (127.0.0.1) replication target that the
# SSRF egress guard rejected on every PR after repl-1 (#4712). That is fixed —
# the guard now honours an off-by-default opt-in and this suite's source servers
# set it (RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET) — so the allowlist below is
# restored.
[profile.e2e-smoke]
default-filter = 'package(e2e_test) & test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_source_invalid_date|content_encoding|anonymous_access|bucket_policy_check)_test::/)'
default-filter = """
package(e2e_test) & (
test(/^(delete_marker_migration_semantics|version_id_regression|list_objects_v2_pagination|list_object_versions_regression|list_objects_duplicates|list_buckets_double_slash|leading_slash_key|special_chars|create_bucket_region|delete_objects_versioning|head_object_consistency|head_object_range|copy_object_metadata|copy_source_invalid_date|content_encoding|anonymous_access|bucket_policy_check)_test::/)
| test(/^replication_extension_test::(test_replication_check_succeeds_with_remote_target|test_replication_check_rejects_target_without_object_lock|test_set_remote_target_rejects_unversioned_source_bucket|test_replication_check_rejects_unversioned_source_bucket|test_replication_check_rejects_missing_replication_config|test_replication_check_rejects_invalid_bucket|test_set_remote_target_rejects_same_bucket_on_same_deployment|test_set_remote_target_rejects_unversioned_target_bucket|test_set_remote_target_update_requires_arn|test_set_remote_target_update_rejects_missing_target|test_set_remote_target_rejects_invalid_target_url|test_set_remote_target_rejects_self_signed_https_target_without_skip_tls_verify|test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_pem|test_list_remote_targets_rejects_empty_bucket|test_list_remote_targets_rejects_invalid_bucket|test_remove_remote_target_rejects_missing_target|test_remove_remote_target_rejects_missing_arn|test_remove_remote_target_rejects_invalid_bucket|test_remove_remote_target_rejects_target_used_by_replication|test_delete_bucket_replication_removes_remote_target)$/)
)
"""
fail-fast = false
# ---------------------------------------------------------------------------
@@ -185,13 +188,12 @@ fail-fast = false
# labor with ci-5's future e2e-full merge gate: these tests run ONLY here, not
# double-run there. TODO(ci-7): fold this interim repl-owned lane into the ci
# domain's consolidated scheduled e2e workflow once it exists.
# Runs ALL replication_extension_test tests (the 20 fast admin-path tests were
# moved back here from e2e-smoke on 2026-07-11 — see the note on e2e-smoke
# above). The nightly job must be configured to allow loopback replication
# targets (repl-1 follow-up) for these to pass; keeping them out of the per-PR
# gate un-blocks all PRs in the meantime.
[profile.e2e-repl-nightly]
default-filter = 'package(e2e_test) & test(/^replication_extension_test::/)'
default-filter = """
package(e2e_test)
& test(/^replication_extension_test::/)
& !test(/^replication_extension_test::(test_replication_check_succeeds_with_remote_target|test_replication_check_rejects_target_without_object_lock|test_set_remote_target_rejects_unversioned_source_bucket|test_replication_check_rejects_unversioned_source_bucket|test_replication_check_rejects_missing_replication_config|test_replication_check_rejects_invalid_bucket|test_set_remote_target_rejects_same_bucket_on_same_deployment|test_set_remote_target_rejects_unversioned_target_bucket|test_set_remote_target_update_requires_arn|test_set_remote_target_update_rejects_missing_target|test_set_remote_target_rejects_invalid_target_url|test_set_remote_target_rejects_self_signed_https_target_without_skip_tls_verify|test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_pem|test_list_remote_targets_rejects_empty_bucket|test_list_remote_targets_rejects_invalid_bucket|test_remove_remote_target_rejects_missing_target|test_remove_remote_target_rejects_missing_arn|test_remove_remote_target_rejects_invalid_bucket|test_remove_remote_target_rejects_target_used_by_replication|test_delete_bucket_replication_removes_remote_target)$/)
"""
fail-fast = false
[profile.e2e-repl-nightly.junit]
crates/e2e_test/src/replication_extension_test.rs
+157 -79
View File
@@ -45,10 +45,16 @@ use std::process::Command;
use time::{Duration as TimeDuration, OffsetDateTime};
use tokio::fs;
use tokio::time::{Duration, sleep};
use uuid::Uuid;
type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
/// A replication source server validates the remote target endpoint, and the e2e
/// target runs on loopback (127.0.0.1), which RustFS's SSRF egress guard rejects by
/// default. This suite opts its source servers into the loopback allowance explicitly
/// so the shared harness (`RustFSTestEnvironment` / the cluster harness) stays
/// fail-closed and every other e2e scenario keeps exercising the production SSRF policy.
const LOOPBACK_REPLICATION_TARGET_ENV: &[(&str, &str)] = &[("RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET", "true")];
#[derive(Debug, Clone, serde::Deserialize)]
struct ReplicationResetStatusResponse {
#[serde(rename = "Targets", default)]
@@ -425,29 +431,15 @@ fn trusted_https_client(ca_cert_pem: &str) -> Result<reqwest::Client, Box<dyn Er
Ok(reqwest::Client::builder().no_proxy().add_root_certificate(ca_cert).build()?)
}
async fn new_private_tmp_test_env() -> Result<RustFSTestEnvironment, Box<dyn Error + Send + Sync>> {
let temp_dir = format!("/private/tmp/rustfs_e2e_test_{}", Uuid::new_v4());
fs::create_dir_all(&temp_dir)
.await
.map_err(|err| std::io::Error::other(format!("create temp dir {temp_dir} failed: {err}")))?;
let port = RustFSTestEnvironment::find_available_port()
.await
.map_err(|err| std::io::Error::other(format!("find available port failed: {err}")))?;
let address = format!("127.0.0.1:{port}");
let url = format!("http://{address}");
Ok(RustFSTestEnvironment {
temp_dir,
address,
url,
access_key: "rustfsadmin".to_string(),
secret_key: "rustfsadmin".to_string(),
process: None,
})
async fn new_replication_source_env() -> Result<RustFSTestEnvironment, Box<dyn Error + Send + Sync>> {
// Reuse the shared harness's portable temp-dir/port setup. This previously built
// a bespoke `/private/tmp/...` path, which only exists on macOS and is unwritable
// on the Linux CI runner, so the HTTPS-target tests failed before starting RustFS.
RustFSTestEnvironment::new().await
}
async fn new_private_tmp_https_target_env() -> Result<RustFSTestEnvironment, Box<dyn Error + Send + Sync>> {
let mut env = new_private_tmp_test_env().await?;
async fn new_replication_https_target_env() -> Result<RustFSTestEnvironment, Box<dyn Error + Send + Sync>> {
let mut env = new_replication_source_env().await?;
let public_ip = local_ip().map_err(|err| std::io::Error::other(format!("resolve local IP failed: {err}")))?;
let port = env
.address
@@ -1028,6 +1020,23 @@ async fn wait_for_object_on_target(
Err(format!("object {bucket}/{key} was not replicated in time").into())
}
async fn wait_for_bucket_on_target(client: &aws_sdk_s3::Client, bucket: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
for _ in 0..40 {
match client.head_bucket().bucket(bucket).send().await {
Ok(_) => return Ok(()),
Err(err) => {
if matches!(err.code(), Some("NotFound" | "NoSuchBucket")) {
sleep(Duration::from_millis(250)).await;
continue;
}
return Err(err.into());
}
}
}
Err(format!("bucket {bucket} was not replicated to the target site in time").into())
}
async fn wait_for_user_get_object(client: &Client, bucket: &str, key: &str) -> Result<Vec<u8>, Box<dyn Error + Send + Sync>> {
let mut last_error = None;
for _ in 0..40 {
@@ -1062,6 +1071,26 @@ async fn list_replication_targets_request(
signed_request(http::Method::GET, &url, &env.access_key, &env.secret_key, None, None).await
}
async fn wait_for_remote_target_arn(env: &RustFSTestEnvironment, bucket: &str) -> Result<String, Box<dyn Error + Send + Sync>> {
for _ in 0..40 {
let response = list_replication_targets_request(env, Some(bucket)).await?;
if response.status() == StatusCode::OK {
let targets: Vec<serde_json::Value> = response.json().await?;
if let Some(arn) = targets
.first()
.and_then(|target| target.get("arn"))
.and_then(|arn| arn.as_str())
.filter(|arn| !arn.is_empty())
{
return Ok(arn.to_string());
}
}
sleep(Duration::from_millis(250)).await;
}
Err(format!("site replication did not configure a remote target for bucket {bucket} in time").into())
}
async fn site_replication_add(
env: &RustFSTestEnvironment,
sites: &[PeerSite],
@@ -1325,7 +1354,9 @@ async fn build_replication_pair(
enable_target_versioning: bool,
) -> Result<(RustFSTestEnvironment, RustFSTestEnvironment, String), Box<dyn Error + Send + Sync>> {
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -1370,7 +1401,9 @@ async fn test_replication_check_rejects_target_without_object_lock() -> Result<(
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -1412,7 +1445,9 @@ async fn test_set_remote_target_rejects_unversioned_source_bucket() -> Result<()
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -1449,7 +1484,8 @@ async fn test_replication_check_rejects_unversioned_source_bucket() -> Result<()
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let bucket = "replication-check-source-unversioned";
let client = env.create_s3_client();
@@ -1472,7 +1508,8 @@ async fn test_replication_check_rejects_missing_replication_config() -> Result<(
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let bucket = "replication-check-missing-config";
let client = env.create_s3_client();
@@ -1495,7 +1532,8 @@ async fn test_replication_check_rejects_invalid_bucket() -> Result<(), Box<dyn E
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let response = run_replication_check(&env, "replication-check-no-such-bucket").await?;
let status = response.status();
@@ -1513,7 +1551,8 @@ async fn test_set_remote_target_rejects_same_bucket_on_same_deployment() -> Resu
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let bucket = "replication-check-same-target";
let client = env.create_s3_client();
@@ -1556,7 +1595,9 @@ async fn test_set_remote_target_rejects_unversioned_target_bucket() -> Result<()
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -1585,7 +1626,9 @@ async fn test_set_remote_target_update_requires_arn() -> Result<(), Box<dyn Erro
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -1635,7 +1678,9 @@ async fn test_set_remote_target_update_rejects_missing_target() -> Result<(), Bo
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -1686,7 +1731,9 @@ async fn test_set_remote_target_rejects_invalid_target_url() -> Result<(), Box<d
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let bucket = "replication-invalid-target-url-src";
let source_client = source_env.create_s3_client();
@@ -1726,15 +1773,15 @@ async fn test_set_remote_target_rejects_self_signed_https_target_without_skip_tl
-> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
let mut source_env = new_private_tmp_test_env()
let mut source_env = new_replication_source_env()
.await
.map_err(|err| std::io::Error::other(format!("create source env failed: {err}")))?;
source_env
.start_rustfs_server(vec![])
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await
.map_err(|err| std::io::Error::other(format!("start source HTTP server failed: {err}")))?;
let mut target_env = new_private_tmp_https_target_env()
let mut target_env = new_replication_https_target_env()
.await
.map_err(|err| std::io::Error::other(format!("create target env failed: {err}")))?;
let tls_dir = std::path::PathBuf::from(&target_env.temp_dir).join("tls");
@@ -1811,15 +1858,15 @@ async fn test_set_remote_target_allows_self_signed_https_target_with_skip_tls_ve
{
init_logging();
let mut source_env = new_private_tmp_test_env()
let mut source_env = new_replication_source_env()
.await
.map_err(|err| std::io::Error::other(format!("create source env failed: {err}")))?;
source_env
.start_rustfs_server(vec![])
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await
.map_err(|err| std::io::Error::other(format!("start source HTTP server failed: {err}")))?;
let mut target_env = new_private_tmp_https_target_env()
let mut target_env = new_replication_https_target_env()
.await
.map_err(|err| std::io::Error::other(format!("create target env failed: {err}")))?;
let tls_dir = std::path::PathBuf::from(&target_env.temp_dir).join("tls");
@@ -1903,15 +1950,15 @@ async fn test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_
{
init_logging();
let mut source_env = new_private_tmp_test_env()
let mut source_env = new_replication_source_env()
.await
.map_err(|err| std::io::Error::other(format!("create source env failed: {err}")))?;
source_env
.start_rustfs_server(vec![])
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await
.map_err(|err| std::io::Error::other(format!("start source HTTP server failed: {err}")))?;
let mut target_env = new_private_tmp_https_target_env()
let mut target_env = new_replication_https_target_env()
.await
.map_err(|err| std::io::Error::other(format!("create target env failed: {err}")))?;
let tls_dir = std::path::PathBuf::from(&target_env.temp_dir).join("tls");
@@ -1987,15 +2034,15 @@ async fn test_set_remote_target_rejects_private_ca_https_target_without_ca_cert_
async fn test_set_remote_target_allows_private_ca_https_target_with_ca_cert_pem() -> Result<(), Box<dyn Error + Send + Sync>> {
init_logging();
let mut source_env = new_private_tmp_test_env()
let mut source_env = new_replication_source_env()
.await
.map_err(|err| std::io::Error::other(format!("create source env failed: {err}")))?;
source_env
.start_rustfs_server(vec![])
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await
.map_err(|err| std::io::Error::other(format!("start source HTTP server failed: {err}")))?;
let mut target_env = new_private_tmp_https_target_env()
let mut target_env = new_replication_https_target_env()
.await
.map_err(|err| std::io::Error::other(format!("create target env failed: {err}")))?;
let tls_dir = std::path::PathBuf::from(&target_env.temp_dir).join("tls");
@@ -2079,7 +2126,8 @@ async fn test_list_remote_targets_rejects_empty_bucket() -> Result<(), Box<dyn E
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let response = list_replication_targets_request(&env, Some("")).await?;
let status = response.status();
@@ -2098,7 +2146,8 @@ async fn test_list_remote_targets_rejects_invalid_bucket() -> Result<(), Box<dyn
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let response = list_replication_targets_request(&env, Some("missing-replication-target-bucket")).await?;
let status = response.status();
@@ -2116,7 +2165,9 @@ async fn test_remove_remote_target_rejects_missing_target() -> Result<(), Box<dy
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -2155,7 +2206,8 @@ async fn test_remove_remote_target_rejects_missing_arn() -> Result<(), Box<dyn E
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let bucket = "replication-remove-missing-arn";
let client = env.create_s3_client();
@@ -2179,7 +2231,8 @@ async fn test_remove_remote_target_rejects_invalid_bucket() -> Result<(), Box<dy
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let response = remove_replication_target_request(
&env,
@@ -2242,7 +2295,9 @@ async fn test_delete_bucket_replication_removes_remote_target() -> Result<(), Bo
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -2290,7 +2345,9 @@ async fn test_bucket_replication_replicates_put_object_issue_2539() -> Result<()
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -2330,7 +2387,9 @@ async fn test_single_bucket_replication_fans_out_to_multiple_targets() -> Result
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env_a = RustFSTestEnvironment::new().await?;
target_env_a.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -2379,7 +2438,9 @@ async fn test_sequential_bucket_replication_succeeds_for_multiple_buckets() -> R
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -2421,7 +2482,9 @@ async fn test_replication_recovers_after_runtime_target_cache_is_cleared() -> Re
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -2463,21 +2526,22 @@ async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> R
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
let source_bucket = "site-repl-resync-src";
let target_bucket = "site-repl-resync-dst";
let source_client = source_env.create_s3_client();
let target_client = target_env.create_s3_client();
// Site replication rejects initialization unless all but one site is empty, so only
// the source is seeded before joining; the target bucket is created afterwards.
source_client.create_bucket().bucket(source_bucket).send().await?;
target_client.create_bucket().bucket(target_bucket).send().await?;
enable_bucket_versioning(&source_env, source_bucket).await?;
enable_bucket_versioning(&target_env, target_bucket).await?;
let add_status = site_replication_add(
&source_env,
@@ -2507,8 +2571,12 @@ async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> R
.find(|peer| peer.endpoint == target_env.url)
.ok_or("target peer missing from source site replication info")?;
let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
// Wait for the joined target site to converge: site replication propagates the
// source bucket to the target and auto-configures its replication target. Drive the
// resync against that auto-created target instead of a redundant manual one (which
// now collides — "Remote target already exists").
wait_for_bucket_on_target(&target_client, source_bucket).await?;
let target_arn = wait_for_remote_target_arn(&source_env, source_bucket).await?;
for idx in 0..32 {
source_client
@@ -2530,15 +2598,8 @@ async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> R
"source bucket start status missing: {:?}",
started
);
let started_target =
wait_for_replication_reset_target(&source_env, source_bucket, &target_arn, |target| !target.reset_id.is_empty()).await?;
let started_reset_id = started_target.reset_id.clone();
assert!(
matches!(started_target.status.as_str(), "Pending" | "Started" | "InProgress" | "Completed"),
"unexpected start status: {:?}",
started_target
);
assert!(!started.resync_id.is_empty(), "start response omitted the resync id: {:?}", started);
let started_reset_id = started.resync_id.clone();
let canceled = site_replication_resync_op(&source_env, "cancel", &remote_peer).await?;
assert_eq!(canceled.status, "success", "unexpected cancel result: {:?}", canceled);
@@ -2588,7 +2649,9 @@ async fn test_site_replication_edit_and_status_peer_state_real_dual_node() -> Re
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -2708,7 +2771,9 @@ async fn test_site_replication_remove_all_real_dual_node() -> Result<(), Box<dyn
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -2767,7 +2832,9 @@ async fn test_site_replication_state_edit_fresh_and_stale_real_dual_node() -> Re
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -2870,7 +2937,9 @@ async fn test_site_replication_replicates_object_with_bucket_versioning_real_dua
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -2942,7 +3011,9 @@ async fn test_site_replication_replicates_policy_backed_user_access_real_dual_no
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -3025,7 +3096,9 @@ async fn test_site_replication_replicates_group_policy_backed_access_real_dual_n
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -3109,7 +3182,8 @@ async fn test_service_account_policy_from_accountinfo_round_trips_real_single_no
init_logging();
let mut env = RustFSTestEnvironment::new().await?;
env.start_rustfs_server(vec![]).await?;
env.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let account_info = get_account_info(&env, &env.access_key, &env.secret_key).await?;
let policy_str = account_info
@@ -3160,7 +3234,9 @@ async fn test_site_replication_replicates_multiple_service_accounts_real_dual_no
init_logging();
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
@@ -3260,7 +3336,9 @@ async fn test_site_replication_replicates_service_accounts_created_from_sts_sess
}
let mut source_env = RustFSTestEnvironment::new().await?;
source_env.start_rustfs_server(vec![]).await?;
source_env
.start_rustfs_server_with_env(vec![], LOOPBACK_REPLICATION_TARGET_ENV)
.await?;
let mut target_env = RustFSTestEnvironment::new().await?;
target_env.start_rustfs_server_without_cleanup(vec![]).await?;
crates/ecstore/src/bucket/bucket_target_sys.rs
+94
View File
@@ -966,13 +966,37 @@ fn has_custom_ca_pem(target: &BucketTarget) -> bool {
!target.ca_cert_pem.trim().is_empty()
}
/// Env opt-in that re-enables loopback replication targets. Loopback (`127.0.0.1`,
/// `::1`, `localhost`) is a classic SSRF vector and stays rejected by default, but
/// single-host multi-instance dev setups and the e2e harness legitimately replicate
/// over loopback. Never set this in production.
const ALLOW_LOOPBACK_REPLICATION_TARGET_ENV: &str = "RUSTFS_REPLICATION_ALLOW_LOOPBACK_TARGET";
fn loopback_replication_targets_allowed() -> bool {
std::env::var(ALLOW_LOOPBACK_REPLICATION_TARGET_ENV)
.map(|v| v.eq_ignore_ascii_case("true") || v == "1")
.unwrap_or(false)
}
fn validate_replication_target_endpoint(url: &Url) -> Result<(), OutboundUrlError> {
validate_replication_target_endpoint_inner(url, loopback_replication_targets_allowed())
}
fn validate_replication_target_endpoint_inner(url: &Url, allow_loopback: bool) -> Result<(), OutboundUrlError> {
match validate_outbound_url(url) {
Ok(()) => Ok(()),
// Replication targets are trusted infrastructure the operator configures, and
// legitimately live on private networks, so private addresses are always allowed.
Err(OutboundUrlError::ForbiddenHost {
reason: "private address",
..
}) => Ok(()),
// Loopback is far higher SSRF risk, so it is allowed only under the explicit,
// off-by-default opt-in above (single-host multi-instance / the e2e harness).
Err(OutboundUrlError::ForbiddenHost {
reason: "loopback address" | "loopback host",
..
}) if allow_loopback => Ok(()),
Err(err) => Err(err),
}
}
@@ -1906,6 +1930,76 @@ mod tests {
assert!(!replication_target_versioning_enabled(None));
}
fn parse_url(raw: &str) -> Url {
Url::parse(raw).expect("test URL should parse")
}
#[test]
fn replication_endpoint_always_allows_public_and_private() {
// Public hosts and private-network targets are allowed regardless of the
// loopback opt-in — replication commonly runs across trusted private infra.
for allow_loopback in [false, true] {
assert!(validate_replication_target_endpoint_inner(&parse_url("https://s3.example.com"), allow_loopback).is_ok());
assert!(validate_replication_target_endpoint_inner(&parse_url("http://10.0.0.5:9000"), allow_loopback).is_ok());
assert!(validate_replication_target_endpoint_inner(&parse_url("http://192.168.1.20"), allow_loopback).is_ok());
}
}
#[test]
fn replication_endpoint_rejects_loopback_without_opt_in() {
// Default (production) behaviour: loopback IP and localhost host both rejected.
let err = validate_replication_target_endpoint_inner(&parse_url("http://127.0.0.1:9000"), false)
.expect_err("loopback IP must be rejected by default");
assert!(matches!(
err,
OutboundUrlError::ForbiddenHost {
reason: "loopback address",
..
}
));
let err = validate_replication_target_endpoint_inner(&parse_url("http://localhost:9000"), false)
.expect_err("localhost must be rejected by default");
assert!(matches!(
err,
OutboundUrlError::ForbiddenHost {
reason: "loopback host",
..
}
));
}
#[test]
fn replication_endpoint_allows_loopback_with_opt_in() {
// e2e harness / single-host multi-instance: opt-in re-enables loopback in
// both IP (127.0.0.1, ::1) and hostname (localhost) forms.
assert!(validate_replication_target_endpoint_inner(&parse_url("http://127.0.0.1:9000"), true).is_ok());
assert!(validate_replication_target_endpoint_inner(&parse_url("http://[::1]:9000"), true).is_ok());
assert!(validate_replication_target_endpoint_inner(&parse_url("http://localhost:9000"), true).is_ok());
}
#[test]
fn replication_endpoint_opt_in_does_not_open_other_ssrf_targets() {
// The loopback opt-in must not widen into link-local / metadata endpoints.
let err = validate_replication_target_endpoint_inner(&parse_url("http://169.254.169.254/latest/meta-data"), true)
.expect_err("metadata endpoint must stay rejected even with loopback opt-in");
assert!(matches!(
err,
OutboundUrlError::ForbiddenHost {
reason: "metadata endpoint",
..
}
));
let err = validate_replication_target_endpoint_inner(&parse_url("http://[fe80::1]:9000"), true)
.expect_err("link-local must stay rejected even with loopback opt-in");
assert!(matches!(
err,
OutboundUrlError::ForbiddenHost {
reason: "link-local address",
..
}
));
}
#[test]
fn remote_target_connection_error_display_redacts_access_key() {
let err = BucketTargetError::RemoteTargetConnectionErr {
crates/ecstore/src/bucket/replication/replication_resyncer.rs
+32 -1
View File
@@ -318,6 +318,20 @@ impl ReplicationResyncer {
return Ok(());
}
if state.resync_status == ResyncStatusType::ResyncCanceled && status != ResyncStatusType::ResyncCanceled {
debug!(
event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC,
bucket = %opts.bucket,
arn = %opts.arn,
incoming_status = %status,
reason = "canceled_status_is_terminal",
"Skipped resync status update after cancellation"
);
return Ok(());
}
if state.resync_id.is_empty() {
state.resync_id = opts.resync_id.clone();
}
@@ -339,7 +353,24 @@ impl ReplicationResyncer {
(bucket_status.clone(), status_duration)
};
save_resync_status(&opts.bucket, &bucket_status, obj_layer).await?;
save_resync_status(&opts.bucket, &bucket_status, obj_layer.clone()).await?;
if status != ResyncStatusType::ResyncCanceled {
let canceled_status = self
.status_map
.read()
.await
.get(&opts.bucket)
.filter(|current| {
current.targets_map.get(&opts.arn).is_some_and(|target| {
target.resync_id == opts.resync_id && target.resync_status == ResyncStatusType::ResyncCanceled
})
})
.cloned();
if let Some(canceled_status) = canceled_status {
save_resync_status(&opts.bucket, &canceled_status, obj_layer).await?;
return Ok(());
}
}
if let Some(stats) = runtime_sources::replication_stats() {
stats.record_resync_status(&opts.bucket, status, status_duration).await;
}