diff --git a/.config/nextest.toml b/.config/nextest.toml index 6cd6f94fc..2eae4ef04 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -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] diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 53b1bf769..2dad8072f 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -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>; +/// 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 Result> { - 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> { + // 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> { - let mut env = new_private_tmp_test_env().await?; +async fn new_replication_https_target_env() -> Result> { + 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> { + 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, Box> { 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> { + for _ in 0..40 { + let response = list_replication_targets_request(env, Some(bucket)).await?; + if response.status() == StatusCode::OK { + let targets: Vec = 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> { 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 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 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 Result<(), Box> { 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> { 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 Result<(), Box Result<(), Box Result<(), Box Result<(), Box 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 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?; diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index e7529977b..e3f376a5c 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -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 { diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 146e2f8c0..907ec2695 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -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; }