diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 493dece36..efd57a012 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -12,14 +12,18 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::common::{RustFSTestEnvironment, init_logging, local_http_client}; +use crate::common::{ + RustFSTestEnvironment, awscurl_available, awscurl_post_sts_form_urlencoded, init_logging, local_http_client, +}; +use aws_sdk_s3::config::{Credentials, Region}; use aws_sdk_s3::primitives::ByteStream; use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration}; +use aws_sdk_s3::{Client, Config}; use http::header::{CONTENT_TYPE, HOST}; use reqwest::StatusCode; use rustfs_madmin::{ - PeerInfo, PeerSite, ReplicateAddStatus, ReplicateEditStatus, ReplicateRemoveStatus, SRRemoveReq, SRResyncOpStatus, - SRStatusInfo, SiteReplicationInfo, SyncStatus, + AddServiceAccountReq, ListServiceAccountsResp, PeerInfo, PeerSite, ReplicateAddStatus, ReplicateEditStatus, + ReplicateRemoveStatus, SRRemoveReq, SRResyncOpStatus, SRStatusInfo, SiteReplicationInfo, SyncStatus, }; use rustfs_signer::constants::UNSIGNED_PAYLOAD; use rustfs_signer::sign_v4; @@ -30,6 +34,8 @@ use std::error::Error; use time::Duration as TimeDuration; use tokio::time::{Duration, sleep}; +type TestResult = Result<(), Box>; + #[derive(Debug, Clone, serde::Deserialize)] struct ReplicationResetStatusResponse { #[serde(rename = "Targets", default)] @@ -79,6 +85,65 @@ async fn signed_request( Ok(request_builder.send().await?) } +async fn signed_request_with_session_token( + method: http::Method, + url: &str, + access_key: &str, + secret_key: &str, + session_token: &str, + body: Option>, + content_type: Option<&str>, +) -> Result> { + let uri = url.parse::()?; + let authority = uri.authority().ok_or("request URL missing authority")?.to_string(); + let mut request = http::Request::builder().method(method.clone()).uri(uri); + request = request.header(HOST, authority); + request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD); + if !session_token.is_empty() { + request = request.header("x-amz-security-token", session_token); + } + if let Some(content_type) = content_type { + request = request.header(CONTENT_TYPE, content_type); + } + + let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default(); + let signed = sign_v4( + request.body(Body::empty())?, + content_len, + access_key, + secret_key, + session_token, + "us-east-1", + ); + + let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?; + let client = local_http_client(); + let mut request_builder = client.request(reqwest_method, url); + for (name, value) in signed.headers() { + request_builder = request_builder.header(name, value); + } + if let Some(body) = body { + request_builder = request_builder.body(body); + } + + Ok(request_builder.send().await?) +} + +fn extract_xml_tag(xml: &str, tag: &str) -> Option { + let open = format!("<{tag}>"); + let close = format!(""); + let start = xml.find(&open)? + open.len(); + let end = xml[start..].find(&close)? + start; + Some(xml[start..end].to_string()) +} + +fn parse_assume_role_credentials(xml: &str) -> Result<(String, String, String), Box> { + let access_key = extract_xml_tag(xml, "AccessKeyId").ok_or("missing AccessKeyId in AssumeRole response")?; + let secret_key = extract_xml_tag(xml, "SecretAccessKey").ok_or("missing SecretAccessKey in AssumeRole response")?; + let session_token = extract_xml_tag(xml, "SessionToken").ok_or("missing SessionToken in AssumeRole response")?; + Ok((access_key, secret_key, session_token)) +} + async fn set_replication_target( source_env: &RustFSTestEnvironment, source_bucket: &str, @@ -213,6 +278,143 @@ async fn enable_bucket_versioning(env: &RustFSTestEnvironment, bucket: &str) -> Ok(()) } +fn create_user_s3_client(env: &RustFSTestEnvironment, access_key: &str, secret_key: &str) -> Client { + let credentials = Credentials::new(access_key, secret_key, None, None, "e2e-site-replication"); + let config = Config::builder() + .credentials_provider(credentials) + .region(Region::new("us-east-1")) + .endpoint_url(&env.url) + .force_path_style(true) + .behavior_version_latest() + .build(); + Client::from_conf(config) +} + +async fn admin_create_user( + env: &RustFSTestEnvironment, + username: &str, + secret_key: &str, +) -> Result<(), Box> { + let url = format!("{}/rustfs/admin/v3/add-user?accessKey={}", env.url, username); + let body = serde_json::json!({ + "secretKey": secret_key, + "status": "enabled" + }); + let response = signed_request( + http::Method::PUT, + &url, + &env.access_key, + &env.secret_key, + Some(body.to_string().into_bytes()), + Some("application/json"), + ) + .await?; + + if response.status() != StatusCode::OK { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(format!("create user failed: {status} {body}").into()); + } + + Ok(()) +} + +async fn admin_add_canned_policy( + env: &RustFSTestEnvironment, + policy_name: &str, + policy: &serde_json::Value, +) -> Result<(), Box> { + let url = format!("{}/rustfs/admin/v3/add-canned-policy?name={}", env.url, policy_name); + let response = signed_request( + http::Method::PUT, + &url, + &env.access_key, + &env.secret_key, + Some(policy.to_string().into_bytes()), + Some("application/json"), + ) + .await?; + + if response.status() != StatusCode::OK { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(format!("add canned policy failed: {status} {body}").into()); + } + + Ok(()) +} + +async fn admin_attach_policy_to_user( + env: &RustFSTestEnvironment, + policy_name: &str, + username: &str, +) -> Result<(), Box> { + let url = format!( + "{}/rustfs/admin/v3/set-user-or-group-policy?policyName={}&userOrGroup={}&isGroup=false", + env.url, policy_name, username + ); + let response = signed_request(http::Method::PUT, &url, &env.access_key, &env.secret_key, Some(Vec::new()), None).await?; + + if response.status() != StatusCode::OK { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(format!("attach policy to user failed: {status} {body}").into()); + } + + Ok(()) +} + +async fn admin_update_group_members( + env: &RustFSTestEnvironment, + group_name: &str, + members: &[&str], +) -> Result<(), Box> { + let url = format!("{}/rustfs/admin/v3/update-group-members", env.url); + let body = serde_json::json!({ + "group": group_name, + "members": members, + "isRemove": false, + "groupStatus": "enabled" + }); + let response = signed_request( + http::Method::PUT, + &url, + &env.access_key, + &env.secret_key, + Some(body.to_string().into_bytes()), + Some("application/json"), + ) + .await?; + + if response.status() != StatusCode::OK { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(format!("update group members failed: {status} {body}").into()); + } + + Ok(()) +} + +async fn admin_attach_policy_to_group( + env: &RustFSTestEnvironment, + policy_name: &str, + group_name: &str, +) -> Result<(), Box> { + let url = format!( + "{}/rustfs/admin/v3/set-user-or-group-policy?policyName={}&userOrGroup={}&isGroup=true", + env.url, policy_name, group_name + ); + let response = signed_request(http::Method::PUT, &url, &env.access_key, &env.secret_key, Some(Vec::new()), None).await?; + + if response.status() != StatusCode::OK { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(format!("attach policy to group failed: {status} {body}").into()); + } + + Ok(()) +} + async fn run_replication_check( env: &RustFSTestEnvironment, bucket: &str, @@ -259,6 +461,182 @@ async fn remove_replication_target_request( signed_request(http::Method::DELETE, &url, &env.access_key, &env.secret_key, None, None).await } +async fn add_service_account( + env: &RustFSTestEnvironment, + signer_access_key: &str, + signer_secret_key: &str, + req: &AddServiceAccountReq, +) -> Result<(String, String), Box> { + let url = format!("{}/rustfs/admin/v3/add-service-account", env.url); + let response = signed_request( + http::Method::PUT, + &url, + signer_access_key, + signer_secret_key, + Some(serde_json::to_vec(req)?), + Some("application/json"), + ) + .await?; + + if response.status() != StatusCode::OK { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(format!("add service account failed: {status} {body}").into()); + } + + let body = response.bytes().await?; + let parsed: serde_json::Value = serde_json::from_slice(&body)?; + let credentials = parsed + .get("credentials") + .ok_or("add service account response missing credentials")?; + let access_key = credentials + .get("accessKey") + .and_then(|value| value.as_str()) + .ok_or("add service account response missing access key")? + .to_string(); + let secret_key = credentials + .get("secretKey") + .and_then(|value| value.as_str()) + .ok_or("add service account response missing secret key")? + .to_string(); + + Ok((access_key, secret_key)) +} + +async fn add_service_account_with_session_token( + env: &RustFSTestEnvironment, + signer_access_key: &str, + signer_secret_key: &str, + session_token: &str, + req: &AddServiceAccountReq, +) -> Result<(String, String), Box> { + let url = format!("{}/rustfs/admin/v3/add-service-account", env.url); + let response = signed_request_with_session_token( + http::Method::PUT, + &url, + signer_access_key, + signer_secret_key, + session_token, + Some(serde_json::to_vec(req)?), + Some("application/json"), + ) + .await?; + + if response.status() != StatusCode::OK { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(format!("add service account with session token failed: {status} {body}").into()); + } + + let body = response.bytes().await?; + let parsed: serde_json::Value = serde_json::from_slice(&body)?; + let credentials = parsed + .get("credentials") + .ok_or("add service account response missing credentials")?; + let access_key = credentials + .get("accessKey") + .and_then(|value| value.as_str()) + .ok_or("add service account response missing access key")? + .to_string(); + let secret_key = credentials + .get("secretKey") + .and_then(|value| value.as_str()) + .ok_or("add service account response missing secret key")? + .to_string(); + + Ok((access_key, secret_key)) +} + +async fn list_service_accounts( + env: &RustFSTestEnvironment, + signer_access_key: &str, + signer_secret_key: &str, + user: Option<&str>, +) -> Result> { + let mut url = format!("{}/rustfs/admin/v3/list-service-accounts", env.url); + if let Some(user) = user { + url.push_str("?user="); + url.push_str(&urlencoding::encode(user)); + } + + let response = signed_request(http::Method::GET, &url, signer_access_key, signer_secret_key, None, None).await?; + if response.status() != StatusCode::OK { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(format!("list service accounts failed: {status} {body}").into()); + } + + Ok(response.json().await?) +} + +async fn wait_for_service_accounts( + env: &RustFSTestEnvironment, + signer_access_key: &str, + signer_secret_key: &str, + user: Option<&str>, + expected: &[&str], +) -> Result> { + for _ in 0..20 { + let resp = list_service_accounts(env, signer_access_key, signer_secret_key, user).await?; + let access_keys: Vec<&str> = resp.accounts.iter().map(|account| account.access_key.as_str()).collect(); + if expected + .iter() + .all(|expected_key| access_keys.iter().any(|actual| actual == expected_key)) + { + return Ok(resp); + } + sleep(Duration::from_millis(250)).await; + } + + Err(format!("service accounts did not reach expected keys {expected:?} on {}", env.address).into()) +} + +async fn wait_for_object_on_target( + client: &aws_sdk_s3::Client, + bucket: &str, + key: &str, +) -> Result, Box> { + for _ in 0..40 { + match client.get_object().bucket(bucket).key(key).send().await { + Ok(output) => { + let body = output.body.collect().await?.into_bytes().to_vec(); + return Ok(body); + } + Err(err) => { + if err.to_string().contains("NoSuchKey") || err.to_string().contains("NotFound") { + sleep(Duration::from_millis(250)).await; + continue; + } + return Err(err.into()); + } + } + } + + Err(format!("object {bucket}/{key} was not replicated 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 { + match client.get_object().bucket(bucket).key(key).send().await { + Ok(output) => { + let body = output.body.collect().await?.into_bytes().to_vec(); + return Ok(body); + } + Err(err) => { + last_error = Some(err.to_string()); + sleep(Duration::from_millis(250)).await; + } + } + } + + Err(format!( + "user could not read replicated object {bucket}/{key} in time; last error: {}", + last_error.unwrap_or_else(|| "unknown".to_string()) + ) + .into()) +} + async fn list_replication_targets_request( env: &RustFSTestEnvironment, bucket: Option<&str>, @@ -1546,3 +1924,449 @@ async fn test_site_replication_state_edit_fresh_and_stale_real_dual_node() -> Re Ok(()) } + +#[tokio::test] +#[serial] +async fn test_site_replication_replicates_object_with_bucket_versioning_real_dual_node() -> TestResult { + init_logging(); + + let mut source_env = RustFSTestEnvironment::new().await?; + source_env.start_rustfs_server(vec![]).await?; + + let mut target_env = RustFSTestEnvironment::new().await?; + target_env.start_rustfs_server_without_cleanup(vec![]).await?; + + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + let bucket = "site-repl-versioned"; + let key = "hello.txt"; + let payload = b"site replication should replicate after enabling versioning".to_vec(); + + let add_status = site_replication_add( + &source_env, + &[ + PeerSite { + name: "source-site".to_string(), + endpoint: source_env.url.clone(), + access_key: source_env.access_key.clone(), + secret_key: source_env.secret_key.clone(), + }, + PeerSite { + name: "target-site".to_string(), + endpoint: target_env.url.clone(), + access_key: target_env.access_key.clone(), + secret_key: target_env.secret_key.clone(), + }, + ], + ) + .await?; + assert!(add_status.success, "unexpected site add result: {:?}", add_status); + + let _source_info = wait_for_site_replication_enabled(&source_env, 2).await?; + let _target_info = wait_for_site_replication_enabled(&target_env, 2).await?; + + source_client.create_bucket().bucket(bucket).send().await?; + enable_bucket_versioning(&source_env, bucket).await?; + let replication_response = signed_request( + http::Method::GET, + &format!("{}/{bucket}?replication", source_env.url), + &source_env.access_key, + &source_env.secret_key, + None, + None, + ) + .await?; + let replication_status = replication_response.status(); + let replication_body = replication_response.text().await.unwrap_or_default(); + assert_eq!( + replication_status, + StatusCode::OK, + "source bucket replication config missing after site replication setup: {replication_body}" + ); + source_client + .put_object() + .bucket(bucket) + .key(key) + .body(ByteStream::from(payload.clone())) + .send() + .await?; + + let replicated = wait_for_object_on_target(&target_client, bucket, key).await?; + assert_eq!(replicated, payload); + + Ok(()) +} + +#[tokio::test] +#[serial] +async fn test_site_replication_replicates_policy_backed_user_access_real_dual_node() -> Result<(), Box> { + init_logging(); + + let mut source_env = RustFSTestEnvironment::new().await?; + source_env.start_rustfs_server(vec![]).await?; + + let mut target_env = RustFSTestEnvironment::new().await?; + target_env.start_rustfs_server_without_cleanup(vec![]).await?; + + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + let bucket = "site-repl-policy-user"; + let key = "seed.txt"; + let payload = b"site replication policy-backed user access".to_vec(); + let policy_name = "site-repl-readonly"; + let username = "site-repl-user"; + let secret_key = "site-repl-user-secret-key-123456"; + + let add_status = site_replication_add( + &source_env, + &[ + PeerSite { + name: "source-site".to_string(), + endpoint: source_env.url.clone(), + access_key: source_env.access_key.clone(), + secret_key: source_env.secret_key.clone(), + }, + PeerSite { + name: "target-site".to_string(), + endpoint: target_env.url.clone(), + access_key: target_env.access_key.clone(), + secret_key: target_env.secret_key.clone(), + }, + ], + ) + .await?; + assert!(add_status.success, "unexpected site add result: {:?}", add_status); + + let _source_info = wait_for_site_replication_enabled(&source_env, 2).await?; + let _target_info = wait_for_site_replication_enabled(&target_env, 2).await?; + + source_client.create_bucket().bucket(bucket).send().await?; + enable_bucket_versioning(&source_env, bucket).await?; + source_client + .put_object() + .bucket(bucket) + .key(key) + .body(ByteStream::from(payload.clone())) + .send() + .await?; + + let replicated = wait_for_object_on_target(&target_client, bucket, key).await?; + assert_eq!(replicated, payload); + + let policy = serde_json::json!({ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Action": ["s3:GetObject"], + "Resource": [format!("arn:aws:s3:::{bucket}/*")] + }, + { + "Effect": "Allow", + "Action": ["s3:GetBucketLocation", "s3:ListBucket"], + "Resource": [format!("arn:aws:s3:::{bucket}")] + } + ] + }); + admin_add_canned_policy(&source_env, policy_name, &policy).await?; + admin_create_user(&source_env, username, secret_key).await?; + admin_attach_policy_to_user(&source_env, policy_name, username).await?; + + let target_user_client = create_user_s3_client(&target_env, username, secret_key); + let fetched = wait_for_user_get_object(&target_user_client, bucket, key).await?; + assert_eq!(fetched, payload); + + Ok(()) +} + +#[tokio::test] +#[serial] +async fn test_site_replication_replicates_group_policy_backed_access_real_dual_node() -> Result<(), Box> +{ + init_logging(); + + let mut source_env = RustFSTestEnvironment::new().await?; + source_env.start_rustfs_server(vec![]).await?; + + let mut target_env = RustFSTestEnvironment::new().await?; + target_env.start_rustfs_server_without_cleanup(vec![]).await?; + + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + let bucket = "site-repl-policy-group"; + let key = "seed.txt"; + let payload = b"site replication group-policy-backed user access".to_vec(); + let policy_name = "site-repl-group-readonly"; + let group_name = "site-repl-group"; + let username = "site-repl-group-user"; + let secret_key = "site-repl-group-user-secret-key-12"; + + let add_status = site_replication_add( + &source_env, + &[ + PeerSite { + name: "source-site".to_string(), + endpoint: source_env.url.clone(), + access_key: source_env.access_key.clone(), + secret_key: source_env.secret_key.clone(), + }, + PeerSite { + name: "target-site".to_string(), + endpoint: target_env.url.clone(), + access_key: target_env.access_key.clone(), + secret_key: target_env.secret_key.clone(), + }, + ], + ) + .await?; + assert!(add_status.success, "unexpected site add result: {:?}", add_status); + + let _source_info = wait_for_site_replication_enabled(&source_env, 2).await?; + let _target_info = wait_for_site_replication_enabled(&target_env, 2).await?; + + source_client.create_bucket().bucket(bucket).send().await?; + enable_bucket_versioning(&source_env, bucket).await?; + source_client + .put_object() + .bucket(bucket) + .key(key) + .body(ByteStream::from(payload.clone())) + .send() + .await?; + + let replicated = wait_for_object_on_target(&target_client, bucket, key).await?; + assert_eq!(replicated, payload); + + let policy = serde_json::json!({ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Action": ["s3:GetObject"], + "Resource": [format!("arn:aws:s3:::{bucket}/*")] + }, + { + "Effect": "Allow", + "Action": ["s3:GetBucketLocation", "s3:ListBucket"], + "Resource": [format!("arn:aws:s3:::{bucket}")] + } + ] + }); + admin_add_canned_policy(&source_env, policy_name, &policy).await?; + admin_create_user(&source_env, username, secret_key).await?; + admin_update_group_members(&source_env, group_name, &[username]).await?; + admin_attach_policy_to_group(&source_env, policy_name, group_name).await?; + + let target_user_client = create_user_s3_client(&target_env, username, secret_key); + let fetched = wait_for_user_get_object(&target_user_client, bucket, key).await?; + assert_eq!(fetched, payload); + + Ok(()) +} + +#[tokio::test] +#[serial] +async fn test_site_replication_replicates_multiple_service_accounts_real_dual_node() -> Result<(), Box> { + init_logging(); + + let mut source_env = RustFSTestEnvironment::new().await?; + source_env.start_rustfs_server(vec![]).await?; + + let mut target_env = RustFSTestEnvironment::new().await?; + target_env.start_rustfs_server_without_cleanup(vec![]).await?; + + let add_status = site_replication_add( + &source_env, + &[ + PeerSite { + name: "source-site".to_string(), + endpoint: source_env.url.clone(), + access_key: source_env.access_key.clone(), + secret_key: source_env.secret_key.clone(), + }, + PeerSite { + name: "target-site".to_string(), + endpoint: target_env.url.clone(), + access_key: target_env.access_key.clone(), + secret_key: target_env.secret_key.clone(), + }, + ], + ) + .await?; + assert!(add_status.success, "unexpected site add result: {:?}", add_status); + + let _source_info = wait_for_site_replication_enabled(&source_env, 2).await?; + let _target_info = wait_for_site_replication_enabled(&target_env, 2).await?; + + let first_req = AddServiceAccountReq { + policy: None, + target_user: None, + access_key: "svc-alpha".to_string(), + secret_key: "svc-alpha-secret-key-1234567890abcdef".to_string(), + name: Some("svc-alpha".to_string()), + description: Some("first replicated service account".to_string()), + expiration: None, + comment: None, + }; + let first = add_service_account(&source_env, &source_env.access_key, &source_env.secret_key, &first_req).await?; + + let target_after_first = wait_for_service_accounts( + &target_env, + &target_env.access_key, + &target_env.secret_key, + Some(&source_env.access_key), + &["svc-alpha"], + ) + .await?; + assert!( + target_after_first + .accounts + .iter() + .any(|account| account.access_key == "svc-alpha"), + "target accounts missing svc-alpha: {:?}", + target_after_first.accounts + ); + + let second_req = AddServiceAccountReq { + policy: None, + target_user: None, + access_key: "svc-beta".to_string(), + secret_key: "svc-beta-secret-key-1234567890abcdef1".to_string(), + name: Some("svc-beta".to_string()), + description: Some("second replicated service account".to_string()), + expiration: None, + comment: None, + }; + let _second = add_service_account(&source_env, &first.0, &first.1, &second_req).await?; + + let target_after_second = wait_for_service_accounts( + &target_env, + &target_env.access_key, + &target_env.secret_key, + Some(&source_env.access_key), + &["svc-alpha", "svc-beta"], + ) + .await?; + assert!( + target_after_second + .accounts + .iter() + .any(|account| account.access_key == "svc-beta"), + "target accounts missing svc-beta: {:?}", + target_after_second.accounts + ); + + Ok(()) +} + +#[tokio::test] +#[serial] +async fn test_site_replication_replicates_service_accounts_created_from_sts_session_real_dual_node() -> TestResult { + init_logging(); + + if !awscurl_available() { + eprintln!("Skipping STS site replication service-account test because awscurl is unavailable"); + return Ok(()); + } + + let mut source_env = RustFSTestEnvironment::new().await?; + source_env.start_rustfs_server(vec![]).await?; + + let mut target_env = RustFSTestEnvironment::new().await?; + target_env.start_rustfs_server_without_cleanup(vec![]).await?; + + let add_status = site_replication_add( + &source_env, + &[ + PeerSite { + name: "source-site".to_string(), + endpoint: source_env.url.clone(), + access_key: source_env.access_key.clone(), + secret_key: source_env.secret_key.clone(), + }, + PeerSite { + name: "target-site".to_string(), + endpoint: target_env.url.clone(), + access_key: target_env.access_key.clone(), + secret_key: target_env.secret_key.clone(), + }, + ], + ) + .await?; + assert!(add_status.success, "unexpected site add result: {:?}", add_status); + + let _source_info = wait_for_site_replication_enabled(&source_env, 2).await?; + let _target_info = wait_for_site_replication_enabled(&target_env, 2).await?; + + let assume_role_body = "Action=AssumeRole&Version=2011-06-15&DurationSeconds=3600"; + let sts_xml = awscurl_post_sts_form_urlencoded( + &format!("{}/", source_env.url.trim_end_matches('/')), + assume_role_body, + &source_env.access_key, + &source_env.secret_key, + ) + .await?; + let (sts_access_key, sts_secret_key, sts_session_token) = parse_assume_role_credentials(&sts_xml)?; + + let first_req = AddServiceAccountReq { + policy: None, + target_user: None, + access_key: "svc-sts-alpha".to_string(), + secret_key: "svc-sts-alpha-secret-key-1234567890".to_string(), + name: Some("svc-sts-alpha".to_string()), + description: Some("sts-created replicated service account".to_string()), + expiration: None, + comment: None, + }; + let first = + add_service_account_with_session_token(&source_env, &sts_access_key, &sts_secret_key, &sts_session_token, &first_req) + .await?; + + let target_after_first = wait_for_service_accounts( + &target_env, + &target_env.access_key, + &target_env.secret_key, + Some(&source_env.access_key), + &["svc-sts-alpha"], + ) + .await?; + assert!( + target_after_first + .accounts + .iter() + .any(|account| account.access_key == "svc-sts-alpha"), + "target accounts missing svc-sts-alpha: {:?}", + target_after_first.accounts + ); + + let second_req = AddServiceAccountReq { + policy: None, + target_user: None, + access_key: "svc-sts-beta".to_string(), + secret_key: "svc-sts-beta-secret-key-1234567890a".to_string(), + name: Some("svc-sts-beta".to_string()), + description: Some("second replicated service account from sts-created ak".to_string()), + expiration: None, + comment: None, + }; + let _second = add_service_account(&source_env, &first.0, &first.1, &second_req).await?; + + let target_after_second = wait_for_service_accounts( + &target_env, + &target_env.access_key, + &target_env.secret_key, + Some(&source_env.access_key), + &["svc-sts-alpha", "svc-sts-beta"], + ) + .await?; + assert!( + target_after_second + .accounts + .iter() + .any(|account| account.access_key == "svc-sts-beta"), + "target accounts missing svc-sts-beta: {:?}", + target_after_second.accounts + ); + + Ok(()) +} diff --git a/crates/policy/src/policy/effect.rs b/crates/policy/src/policy/effect.rs index 3c642772c..3101cfa79 100644 --- a/crates/policy/src/policy/effect.rs +++ b/crates/policy/src/policy/effect.rs @@ -19,7 +19,7 @@ use strum::{EnumString, IntoStaticStr}; use super::Validator; #[derive(Serialize, Clone, Deserialize, EnumString, IntoStaticStr, Default, Debug, PartialEq)] -#[serde(try_from = "&str", into = "&str")] +#[serde(try_from = "String", into = "&str")] pub enum Effect { #[default] #[strum(serialize = "Allow")] @@ -28,6 +28,16 @@ pub enum Effect { Deny, } +impl TryFrom for Effect { + type Error = Error; + + fn try_from(value: String) -> std::result::Result { + value + .parse::() + .map_err(|e: strum::ParseError| Error::StringError(e.to_string())) + } +} + impl Effect { pub fn is_allowed(&self, allowed: bool) -> bool { if matches!(self, Self::Allow) { diff --git a/crates/policy/src/policy/policy.rs b/crates/policy/src/policy/policy.rs index 1176b862b..63c6f0fe4 100644 --- a/crates/policy/src/policy/policy.rs +++ b/crates/policy/src/policy/policy.rs @@ -1761,4 +1761,28 @@ mod test { assert!(!found); assert!(policies.is_empty()); } + + #[test] + fn test_policy_round_trips_through_json_value() { + let policy = Policy::parse_config( + br#"{ + "Version":"2012-10-17", + "Statement":[ + { + "Effect":"Allow", + "Action":["s3:GetObject"], + "Resource":["arn:aws:s3:::bucket/*"] + } + ] +}"#, + ) + .expect("policy should parse"); + + let value = serde_json::to_value(&policy).expect("policy should serialize"); + let round_trip: Policy = serde_json::from_value(value).expect("policy should deserialize from serde_json::Value"); + + assert_eq!(round_trip.version, policy.version); + assert_eq!(round_trip.statements.len(), policy.statements.len()); + assert_eq!(round_trip.statements[0].effect, policy.statements[0].effect); + } } diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index 6c770c582..1958c2e24 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -13,6 +13,7 @@ // limitations under the License. use crate::admin::auth::validate_admin_request; +use crate::admin::handlers::site_replication::site_replication_peer_deployment_id_for_endpoint; use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::admin::utils::read_compatible_admin_body; use crate::auth::{check_key_valid, get_session_token}; @@ -231,11 +232,23 @@ impl Operation for SetRemoteTargetHandler { } remote_target.source_bucket = bucket.clone(); + let site_endpoint = if remote_target.endpoint.starts_with("http://") || remote_target.endpoint.starts_with("https://") { + remote_target.endpoint.clone() + } else if remote_target.secure { + format!("https://{}", remote_target.endpoint) + } else { + format!("http://{}", remote_target.endpoint) + }; + if let Some(deployment_id) = site_replication_peer_deployment_id_for_endpoint(&site_endpoint).await { + remote_target.deployment_id = deployment_id; + } let bucket_target_sys = BucketTargetSys::get(); if !update { - let (arn, exist) = bucket_target_sys.get_remote_arn(bucket, Some(&remote_target), "").await; + let (arn, exist) = bucket_target_sys + .get_remote_arn(bucket, Some(&remote_target), remote_target.deployment_id.as_str()) + .await; remote_target.arn = arn.clone(); if exist && !arn.is_empty() { let arn_str = serde_json::to_string(&arn).unwrap_or_default(); diff --git a/rustfs/src/admin/handlers/service_account.rs b/rustfs/src/admin/handlers/service_account.rs index 914a1f2b7..c04e65234 100644 --- a/rustfs/src/admin/handlers/service_account.rs +++ b/rustfs/src/admin/handlers/service_account.rs @@ -79,6 +79,18 @@ fn delete_service_account_success_status(path: &str) -> StatusCode { } } +fn merge_derived_service_account_claims( + target_claims: &mut HashMap, + source_claims: &HashMap, +) { + for (key, value) in source_claims { + if key == "exp" { + continue; + } + target_claims.insert(key.clone(), value.clone()); + } +} + fn is_service_account_owner_of(caller: &StoredCredentials, target_parent_user: &str) -> bool { let caller_parent = if caller.parent_user.is_empty() { caller.access_key.as_str() @@ -303,13 +315,7 @@ impl Operation for AddServiceAccount { opts.claims = Some(HashMap::new()); } - for (k, v) in claims.iter() { - if claims.contains_key("exp") { - continue; - } - - opts.claims.as_mut().unwrap().insert(k.clone(), v.clone()); - } + merge_derived_service_account_claims(opts.claims.as_mut().unwrap(), &claims); } } @@ -1552,4 +1558,20 @@ mod tests { assert!(is_service_account_owner_of(&derived_owner, "owner-user")); assert!(!is_service_account_owner_of(&foreign_user, "owner-user")); } + + #[test] + fn merge_derived_service_account_claims_skips_only_expiration() { + let mut merged = HashMap::new(); + let source = HashMap::from([ + ("exp".to_string(), json!(123456)), + ("parent".to_string(), json!("owner-user")), + ("custom".to_string(), json!("value")), + ]); + + merge_derived_service_account_claims(&mut merged, &source); + + assert!(!merged.contains_key("exp")); + assert_eq!(merged.get("parent"), Some(&json!("owner-user"))); + assert_eq!(merged.get("custom"), Some(&json!("value"))); + } } diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index e2c7e4dda..a27fb5477 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -34,14 +34,15 @@ use rustfs_ecstore::bucket::metadata::{ use rustfs_ecstore::bucket::metadata_sys; use rustfs_ecstore::bucket::replication::GLOBAL_REPLICATION_STATS; use rustfs_ecstore::bucket::replication::{ReplicationConfigurationExt, ResyncOpts, get_global_replication_pool}; -use rustfs_ecstore::bucket::target::{BucketTarget, BucketTargetType}; -use rustfs_ecstore::bucket::utils::serialize; +use rustfs_ecstore::bucket::target::{ARN, BucketTarget, BucketTargetType, BucketTargets, Credentials}; +use rustfs_ecstore::bucket::utils::{deserialize, serialize}; use rustfs_ecstore::config::com::{delete_config, read_config, save_config}; use rustfs_ecstore::config::get_global_server_config; use rustfs_ecstore::error::Error as StorageError; use rustfs_ecstore::global::{get_global_deployment_id, get_global_endpoints_opt, get_global_region, global_rustfs_port}; use rustfs_ecstore::new_object_layer_fn; use rustfs_ecstore::store_api::{BucketOperations, BucketOptions, DeleteBucketOptions, MakeBucketOptions, SRBucketDeleteOp}; +use rustfs_iam::error::is_err_no_such_service_account; use rustfs_iam::store::{MappedPolicy, UserType}; use rustfs_iam::sys::{NewServiceAccountOpts, UpdateServiceAccountOpts, get_claims_from_token_with_secret}; use rustfs_iam::{get_global_iam_sys, get_oidc}; @@ -59,7 +60,11 @@ use rustfs_policy::policy::{ }; use rustfs_signer::constants::UNSIGNED_PAYLOAD; use rustfs_signer::sign_v4; -use s3s::dto::{BucketVersioningStatus, VersioningConfiguration}; +use s3s::dto::{ + BucketVersioningStatus, DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, + Destination, ExistingObjectReplication, ExistingObjectReplicationStatus, ReplicationConfiguration, ReplicationRule, + ReplicationRuleStatus, VersioningConfiguration, +}; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; use serde::Deserialize; use serde::Serialize; @@ -695,6 +700,12 @@ fn existing_peer_for_endpoint(state: &SiteReplicationState, endpoint: &str) -> O .cloned() } +fn peer_deployment_id_for_endpoint(state: &SiteReplicationState, endpoint: &str) -> Option { + existing_peer_for_endpoint(state, endpoint) + .map(|peer| peer.deployment_id) + .filter(|deployment_id| !deployment_id.is_empty()) +} + fn normalize_peer_info(mut peer: PeerInfo) -> PeerInfo { if peer.deployment_id.is_empty() { peer.deployment_id = deployment_id_for_endpoint(&peer.endpoint); @@ -938,10 +949,13 @@ async fn broadcast_site_replication_json(path: &str, body: &T) -> } pub async fn site_replication_make_bucket_hook(bucket: &str, lock_enabled: bool) -> S3Result<()> { - let Some((_, _)) = runtime_site_replication_targets().await? else { + let Some((state, local_peer)) = runtime_site_replication_targets().await? else { return Ok(()); }; + ensure_site_replication_bucket_targets(bucket, &state, &local_peer, None).await?; + ensure_site_replication_bucket_replication_config(bucket, &state, &local_peer).await?; + let created_at = new_object_layer_fn() .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()))? .get_bucket_info(bucket, &BucketOptions::default()) @@ -1533,6 +1547,244 @@ fn bucket_target_matches_peer(target: &BucketTarget, peer: &PeerInfo) -> bool { || bucket_target_endpoint(target) == canonical_endpoint(&peer.endpoint) } +fn site_replication_target_arns_by_peer(config: Option<&s3s::dto::ReplicationConfiguration>) -> HashMap { + let mut arns_by_peer = HashMap::new(); + let Some(config) = config else { + return arns_by_peer; + }; + + let mut configured_arns = Vec::new(); + if !config.role.trim().is_empty() { + configured_arns.push(config.role.clone()); + } + for rule in &config.rules { + let arn = rule.destination.bucket.trim(); + if !arn.is_empty() { + configured_arns.push(arn.to_string()); + } + } + + for arn in configured_arns { + if let Ok(parsed) = arn.parse::() + && parsed.arn_type == BucketTargetType::ReplicationService + && !parsed.id.is_empty() + { + arns_by_peer.entry(parsed.id).or_insert(arn); + } + } + + arns_by_peer +} + +fn site_replication_bucket_target_for_peer( + bucket: &str, + state: &SiteReplicationState, + peer: &PeerInfo, + arn_override: Option, +) -> Option { + if state.service_account_access_key.is_empty() || state.service_account_secret_key.is_empty() { + return None; + } + + let parsed = Url::parse(&peer.endpoint) + .ok() + .or_else(|| Url::parse(&format!("http://{}", peer.endpoint.trim())).ok())?; + let host = parsed.host_str()?; + let port = parsed.port_or_known_default()?; + let arn = arn_override.unwrap_or_else(|| { + ARN::new( + BucketTargetType::ReplicationService, + peer.deployment_id.clone(), + String::new(), + bucket.to_string(), + ) + .to_string() + }); + + Some(BucketTarget { + source_bucket: bucket.to_string(), + endpoint: format!("{host}:{port}"), + credentials: Some(Credentials { + access_key: state.service_account_access_key.clone(), + secret_key: state.service_account_secret_key.clone(), + session_token: None, + expiration: None, + }), + target_bucket: bucket.to_string(), + secure: parsed.scheme().eq_ignore_ascii_case("https"), + arn, + target_type: BucketTargetType::ReplicationService, + deployment_id: peer.deployment_id.clone(), + ..Default::default() + }) +} + +fn reconcile_site_replication_bucket_targets( + existing: BucketTargets, + bucket: &str, + state: &SiteReplicationState, + local_peer: &PeerInfo, + config: Option<&s3s::dto::ReplicationConfiguration>, +) -> BucketTargets { + if !state.enabled() || state.service_account_access_key.is_empty() || state.service_account_secret_key.is_empty() { + return existing; + } + + let configured_arns = site_replication_target_arns_by_peer(config); + let mut targets = existing.targets; + + for peer in state.peers.values() { + if peer.deployment_id == local_peer.deployment_id || same_endpoint(&peer.endpoint, &local_peer.endpoint) { + continue; + } + + let Some(mut target) = + site_replication_bucket_target_for_peer(bucket, state, peer, configured_arns.get(&peer.deployment_id).cloned()) + else { + continue; + }; + + if let Some(index) = targets.iter().position(|existing| { + existing.target_type == BucketTargetType::ReplicationService + && (bucket_target_matches_peer(existing, peer) || existing.arn == target.arn) + }) { + let existing = targets[index].clone(); + target.path = existing.path; + target.region = existing.region; + target.bandwidth_limit = existing.bandwidth_limit; + target.replication_sync = existing.replication_sync; + target.storage_class = existing.storage_class; + target.health_check_duration = existing.health_check_duration; + target.disable_proxy = existing.disable_proxy; + target.reset_before_date = existing.reset_before_date; + target.reset_id = existing.reset_id; + target.total_downtime = existing.total_downtime; + target.last_online = existing.last_online; + target.online = existing.online; + target.latency = existing.latency; + target.edge = existing.edge; + target.edge_sync_before_expiry = existing.edge_sync_before_expiry; + target.offline_count = existing.offline_count; + targets[index] = target; + } else { + targets.push(target); + } + } + + BucketTargets { targets } +} + +fn build_site_replication_rule(arn: &str, priority: i32, rule_id: &str) -> ReplicationRule { + ReplicationRule { + delete_marker_replication: Some(DeleteMarkerReplication { + status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)), + }), + delete_replication: Some(DeleteReplication { + status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED), + }), + destination: Destination { + bucket: arn.to_string(), + ..Default::default() + }, + existing_object_replication: Some(ExistingObjectReplication { + status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::ENABLED), + }), + filter: None, + id: Some(rule_id.to_string()), + prefix: None, + priority: Some(priority), + source_selection_criteria: None, + status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED), + } +} + +fn build_site_replication_config( + bucket: &str, + state: &SiteReplicationState, + local_peer: &PeerInfo, +) -> Option { + let mut rules = Vec::new(); + for peer in state.peers.values() { + if peer.deployment_id == local_peer.deployment_id || same_endpoint(&peer.endpoint, &local_peer.endpoint) { + continue; + } + + let Some(target) = site_replication_bucket_target_for_peer(bucket, state, peer, None) else { + continue; + }; + rules.push(build_site_replication_rule( + &target.arn, + (rules.len() + 1) as i32, + &format!("site-repl-{}", peer.deployment_id), + )); + } + + if rules.is_empty() { + None + } else { + Some(ReplicationConfiguration { + role: String::new(), + rules, + }) + } +} + +async fn ensure_site_replication_bucket_targets( + bucket: &str, + state: &SiteReplicationState, + local_peer: &PeerInfo, + config: Option<&s3s::dto::ReplicationConfiguration>, +) -> S3Result<()> { + let existing = match metadata_sys::list_bucket_targets(bucket).await { + Ok(targets) => targets, + Err(StorageError::ConfigNotFound) => BucketTargets::default(), + Err(err) => return Err(ApiError::from(err).into()), + }; + + let updated = reconcile_site_replication_bucket_targets(existing, bucket, state, local_peer, config); + if updated.targets.is_empty() { + return Ok(()); + } + + let json_targets = serde_json::to_vec(&updated) + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize bucket targets failed: {e}")))?; + metadata_sys::update(bucket, BUCKET_TARGETS_FILE, json_targets) + .await + .map_err(ApiError::from)?; + BucketTargetSys::get().update_all_targets(bucket, Some(&updated)).await; + + Ok(()) +} + +async fn ensure_site_replication_bucket_replication_config( + bucket: &str, + state: &SiteReplicationState, + local_peer: &PeerInfo, +) -> S3Result<()> { + match metadata_sys::get_replication_config(bucket).await { + Ok(_) => return Ok(()), + Err(StorageError::ConfigNotFound) => {} + Err(err) => return Err(ApiError::from(err).into()), + } + + let Some(config) = build_site_replication_config(bucket, state, local_peer) else { + return Ok(()); + }; + + let data = serialize(&config) + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize replication failed: {e}")))?; + metadata_sys::update(bucket, BUCKET_REPLICATION_CONFIG, data) + .await + .map_err(ApiError::from)?; + + Ok(()) +} + +pub async fn site_replication_peer_deployment_id_for_endpoint(endpoint: &str) -> Option { + let state = load_site_replication_state().await.ok()?; + peer_deployment_id_for_endpoint(&state, endpoint) +} + async fn start_site_bucket_resync(bucket: &str, peer: &PeerInfo, resync_id: &str) -> ResyncBucketStatus { let mut bucket_status = ResyncBucketStatus { bucket: bucket.to_string(), @@ -1744,6 +1996,16 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> { } }; + let replication_config = if item.r#type == "replication-config" { + item.replication_config + .as_ref() + .map(|raw| deserialize::(raw.as_bytes())) + .transpose() + .map_err(|e| s3_error!(InvalidRequest, "invalid replication config: {e}"))? + } else { + None + }; + let data = match item.r#type.as_str() { "policy" => item .policy @@ -1776,6 +2038,12 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> { .await .map_err(ApiError::from)?; } + + if item.r#type == "replication-config" + && let Some((state, local_peer)) = runtime_site_replication_targets().await? + { + ensure_site_replication_bucket_targets(&item.bucket, &state, &local_peer, replication_config.as_ref()).await?; + } Ok(()) } @@ -1891,23 +2159,51 @@ async fn apply_iam_item(item: SRIAMItem) -> S3Result<()> { }; if let Some(create) = change.create { let session_policy = create.session_policy.as_str().and_then(|raw| serde_json::from_str(raw).ok()); - iam_sys - .new_service_account( - &create.parent, - Some(create.groups), - NewServiceAccountOpts { - session_policy, - access_key: create.access_key, - secret_key: create.secret_key, - name: (!create.name.is_empty()).then_some(create.name), - description: (!create.description.is_empty()).then_some(create.description), - expiration: create.expiration, - allow_site_replicator_account: true, - claims: Some(create.claims), - }, - ) - .await - .map_err(ApiError::from)?; + match iam_sys.get_service_account(&create.access_key).await { + Ok((existing, _)) => { + if existing.parent_user != create.parent { + return Err(s3_error!( + InvalidRequest, + "service account {} already exists with a different parent user", + create.access_key + )); + } + iam_sys + .update_service_account( + &create.access_key, + UpdateServiceAccountOpts { + session_policy, + secret_key: Some(create.secret_key), + name: (!create.name.is_empty()).then_some(create.name), + description: (!create.description.is_empty()).then_some(create.description), + expiration: create.expiration, + status: (!create.status.is_empty()).then_some(create.status), + }, + ) + .await + .map_err(ApiError::from)?; + } + Err(err) if is_err_no_such_service_account(&err) => { + iam_sys + .new_service_account( + &create.parent, + Some(create.groups), + NewServiceAccountOpts { + session_policy, + access_key: create.access_key, + secret_key: create.secret_key, + name: (!create.name.is_empty()).then_some(create.name), + description: (!create.description.is_empty()).then_some(create.description), + expiration: create.expiration, + allow_site_replicator_account: true, + claims: Some(create.claims), + }, + ) + .await + .map_err(ApiError::from)?; + } + Err(err) => return Err(ApiError::from(err).into()), + } return Ok(()); } @@ -2267,6 +2563,14 @@ impl Operation for SRPeerBucketOpsHandler { .get_bucket_info(&bucket, &BucketOptions::default()) .await .map_err(ApiError::from)?; + if let Some((state, local_peer)) = runtime_site_replication_targets().await? { + let replication_config = metadata_sys::get_replication_config(&bucket) + .await + .ok() + .map(|(config, _)| config); + ensure_site_replication_bucket_targets(&bucket, &state, &local_peer, replication_config.as_ref()).await?; + ensure_site_replication_bucket_replication_config(&bucket, &state, &local_peer).await?; + } } "delete-bucket" => { store @@ -2819,6 +3123,67 @@ mod tests { assert!(bucket_target_matches_peer(&target, &remote)); } + #[test] + fn test_peer_deployment_id_for_endpoint_matches_normalized_endpoint() { + let mut state = SiteReplicationState::default(); + let mut remote = peer("remote", "https://remote.example.com"); + remote.deployment_id = "remote-dep".to_string(); + state.peers.insert(remote.deployment_id.clone(), remote); + + let deployment_id = peer_deployment_id_for_endpoint(&state, "https://remote.example.com/"); + + assert_eq!(deployment_id.as_deref(), Some("remote-dep")); + } + + #[test] + fn test_reconcile_site_replication_bucket_targets_upserts_remote_peer_targets() { + let mut state = SiteReplicationState { + service_account_access_key: "site-replicator-0".to_string(), + service_account_secret_key: "secret".to_string(), + ..Default::default() + }; + state.peers.insert( + "local".to_string(), + PeerInfo { + deployment_id: "local".to_string(), + ..peer("local", "https://local.example.com") + }, + ); + state.peers.insert( + "remote".to_string(), + PeerInfo { + deployment_id: "remote".to_string(), + ..peer("remote", "http://remote.example.com:9000") + }, + ); + + let targets = reconcile_site_replication_bucket_targets( + BucketTargets::default(), + "photos", + &state, + &PeerInfo { + deployment_id: "local".to_string(), + ..peer("local", "https://local.example.com") + }, + None, + ); + + assert_eq!(targets.targets.len(), 1); + let target = &targets.targets[0]; + assert_eq!(target.target_type, BucketTargetType::ReplicationService); + assert_eq!(target.endpoint, "remote.example.com:9000"); + assert!(!target.secure); + assert_eq!(target.target_bucket, "photos"); + assert_eq!(target.deployment_id, "remote"); + assert_eq!(target.arn, "arn:rustfs:replication::remote:photos"); + let credentials = target + .credentials + .as_ref() + .expect("site replication target should carry credentials"); + assert_eq!(credentials.access_key, "site-replicator-0"); + assert_eq!(credentials.secret_key, "secret"); + } + #[test] fn test_apply_state_edit_req_only_updates_ilm_expiry_flags() { let mut state = SiteReplicationState::default(); diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 03f8f5955..18343febf 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -432,7 +432,6 @@ impl DefaultMultipartUsecase { let mt2 = HashMap::new(); let replicate_options = get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts.clone()); - let dsc = must_replicate(&bucket, &key, replicate_options).await; if dsc.replicate_any() { diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index bfe053c15..6e7de2b37 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -1837,7 +1837,6 @@ impl DefaultObjectUsecase { let repoptions = get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts.clone()); - let dsc = must_replicate(&bucket, &key, repoptions).await; if dsc.replicate_any() {