From 1e8c8d4cd538b11ec40612a5e88777b24dd47449 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 30 Aug 2026 08:44:34 +0800 Subject: [PATCH] feat(replication): support temporary target credentials (#6860) --- crates/ecstore/src/api/mod.rs | 7 +- .../ecstore/src/bucket/bucket_target_sys.rs | 451 ++++++++++++++++-- crates/ecstore/src/bucket/replication/mod.rs | 16 +- .../replication_config_boundary.rs | 16 +- .../replication/replication_object_config.rs | 34 +- .../replication/replication_resyncer.rs | 162 +++++-- .../src/bucket/target/bucket_target.rs | 40 ++ crates/replication/src/config.rs | 8 +- crates/replication/src/lib.rs | 17 +- rustfs/src/admin/handlers/replication.rs | 228 +++++---- rustfs/src/admin/handlers/system.rs | 24 +- rustfs/src/admin/storage_api.rs | 8 +- 12 files changed, 792 insertions(+), 219 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 42d2dd6b0..24b587bd2 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -194,10 +194,9 @@ pub mod bucket { BucketReplicationResyncStatus, BucketReplicationStat, BucketReplicationStats, BucketStats, DeleteReplicationConfigSnapshot, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, InQueueMetric, MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, OperatorRuleContract, - REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, - REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE, - REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, - ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig, + REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, + REPLICATE_INCOMING_DELETE, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, + REPLICATION_WRITABLE_FIELDS, ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge, diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index 682cb1bff..2966647c2 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -22,6 +22,7 @@ use crate::bucket::target::{self, BucketTarget, BucketTargets, Credentials}; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::runtime::sources as runtime_sources; use aws_credential_types::Credentials as SdkCredentials; +use aws_credential_types::provider::{ProvideCredentials, error::CredentialsError, future}; use aws_sdk_s3::config::Region as SdkRegion; use aws_sdk_s3::config::SharedHttpClient; use aws_sdk_s3::error::ProvideErrorMetadata; @@ -77,7 +78,7 @@ use std::str::FromStr as _; use std::sync::Arc; use std::sync::OnceLock; use std::sync::Weak; -use std::time::{Duration, Instant}; +use std::time::{Duration, Instant, SystemTime}; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tokio::sync::Mutex; use tokio::sync::RwLock; @@ -89,6 +90,71 @@ use uuid::Uuid; const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16; const REDACTED_CREDENTIAL: &str = ""; +const EXPIRED_REMOTE_TARGET_CREDENTIALS: &str = "remote target credentials have expired"; + +#[derive(Clone)] +struct RemoteTargetCredentialsProvider { + credentials: SdkCredentials, +} + +impl RemoteTargetCredentialsProvider { + fn resolve_at(&self, now: SystemTime) -> aws_credential_types::provider::Result { + if self.credentials.expiry().is_some_and(|expiration| expiration <= now) { + return Err(CredentialsError::provider_error(std::io::Error::other(EXPIRED_REMOTE_TARGET_CREDENTIALS))); + } + Ok(self.credentials.clone()) + } +} + +impl fmt::Debug for RemoteTargetCredentialsProvider { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("RemoteTargetCredentialsProvider") + .field("temporary", &self.credentials.session_token().is_some()) + .field("expiration", &self.credentials.expiry()) + .finish() + } +} + +impl ProvideCredentials for RemoteTargetCredentialsProvider { + fn provide_credentials<'a>(&'a self) -> future::ProvideCredentials<'a> + where + Self: 'a, + { + future::ProvideCredentials::ready(self.resolve_at(SystemTime::now())) + } + + fn fallback_on_interrupt(&self) -> Option { + self.resolve_at(SystemTime::now()).ok() + } +} + +fn remote_target_sdk_credentials( + credentials: &Credentials, + account_id: &str, + now: SystemTime, +) -> Result { + let session_token = credentials.effective_session_token(); + let expiration = credentials.effective_expiration().map(SystemTime::from); + if expiration.is_some() && session_token.is_none() { + return Err("remote target credential expiration requires a session token"); + } + if expiration.is_some_and(|expiration| expiration <= now) { + return Err(EXPIRED_REMOTE_TARGET_CREDENTIALS); + } + + let mut builder = SdkCredentials::builder() + .access_key_id(credentials.access_key.clone()) + .secret_access_key(credentials.secret_key.clone()) + .account_id(account_id.to_string()) + .provider_name("bucket_target_sys"); + if let Some(session_token) = session_token { + builder = builder.session_token(session_token.to_string()); + } + if let Some(expiration) = expiration { + builder = builder.expiry(expiration); + } + Ok(builder.build()) +} pub type HeadObjectSdkError = Box>; pub type GetObjectSdkError = Box>; @@ -845,13 +911,26 @@ impl BucketTargetSys { Ok(BucketTargets { targets: new_targets }) } + async fn mark_refresh_attempt(&self, arn: &str) { + // Rate-limit a failed config fetch as well as a failed client build. + // A successful rebuild replaces this timestamp during publication. + self.arn_remotes_map + .write() + .await + .entry(arn.to_string()) + .or_default() + .last_refresh = OffsetDateTime::now_utc(); + } + pub async fn mark_refresh_in_progress(&self, bucket: &str, arn: &str) { let mut arn_errs = self.arn_errs_map.write().await; - arn_errs.entry(arn.to_string()).or_insert_with(|| ArnErrs { - bucket: bucket.to_string(), - update_in_progress: true, + let err = arn_errs.entry(arn.to_string()).or_insert_with(|| ArnErrs { count: 1, + bucket: bucket.to_string(), + ..Default::default() }); + err.update_in_progress = true; + err.bucket = bucket.to_string(); } pub async fn mark_refresh_done(&self, bucket: &str, arn: &str) { @@ -863,15 +942,21 @@ impl BucketTargetSys { } pub async fn is_reloading_target(&self, _bucket: &str, arn: &str) -> bool { - let arn_errs = self.arn_errs_map.read().await; - arn_errs.get(arn).map(|err| err.update_in_progress).unwrap_or(false) + self.arn_errs_map + .read() + .await + .get(arn) + .is_some_and(|err| err.update_in_progress) } - pub async fn inc_arn_errs(&self, _bucket: &str, arn: &str) { + pub async fn inc_arn_errs(&self, bucket: &str, arn: &str) { let mut arn_errs = self.arn_errs_map.write().await; - if let Some(err) = arn_errs.get_mut(arn) { - err.count += 1; - } + let err = arn_errs.entry(arn.to_string()).or_insert_with(|| ArnErrs { + bucket: bucket.to_string(), + ..Default::default() + }); + err.count += 1; + err.bucket = bucket.to_string(); } pub async fn get_remote_target_client(&self, bucket: &str, arn: &str) -> Option> { @@ -884,15 +969,15 @@ impl BucketTargetSys { .unwrap_or((None, None)) }; - if let Some(cli) = cli { + let credentials_expired = cli + .as_ref() + .is_some_and(|client| client.credentials_expired_at(jiff::Timestamp::now())); + if let Some(cli) = cli + && !credentials_expired + { return Some(cli); } - // TODO(backlog): spawn an async task to proactively reload the replication target - if self.is_reloading_target(bucket, arn).await { - return None; - } - if let Some(last_refresh) = last_refresh { let now = OffsetDateTime::now_utc(); if now - last_refresh < Duration::from_secs(60 * 5) { @@ -900,16 +985,24 @@ impl BucketTargetSys { } } + // The existing per-bucket publication lock is also the reload claim: + // try-locking keeps the request path non-blocking, is cancellation-safe, + // and prevents a stale reload from publishing after a credential update. + let update_mutex = self.target_update_mutex(bucket).await; + let Ok(update_guard) = update_mutex.try_lock() else { + return None; + }; + self.mark_refresh_attempt(arn).await; + match get_bucket_targets_config(bucket).await { Ok(bucket_targets) => { - self.mark_refresh_in_progress(bucket, arn).await; - self.update_all_targets(bucket, Some(&bucket_targets)).await; - self.mark_refresh_done(bucket, arn).await; + self.update_all_targets_locked(bucket, Some(&bucket_targets)).await; } Err(e) => { error!("get bucket targets config error:{}", e); } }; + drop(update_guard); let cli = self .arn_remotes_map @@ -917,8 +1010,10 @@ impl BucketTargetSys { .await .get(arn) .and_then(|target| target.client.clone()); - if cli.is_some() { - return cli; + if let Some(cli) = cli + && !cli.credentials_expired_at(jiff::Timestamp::now()) + { + return Some(cli); } self.inc_arn_errs(bucket, arn).await; @@ -948,12 +1043,13 @@ impl BucketTargetSys { }); }; - let creds = SdkCredentials::builder() - .access_key_id(credentials.access_key.clone()) - .secret_access_key(credentials.secret_key.clone()) - .account_id(target.reset_id.clone()) - .provider_name("bucket_target_sys") - .build(); + let creds = remote_target_sdk_credentials(credentials, &target.reset_id, SystemTime::now()).map_err(|error| { + BucketTargetError::RemoteTargetConnectionErr { + bucket: target.target_bucket.clone(), + access_key: credentials.access_key.clone(), + error: error.to_string(), + } + })?; let endpoint = if target.secure { format!("https://{}", target.endpoint) @@ -973,7 +1069,7 @@ impl BucketTargetSys { let mut config_builder = S3Config::builder() .endpoint_url(endpoint.clone()) - .credentials_provider(SharedCredentialsProvider::new(creds)) + .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider { credentials: creds })) .region(SdkRegion::new(target.region.clone())) .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest()); @@ -1047,6 +1143,13 @@ impl BucketTargetSys { let update_mutex = self.target_update_mutex(bucket).await; let _update_guard = update_mutex.lock().await; + self.update_all_targets_locked(bucket, targets).await; + } + + /// Builds and publishes one bucket snapshot while its update mutex is held. + /// Keeping persisted-config reads under the same mutex prevents a stale + /// reload from overwriting a concurrent credential rotation. + async fn update_all_targets_locked(&self, bucket: &str, targets: Option<&BucketTargets>) { let mut clients = Vec::new(); if let Some(new_targets) = targets { for target in &new_targets.targets { @@ -1078,6 +1181,17 @@ impl BucketTargetSys { && !new_targets.is_empty() { for (target, client) in clients { + // Keep a timestamped placeholder for configured targets whose + // client cannot be built. Replication records these attempts as + // failed, while the placeholder prevents every object from + // triggering another metadata reload/client build for five minutes. + arn_remotes_map.insert( + target.arn.clone(), + ArnTarget { + client: None, + last_refresh: OffsetDateTime::now_utc(), + }, + ); match client { Ok(client) => { arn_remotes_map.insert( @@ -1090,11 +1204,6 @@ impl BucketTargetSys { health_map.insert(client.arn.clone(), target_health(&client)); self.update_bandwidth_limit(bucket, &target.arn, target.bandwidth_limit); } - // The target stays in `targets_map`, so it keeps showing up in - // `bucket remote ls` while no client exists to replicate through it — - // replication then drops every object for this ARN. Without this the - // rejection (loopback endpoint, bad CA, unparseable URL) left no trace - // anywhere. Err(err) => warn!( bucket = %bucket, arn = %target.arn, @@ -1962,6 +2071,13 @@ pub struct TargetClient { } impl TargetClient { + fn credentials_expired_at(&self, now: jiff::Timestamp) -> bool { + self.credentials + .as_ref() + .and_then(Credentials::effective_expiration) + .is_some_and(|expiration| expiration <= now) + } + pub fn to_url(&self) -> Url { Url::parse(&self.endpoint).unwrap() } @@ -2557,6 +2673,26 @@ mod tests { } } + #[derive(Clone, Debug)] + struct RecordingAuthConnector { + signed_requests: Arc>>, + } + + impl SmithyHttpConnector for RecordingAuthConnector { + fn call(&self, request: HttpRequest) -> HttpConnectorFuture { + let has_expected_token = request.headers().get("x-amz-security-token") == Some("temporary-session-token"); + let has_authorization = request.headers().contains_key("authorization"); + self.signed_requests + .lock() + .expect("recorded auth request lock should not be poisoned") + .push((has_expected_token, has_authorization)); + HttpConnectorFuture::ready(Ok(HttpResponse::new( + aws_smithy_runtime_api::http::StatusCode::try_from(200_u16).expect("200 should be a valid response status"), + SdkBody::empty(), + ))) + } + } + fn recording_target_client() -> (TargetClient, Arc>>) { let request_uris = Arc::new(std::sync::Mutex::new(Vec::new())); let connector = SharedHttpConnector::new(RecordingHttpConnector { @@ -2582,6 +2718,150 @@ mod tests { ) } + #[test] + fn remote_target_sdk_credentials_preserve_temporary_credential_fields() { + let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000); + let expiration = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000); + let credentials = Credentials { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + session_token: Some("temporary-session-token".to_string()), + expiration: Some(jiff::Timestamp::try_from(expiration).expect("test expiration should convert")), + }; + + let sdk_credentials = + remote_target_sdk_credentials(&credentials, "account", now).expect("unexpired temporary credentials should build"); + + assert_eq!(sdk_credentials.session_token(), Some("temporary-session-token")); + assert_eq!(sdk_credentials.expiry(), Some(expiration)); + assert_eq!(sdk_credentials.account_id().map(|id| id.as_str()), Some("account")); + } + + #[test] + fn remote_target_sdk_credentials_normalize_go_zero_expiration() { + let credentials = Credentials { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + session_token: None, + expiration: Some("0001-01-01T00:00:00Z".parse().expect("Go zero time should parse")), + }; + + let sdk_credentials = remote_target_sdk_credentials(&credentials, "", SystemTime::now()) + .expect("Go zero expiration should remain compatible with static credentials"); + + assert!(sdk_credentials.session_token().is_none()); + assert!(sdk_credentials.expiry().is_none()); + } + + #[test] + fn remote_target_sdk_credentials_reject_invalid_expiration_boundaries() { + let expiration = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000); + let mut credentials = Credentials { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + session_token: None, + expiration: Some(jiff::Timestamp::try_from(expiration).expect("test expiration should convert")), + }; + + assert_eq!( + remote_target_sdk_credentials(&credentials, "", SystemTime::UNIX_EPOCH + Duration::from_secs(1_000)) + .expect_err("expiration without a session token must fail"), + "remote target credential expiration requires a session token" + ); + + credentials.session_token = Some("temporary-session-token".to_string()); + assert_eq!( + remote_target_sdk_credentials(&credentials, "", expiration) + .expect_err("credentials expire at the exact expiration boundary"), + EXPIRED_REMOTE_TARGET_CREDENTIALS + ); + } + + #[test] + fn remote_target_credentials_provider_fails_closed_after_expiration() { + let expiration = SystemTime::UNIX_EPOCH + Duration::from_secs(2_000); + let provider = RemoteTargetCredentialsProvider { + credentials: SdkCredentials::new( + "access", + "secret", + Some("temporary-session-token".to_string()), + Some(expiration), + "test", + ), + }; + + assert!(provider.resolve_at(expiration - Duration::from_nanos(1)).is_ok()); + let err = provider + .resolve_at(expiration) + .expect_err("expired credentials must not be returned"); + assert_eq!(err.source().map(ToString::to_string).as_deref(), Some(EXPIRED_REMOTE_TARGET_CREDENTIALS)); + assert!(!format!("{provider:?}").contains("temporary-session-token")); + assert!(!format!("{provider:?}").contains("secret")); + } + + #[test] + fn target_client_detects_expiration_for_cache_refresh() { + let expiration: jiff::Timestamp = "2099-01-01T00:00:00Z".parse().expect("expiration should parse"); + let (mut client, _) = recording_target_client(); + client.credentials = Some(Credentials { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + session_token: Some("temporary-session-token".to_string()), + expiration: Some(expiration), + }); + + assert!(!client.credentials_expired_at("2098-12-31T23:59:59Z".parse().expect("pre-expiration timestamp should parse"))); + assert!(client.credentials_expired_at(expiration)); + + client.credentials.as_mut().expect("credentials should exist").expiration = + Some("0001-01-01T00:00:00Z".parse().expect("Go zero time should parse")); + assert!(!client.credentials_expired_at(jiff::Timestamp::now())); + } + + #[tokio::test] + async fn temporary_credentials_add_security_token_to_sigv4_requests() { + let signed_requests = Arc::new(std::sync::Mutex::new(Vec::new())); + let connector = SharedHttpConnector::new(RecordingAuthConnector { + signed_requests: Arc::clone(&signed_requests), + }); + let http_client = http_client_fn(move |_settings, _components| connector.clone()); + let credentials = Credentials { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + session_token: Some("temporary-session-token".to_string()), + expiration: Some("2099-01-01T00:00:00Z".parse().expect("future expiration should parse")), + }; + let sdk_credentials = remote_target_sdk_credentials(&credentials, "", SystemTime::now()) + .expect("unexpired temporary credentials should build"); + let client = S3Client::from_conf( + S3Config::builder() + .endpoint_url("https://target.example") + .credentials_provider(SharedCredentialsProvider::new(RemoteTargetCredentialsProvider { + credentials: sdk_credentials, + })) + .region(SdkRegion::new("us-east-1")) + .http_client(http_client) + .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest()) + .build(), + ); + + client + .head_bucket() + .bucket("target-bucket") + .send() + .await + .expect("recording connector should accept the signed request"); + + assert_eq!( + signed_requests + .lock() + .expect("recorded auth request lock should not be poisoned") + .as_slice(), + &[(true, true)], + "SigV4 request must include both authorization and the session-token header" + ); + } + fn spawn_https_server(cert: &rcgen::CertifiedKey, requests: usize) -> (u16, std::thread::JoinHandle<()>) { use std::io::{Read, Write}; @@ -3513,6 +3793,29 @@ mod tests { assert!(mutexes.contains_key("second")); } + #[tokio::test] + async fn target_refresh_attempt_updates_retry_timestamp_and_error_count() { + let sys = BucketTargetSys::default(); + + sys.mark_refresh_attempt("arn:reload").await; + let last_refresh = sys.arn_remotes_map.read().await["arn:reload"].last_refresh; + assert!(OffsetDateTime::now_utc() - last_refresh < Duration::from_secs(5)); + + sys.inc_arn_errs("bucket", "arn:reload").await; + sys.inc_arn_errs("bucket", "arn:reload").await; + let errors = sys.arn_errs_map.read().await; + assert_eq!(errors["arn:reload"].count, 2); + assert_eq!(errors["arn:reload"].bucket, "bucket"); + drop(errors); + + sys.mark_refresh_in_progress("bucket", "arn:reload").await; + assert!(sys.is_reloading_target("bucket", "arn:reload").await); + sys.mark_refresh_done("bucket", "arn:reload").await; + assert!(!sys.is_reloading_target("bucket", "arn:reload").await); + sys.mark_refresh_in_progress("bucket", "arn:reload").await; + assert!(sys.is_reloading_target("bucket", "arn:reload").await); + } + #[tokio::test] async fn update_all_targets_publishes_disable_proxy_on_target_client() { // The read-proxy selector (replication_proxy::get_proxy_targets) skips @@ -3551,6 +3854,88 @@ mod tests { assert!(opted_out.disable_proxy, "disable_proxy must reach the published TargetClient"); } + #[tokio::test] + async fn update_all_targets_keeps_failed_client_placeholder() { + let sys = BucketTargetSys::default(); + let target = BucketTarget { + arn: "arn:expired".to_string(), + endpoint: "192.168.1.10:9000".to_string(), + target_bucket: "target-bucket".to_string(), + region: "us-east-1".to_string(), + credentials: Some(Credentials { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + session_token: Some("temporary-session-token".to_string()), + expiration: Some("2000-01-01T00:00:00Z".parse().expect("expired timestamp should parse")), + }), + ..Default::default() + }; + let targets = BucketTargets { targets: vec![target] }; + + sys.update_all_targets("bucket", Some(&targets)).await; + + let remotes = sys.arn_remotes_map.read().await; + let placeholder = remotes + .get("arn:expired") + .expect("configured target should retain a cache entry"); + assert!(placeholder.client.is_none()); + assert!(OffsetDateTime::now_utc() - placeholder.last_refresh < Duration::from_secs(5)); + drop(remotes); + assert!(sys.get_remote_target_client("bucket", "arn:expired").await.is_none()); + } + + #[tokio::test] + async fn credential_rotation_atomically_replaces_published_client() { + let sys = BucketTargetSys::default(); + let target = |session_token: &str| BucketTarget { + arn: "arn:rotating".to_string(), + endpoint: "192.168.1.10:9000".to_string(), + target_bucket: "target-bucket".to_string(), + region: "us-east-1".to_string(), + credentials: Some(Credentials { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + session_token: Some(session_token.to_string()), + expiration: None, + }), + ..Default::default() + }; + + sys.update_all_targets( + "bucket", + Some(&BucketTargets { + targets: vec![target("old-session-token")], + }), + ) + .await; + let old_client = sys + .get_remote_target_client("bucket", "arn:rotating") + .await + .expect("initial client should be published"); + + sys.update_all_targets( + "bucket", + Some(&BucketTargets { + targets: vec![target("new-session-token")], + }), + ) + .await; + let new_client = sys + .get_remote_target_client("bucket", "arn:rotating") + .await + .expect("rotated client should be published"); + + assert!(!Arc::ptr_eq(&old_client, &new_client)); + assert_eq!( + old_client.credentials.as_ref().and_then(Credentials::effective_session_token), + Some("old-session-token") + ); + assert_eq!( + new_client.credentials.as_ref().and_then(Credentials::effective_session_token), + Some("new-session-token") + ); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn target_updates_serialize_client_build_through_publication_per_bucket() { let sys = Arc::new(BucketTargetSys::default()); diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index a60dd018d..091b3a964 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -44,14 +44,14 @@ mod replication_versioning_boundary; mod runtime_boundary; pub use replication_config_boundary::{ - ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, - REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, - REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, - ReplicationConfigurationExt, ReplicationTargetValidationError, assign_site_replication_rule_priorities, - invalid_replication_config_status_field, is_site_replication_role, is_site_replication_rule, - merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id, - replication_target_arns, should_remove_replication_target, site_replication_rule_deployment_id, - unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, + ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, + REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, + REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationTargetValidationError, + assign_site_replication_rule_priorities, invalid_replication_config_status_field, is_site_replication_role, + is_site_replication_rule, merge_incoming_replication_config, merge_user_replication_config, + replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target, + site_replication_rule_deployment_id, unsupported_replication_config_field, validate_replication_config_structure, + validate_replication_config_target_arns, }; pub(crate) use replication_filemeta_boundary::version_purge_statuses_map; pub use replication_filemeta_boundary::{ diff --git a/crates/ecstore/src/bucket/replication/replication_config_boundary.rs b/crates/ecstore/src/bucket/replication/replication_config_boundary.rs index 1eb8c3ad3..fbd9e88b1 100644 --- a/crates/ecstore/src/bucket/replication/replication_config_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_config_boundary.rs @@ -13,12 +13,12 @@ // limitations under the License. pub use rustfs_replication::{ - ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, - REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, - REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, - ReplicationConfigurationExt, ReplicationRuleExt, ReplicationTargetValidationError, assign_site_replication_rule_priorities, - invalid_replication_config_status_field, is_site_replication_role, is_site_replication_rule, - merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id, - replication_target_arns, should_remove_replication_target, site_replication_rule_deployment_id, - unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, + ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, + REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, + REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationRuleExt, + ReplicationTargetValidationError, assign_site_replication_rule_priorities, invalid_replication_config_status_field, + is_site_replication_role, is_site_replication_rule, merge_incoming_replication_config, merge_user_replication_config, + replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target, + site_replication_rule_deployment_id, unsupported_replication_config_field, validate_replication_config_structure, + validate_replication_config_target_arns, }; diff --git a/crates/ecstore/src/bucket/replication/replication_object_config.rs b/crates/ecstore/src/bucket/replication/replication_object_config.rs index 3f44a9d10..e059d39e8 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_config.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_config.rs @@ -436,16 +436,21 @@ pub(crate) async fn check_replicate_delete_strict( } for target in decision.targets_map.values_mut() { - if let Some(client) = ReplicationTargetStore::remote_target_client(bucket, &target.arn).await { - target.synchronous = client.replicate_sync; - } else { - target.replicate = false; - target.synchronous = false; - } + let replicate_sync = ReplicationTargetStore::remote_target_client(bucket, &target.arn) + .await + .map(|client| client.replicate_sync); + apply_target_delivery_mode(target, replicate_sync); } Ok(decision) } +fn apply_target_delivery_mode(target: &mut ReplicateTargetDecision, replicate_sync: Option) { + // A missing runtime client is a delivery failure, not a rule mismatch. + // Preserve admission and fall back to the asynchronous worker, which can + // persist FAILED state for the heal/retry path. + target.synchronous = replicate_sync.unwrap_or(false); +} + pub(crate) fn check_replicate_delete_with_snapshot( dobj: &ObjectToDelete, oi: &ObjectInfo, @@ -629,6 +634,23 @@ mod tests { })); } + #[test] + fn missing_target_client_preserves_delete_admission_as_async() { + let mut target = ReplicateTargetDecision::new("arn:target".to_string(), true, true); + + apply_target_delivery_mode(&mut target, None); + + assert!(target.replicate, "a runtime client miss must not erase the replication rule decision"); + assert!( + !target.synchronous, + "unavailable synchronous targets must fall back to the async retry path" + ); + + apply_target_delivery_mode(&mut target, Some(true)); + assert!(target.replicate); + assert!(target.synchronous); + } + #[test] fn must_replicate_options_preserve_request_flag() { let user_defined = HashMap::new(); diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 3d8008f87..e62a6b04f 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -102,6 +102,7 @@ const BACKGROUND_WALKDIR_TIMEOUT: TokioDuration = TokioDuration::from_secs(60); const ENV_REPL_RESYNC_MAX_JOBS: &str = "RUSTFS_REPL_RESYNC_MAX_JOBS"; const DEFAULT_REPL_RESYNC_MAX_JOBS: usize = 2; const MAX_REPL_RESYNC_MAX_JOBS: usize = 32; +const TARGET_CLIENT_UNAVAILABLE_ERROR: &str = "replication target client is unavailable"; use uuid::Uuid; const EVENT_RESYNC_STATUS_UPDATE_SKIPPED: &str = "replication_resync_status_update_skipped"; @@ -1847,19 +1848,7 @@ pub(crate) async fn replicate_delete_with_outcome( reason = "target_client_missing", "Skipping replication delete because target client is unavailable" ); - send_local_event(EventArgs { - event_name: EventName::ObjectReplicationNotTracked.to_string(), - bucket_name: bucket.clone(), - object: ObjectInfo { - bucket: bucket.clone(), - name: dobj.delete_object.object_name.clone(), - version_id, - delete_marker: dobj.delete_object.delete_marker, - ..Default::default() - }, - user_agent: "Internal: [Replication]".to_string(), - ..Default::default() - }); + rinfos.targets.push(unavailable_delete_target_info(&dobj, &tgt_entry.arn)); continue; }; @@ -2584,6 +2573,32 @@ async fn replicate_force_delete_to_targets(dobj: &Deleted all_succeeded } +fn unavailable_delete_target_info(dobj: &DeletedObjectReplicationInfo, arn: &str) -> ReplicatedTargetInfo { + let mut rinfo = dobj + .delete_object + .replication_state + .as_ref() + .map(|state| state.target_state(arn)) + .unwrap_or_else(|| ReplicatedTargetInfo { + arn: arn.to_string(), + ..Default::default() + }); + rinfo.op_type = dobj.op_type; + if is_version_delete_replication(&dobj.delete_object) { + if rinfo.version_purge_status != VersionPurgeStatusType::Complete { + rinfo.version_purge_status = VersionPurgeStatusType::Failed; + rinfo.error = Some(TARGET_CLIENT_UNAVAILABLE_ERROR.to_string()); + } + } else if rinfo.prev_replication_status == ReplicationStatusType::Completed && dobj.op_type != ReplicationType::ExistingObject + { + rinfo.replication_status = ReplicationStatusType::Completed; + } else { + rinfo.replication_status = ReplicationStatusType::Failed; + rinfo.error = Some(TARGET_CLIENT_UNAVAILABLE_ERROR.to_string()); + } + rinfo +} + async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_client: Arc) -> ReplicatedTargetInfo { let version_id = if let Some(version_id) = &dobj.delete_object.delete_marker_version_id { version_id.to_owned() @@ -2796,6 +2811,10 @@ pub(crate) async fn replicate_object_with_outcome( }; let mut join_set = JoinSet::new(); + let mut rinfos = ReplicatedInfos { + replication_timestamp: Some(OffsetDateTime::now_utc()), + targets: Vec::with_capacity(tgt_arns.len()), + }; for arn in tgt_arns { let Some(tgt_client) = ReplicationTargetStore::remote_target_client(&bucket, &arn).await else { @@ -2803,7 +2822,8 @@ pub(crate) async fn replicate_object_with_outcome( // stays unreachable would flood the log from the replication hot path. The // condition is reported once per pass by the site-replication reconciler and // once per rebuild by `update_all_targets`, which is where an operator can act - // on it; the per-object event below still records each dropped object. + // on it; the FAILED state below preserves retry visibility and the + // aggregate result emits the user-visible failure event once. debug!( event = EVENT_RESYNC_RUNTIME_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -2812,15 +2832,9 @@ pub(crate) async fn replicate_object_with_outcome( object = %object, arn = %arn, reason = "target_client_missing", - "Replication rule has no bucket target for its destination ARN; object not replicated" + "Replication target client unavailable" ); - send_local_event(EventArgs { - event_name: EventName::ObjectReplicationNotTracked.to_string(), - bucket_name: bucket.clone(), - object: roi.to_object_info(), - user_agent: "Internal: [Replication]".to_string(), - ..Default::default() - }); + rinfos.targets.push(unavailable_object_target_info(&roi, &arn)); continue; }; @@ -2835,11 +2849,6 @@ pub(crate) async fn replicate_object_with_outcome( }); } - let mut rinfos = ReplicatedInfos { - replication_timestamp: Some(OffsetDateTime::now_utc()), - targets: Vec::with_capacity(join_set.len()), - }; - while let Some(result) = join_set.join_next().await { match result { Ok(tgt_info) => { @@ -2945,6 +2954,23 @@ pub(crate) async fn replicate_object_with_outcome( (merged_state, state_persisted) } +fn unavailable_object_target_info(roi: &ReplicateObjectInfo, arn: &str) -> ReplicatedTargetInfo { + ReplicatedTargetInfo { + arn: arn.to_string(), + size: roi.actual_size, + replication_action: if roi.op_type == ReplicationType::Object { + ReplicationAction::All + } else { + ReplicationAction::Metadata + }, + op_type: roi.op_type, + replication_status: ReplicationStatusType::Failed, + prev_replication_status: roi.target_replication_status(arn), + error: Some(TARGET_CLIENT_UNAVAILABLE_ERROR.to_string()), + ..Default::default() + } +} + trait ReplicateObjectInfoExt { async fn replicate_object( &self, @@ -4157,6 +4183,88 @@ async fn replicate_multipart_parts_and_complete( #[cfg(test)] mod tests { use super::super::replication_filemeta_boundary::ReplicateTargetDecision; + + #[test] + fn unavailable_object_target_is_persisted_as_failed() { + let arn = "arn:object-target"; + let roi = ReplicateObjectInfo { + actual_size: 42, + op_type: ReplicationType::Object, + replication_status_internal: Some(format!("{arn}=PENDING;")), + ..Default::default() + }; + + let target_info = unavailable_object_target_info(&roi, arn); + let merged = get_replication_state( + &ReplicatedInfos { + replication_timestamp: Some(OffsetDateTime::now_utc()), + targets: vec![target_info.clone()], + }, + &ReplicationState::default(), + None, + ); + + assert_eq!(target_info.replication_status, ReplicationStatusType::Failed); + assert_eq!(target_info.prev_replication_status, ReplicationStatusType::Pending); + assert_eq!(target_info.replication_action, ReplicationAction::All); + assert_eq!(target_info.error.as_deref(), Some(TARGET_CLIENT_UNAVAILABLE_ERROR)); + assert_eq!(merged.targets.get(arn), Some(&ReplicationStatusType::Failed)); + } + + #[test] + fn unavailable_delete_target_is_failed_without_overwriting_completed_state() { + let arn = "arn:delete-target"; + let mut previous_state = ReplicationState::default(); + previous_state.targets.insert(arn.to_string(), ReplicationStatusType::Pending); + let mut dobj = DeletedObjectReplicationInfo { + delete_object: ReplicationDeletedObject { + delete_marker: true, + replication_state: Some(previous_state), + ..Default::default() + }, + op_type: ReplicationType::Delete, + ..Default::default() + }; + + let failed = unavailable_delete_target_info(&dobj, arn); + assert_eq!(failed.replication_status, ReplicationStatusType::Failed); + assert_eq!(failed.prev_replication_status, ReplicationStatusType::Pending); + assert_eq!(failed.error.as_deref(), Some(TARGET_CLIENT_UNAVAILABLE_ERROR)); + + dobj.delete_object + .replication_state + .as_mut() + .expect("previous state should exist") + .targets + .insert(arn.to_string(), ReplicationStatusType::Completed); + let completed = unavailable_delete_target_info(&dobj, arn); + assert_eq!(completed.replication_status, ReplicationStatusType::Completed); + assert!(completed.error.is_none()); + } + + #[test] + fn unavailable_version_purge_target_is_persisted_as_failed() { + let arn = "arn:purge-target"; + let mut previous_state = ReplicationState::default(); + previous_state + .purge_targets + .insert(arn.to_string(), VersionPurgeStatusType::Pending); + let dobj = DeletedObjectReplicationInfo { + delete_object: ReplicationDeletedObject { + version_id: Some(Uuid::new_v4()), + replication_state: Some(previous_state), + ..Default::default() + }, + op_type: ReplicationType::Delete, + ..Default::default() + }; + + let target_info = unavailable_delete_target_info(&dobj, arn); + + assert_eq!(target_info.version_purge_status, VersionPurgeStatusType::Failed); + assert_eq!(target_info.error.as_deref(), Some(TARGET_CLIENT_UNAVAILABLE_ERROR)); + } + fn resync_target_state(resync_id: &str, status: ResyncStatusType, replicated_count: i64) -> TargetReplicationResyncStatus { TargetReplicationResyncStatus { resync_id: resync_id.to_string(), diff --git a/crates/ecstore/src/bucket/target/bucket_target.rs b/crates/ecstore/src/bucket/target/bucket_target.rs index 0f5e3f62b..bfb6af305 100644 --- a/crates/ecstore/src/bucket/target/bucket_target.rs +++ b/crates/ecstore/src/bucket/target/bucket_target.rs @@ -25,6 +25,8 @@ use time::OffsetDateTime; use url::Url; const REDACTED_CREDENTIAL: &str = ""; +const GO_YEAR_ONE_START_UNIX_SECONDS: i64 = -62_135_596_800; +const GO_YEAR_TWO_START_UNIX_SECONDS: i64 = -62_104_060_800; #[derive(Deserialize, Serialize, Default, Clone)] pub struct Credentials { @@ -41,6 +43,26 @@ pub struct Credentials { } impl Credentials { + /// Returns the session token used for request signing. + /// + /// MinIO-compatible payloads may carry an empty token. Treat whitespace-only + /// values as absent without rewriting a real token, whose bytes are opaque. + pub fn effective_session_token(&self) -> Option<&str> { + self.session_token.as_deref().filter(|token| !token.trim().is_empty()) + } + + /// Returns the credential expiry after normalizing Go's zero `time.Time`. + /// + /// Go JSON encoders emit year 1 for an unset `time.Time`; persisted MinIO + /// target metadata can therefore contain that sentinel even for static + /// credentials. + pub fn effective_expiration(&self) -> Option { + self.expiration.filter(|expiration| { + let unix_seconds = expiration.as_second(); + !(GO_YEAR_ONE_START_UNIX_SECONDS..GO_YEAR_TWO_START_UNIX_SECONDS).contains(&unix_seconds) + }) + } + pub fn redacted(&self) -> Self { Self { access_key: self.access_key.clone(), @@ -355,6 +377,24 @@ mod tests { use std::time::Duration; use time::OffsetDateTime; + #[test] + fn credential_effective_values_normalize_only_compatibility_sentinels() { + let mut credentials = Credentials { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + session_token: Some(" ".to_string()), + expiration: Some("0001-01-01T08:00:00+08:00".parse().expect("Go zero time should parse")), + }; + + assert!(credentials.effective_session_token().is_none()); + assert!(credentials.effective_expiration().is_none()); + + credentials.session_token = Some(" opaque token ".to_string()); + credentials.expiration = Some("2099-01-01T00:00:00Z".parse().expect("future timestamp should parse")); + assert_eq!(credentials.effective_session_token(), Some(" opaque token ")); + assert_eq!(credentials.effective_expiration(), credentials.expiration); + } + #[test] fn test_bucket_target_json_deserialize() { let json = r#" diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index 574d6c49c..c85a92d44 100644 --- a/crates/replication/src/config.rs +++ b/crates/replication/src/config.rs @@ -60,9 +60,7 @@ pub const REPLICATION_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &[ "Destination.ReplicationTime", ]; -// v3: temporary-credential fields are advertised as read-only historical -// metadata. They remain decodable for MinIO and persisted-data compatibility, -// but set-remote-target rejects them until refresh and rotation are supported. +// v3: remote targets accept temporary credential session tokens and expiry. pub const REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION: u32 = 3; pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[ @@ -70,6 +68,8 @@ pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[ "endpoint", "credentials.accessKey", "credentials.secretKey", + "credentials.sessionToken", + "credentials.expiration", "targetbucket", "secure", "path", @@ -91,8 +91,6 @@ pub const REMOTE_TARGET_WRITABLE_FIELDS: &[&str] = &[ "disableProxy", ]; -pub const REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS: &[&str] = &["credentials.sessionToken", "credentials.expiration"]; - pub const REMOTE_TARGET_UNSUPPORTED_FIELDS: &[&str] = &["edge", "edgeSyncBeforeExpiry"]; #[derive(Debug, Clone, Serialize, Deserialize, Default)] diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index e9ab2801d..4e32b8a3e 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -29,15 +29,14 @@ mod storage_api; pub mod tagging; pub use config::{ - ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, - REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, - REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, - ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns, - assign_site_replication_rule_priorities, invalid_replication_config_status_field, is_reconciler_owned_site_replication_rule, - is_site_replication_role, is_site_replication_rule, merge_incoming_replication_config, merge_user_replication_config, - replication_target_arn_deployment_id, replication_target_arns, should_remove_replication_target, - site_replication_rule_deployment_id, unsupported_replication_config_field, validate_replication_config_structure, - validate_replication_config_target_arns, + ObjectOpts, OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, + REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, + REPLICATION_WRITABLE_FIELDS, ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationTargetValidationError, + active_replication_rule_destination_arns, assign_site_replication_rule_priorities, invalid_replication_config_status_field, + is_reconciler_owned_site_replication_rule, is_site_replication_role, is_site_replication_rule, + merge_incoming_replication_config, merge_user_replication_config, replication_target_arn_deployment_id, + replication_target_arns, should_remove_replication_target, site_replication_rule_deployment_id, + unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, }; pub use delete::{ DeletedObjectReplicationInfo, delete_marker_purge_mrf_entry, delete_marker_purge_version_id, diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index ed35d614f..957a92816 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -23,9 +23,9 @@ use crate::admin::storage_api::bucket::metadata::BUCKET_TARGETS_FILE; use crate::admin::storage_api::bucket::metadata_sys; use crate::admin::storage_api::bucket::metadata_sys::get_replication_config; use crate::admin::storage_api::bucket::replication::REMOTE_TARGET_UNSUPPORTED_FIELDS; -use crate::admin::storage_api::bucket::replication::{BucketStats, ReplicationStatusType}; #[cfg(test)] -use crate::admin::storage_api::bucket::replication::{REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS}; +use crate::admin::storage_api::bucket::replication::REMOTE_TARGET_WRITABLE_FIELDS; +use crate::admin::storage_api::bucket::replication::{BucketStats, ReplicationStatusType}; use crate::admin::storage_api::bucket::target::{ BucketTarget, BucketTargetType, Credentials as TargetCredentials, LatencyStat, duration_from_secs_or_nanos, }; @@ -57,13 +57,6 @@ use url::Host; const SUPPORTED_REMOTE_TARGET_API: &str = "s3v4"; -/// Go encodes the zero `time.Time` as the year-1 instant -/// (`0001-01-01T00:00:00Z`, possibly re-encoded with an offset); no real -/// credential expiry lives in year 1, so any such timestamp means "unset". -fn is_go_zero_time(timestamp: Timestamp) -> bool { - timestamp.to_zoned(jiff::tz::TimeZone::UTC).year() == 1 -} - /// Field groups a `set-remote-target?update=true` request may modify, mirroring /// MinIO's `TargetUpdateType` / `GetTargetUpdateOps` query contract: the update /// overlays only the requested groups onto the stored target, so a client can @@ -250,7 +243,7 @@ impl RemoteTargetRequest { fn into_bucket_target(self) -> S3Result { self.validate_connection_fields()?; - self.into_bucket_target_common() + self.into_bucket_target_common(true) } /// Partial-update parse: only the field groups named by `ops` are validated; @@ -259,43 +252,18 @@ impl RemoteTargetRequest { if self.arn.trim().is_empty() { return Err(s3_error!(InvalidRequest, "arn is required for update")); } - if ops.contains(&TargetUpdateOp::Credentials) { + let replacing_credentials = ops.contains(&TargetUpdateOp::Credentials); + if replacing_credentials { self.validate_connection_fields()?; } - self.into_bucket_target_common() + self.into_bucket_target_common(replacing_credentials) } - fn into_bucket_target_common(self) -> S3Result { + fn into_bucket_target_common(self, validate_credentials: bool) -> S3Result { if !self.target_type.is_valid() { return Err(s3_error!(InvalidRequest, "type is invalid")); } - if self - .credentials - .session_token - .as_deref() - .is_some_and(|token| !token.trim().is_empty()) - { - return Err(s3_error!( - InvalidRequest, - "remote target field credentials.session_token is not supported by this RustFS version" - )); - } - - // Go's `omitempty` never elides a zero `time.Time`, so every madmin - // marshal carries `"expiration":"0001-01-01T00:00:00Z"`; only a real - // (non-year-1) expiry means the client wants expiring credentials. - if self - .credentials - .expiration - .is_some_and(|expiration| !is_go_zero_time(expiration)) - { - return Err(s3_error!( - InvalidRequest, - "remote target field credentials.expiration is not supported by this RustFS version" - )); - } - if !self.api.is_empty() && self.api != SUPPORTED_REMOTE_TARGET_API { return Err(s3_error!( InvalidRequest, @@ -317,9 +285,17 @@ impl RemoteTargetRequest { } let mut credentials = TargetCredentials::from(self.credentials); - // Past the check above the expiration can only be the zero-value - // sentinel, i.e. "no expiration" — never persist it. - credentials.expiration = None; + credentials.expiration = credentials.effective_expiration(); + if validate_credentials && credentials.expiration.is_some() && credentials.effective_session_token().is_none() { + return Err(s3_error!(InvalidRequest, "credentials.expiration requires credentials.session_token")); + } + if validate_credentials + && credentials + .expiration + .is_some_and(|expiration| expiration <= Timestamp::now()) + { + return Err(s3_error!(InvalidRequest, "credentials.expiration must be in the future")); + } Ok(BucketTarget { source_bucket: self.source_bucket, @@ -1504,10 +1480,10 @@ impl Operation for ReplicationMrfHandler { #[cfg(test)] mod tests { use super::{ - REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, - RemoteTargetCredentialsRequest, RemoteTargetRequest, ReplicationDiffEntry, SUPPORTED_REMOTE_TARGET_API, TargetUpdateOp, - build_mrf_response, extract_query_params, parse_remote_target_update_ops, render_mrf_backlog, render_replication_diff, - unique_replication_peers, validate_remote_target_tls_settings, + REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, RemoteTargetCredentialsRequest, RemoteTargetRequest, + ReplicationDiffEntry, SUPPORTED_REMOTE_TARGET_API, TargetUpdateOp, build_mrf_response, extract_query_params, + parse_remote_target_update_ops, render_mrf_backlog, render_replication_diff, unique_replication_peers, + validate_remote_target_tls_settings, }; use crate::admin::storage_api::bucket::target::{BucketTarget, Credentials as TargetCredentials, LatencyStat}; use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry}; @@ -2104,27 +2080,13 @@ mod tests { #[test] fn remote_target_request_rejects_unimplemented_fields() { - for (field, value, historical_field) in [ - ( - "credentials.session_token", - serde_json::json!("session-token"), - Some("credentials.sessionToken"), - ), - ( - "credentials.expiration", - serde_json::json!("2026-01-01T00:00:00Z"), - Some("credentials.expiration"), - ), - ("api", serde_json::json!("s3v2"), None), - ("edge", serde_json::json!(true), None), - ("edgeSyncBeforeExpiry", serde_json::json!(true), None), + for (field, value) in [ + ("api", serde_json::json!("s3v2")), + ("edge", serde_json::json!(true)), + ("edgeSyncBeforeExpiry", serde_json::json!(true)), ] { let mut request = valid_remote_target_request(); - if let Some((credential_field, credential_name)) = field.split_once('.') { - request[credential_field][credential_name] = value; - } else { - request[field] = value; - } + request[field] = value; let request: RemoteTargetRequest = serde_json::from_value(request).expect("unsupported field should still deserialize"); let err = request @@ -2133,15 +2095,86 @@ mod tests { assert!(err.to_string().contains(field)); assert!(err.to_string().contains("not supported by this RustFS version")); - if let Some(historical_field) = historical_field { - assert!( - REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS.contains(&historical_field), - "rejected field {field} must be advertised as historical-only" - ); - } } } + #[test] + fn remote_target_request_accepts_temporary_credentials() { + let mut request = valid_remote_target_request(); + request["credentials"]["sessionToken"] = serde_json::json!("session-token"); + request["credentials"]["expiration"] = serde_json::json!("2099-01-01T00:00:00Z"); + + let target = serde_json::from_value::(request) + .expect("temporary credentials should deserialize") + .into_bucket_target() + .expect("unexpired temporary credentials should be accepted"); + let credentials = target.credentials.expect("credentials should be preserved"); + + assert_eq!(credentials.session_token.as_deref(), Some("session-token")); + assert_eq!( + serde_json::to_value(credentials.expiration.expect("expiration should be preserved")) + .expect("expiration should serialize"), + serde_json::json!("2099-01-01T00:00:00Z") + ); + for field in ["credentials.sessionToken", "credentials.expiration"] { + assert!(REMOTE_TARGET_WRITABLE_FIELDS.contains(&field)); + } + } + + #[test] + fn remote_target_request_accepts_session_token_without_expiration() { + let mut request = valid_remote_target_request(); + request["credentials"]["session_token"] = serde_json::json!("session-token"); + + let target = serde_json::from_value::(request) + .expect("temporary credentials should deserialize") + .into_bucket_target() + .expect("a session token without a reported expiration should remain compatible"); + + assert_eq!( + target + .credentials + .as_ref() + .and_then(|credentials| credentials.session_token.as_deref()), + Some("session-token") + ); + assert!(target.credentials.and_then(|credentials| credentials.expiration).is_none()); + } + + #[test] + fn remote_target_request_rejects_expiration_without_session_token() { + let mut request = valid_remote_target_request(); + request["credentials"]["expiration"] = serde_json::json!("2099-01-01T00:00:00Z"); + + let err = serde_json::from_value::(request) + .expect("request should deserialize") + .into_bucket_target() + .expect_err("an expiring credential bundle requires a session token"); + + assert!( + err.to_string() + .contains("credentials.expiration requires credentials.session_token") + ); + } + + #[test] + fn remote_target_request_rejects_expired_temporary_credentials_without_leaking_secrets() { + let mut request = valid_remote_target_request(); + request["credentials"]["secretKey"] = serde_json::json!("secret-must-not-leak"); + request["credentials"]["sessionToken"] = serde_json::json!("session-token-must-not-leak"); + request["credentials"]["expiration"] = serde_json::json!("2000-01-01T00:00:00Z"); + + let err = serde_json::from_value::(request) + .expect("request should deserialize") + .into_bucket_target() + .expect_err("expired credentials must fail before persistence"); + let message = err.to_string(); + + assert!(message.contains("credentials.expiration must be in the future")); + assert!(!message.contains("secret-must-not-leak")); + assert!(!message.contains("session-token-must-not-leak")); + } + #[test] fn remote_target_request_accepts_real_madmin_add_marshal() { let request: RemoteTargetRequest = @@ -2192,6 +2225,32 @@ mod tests { ); } + #[test] + fn non_credential_update_accepts_redacted_temporary_credential_round_trip() { + // list-remote-targets redacts both the secret and session token but + // intentionally retains the non-secret expiration. mc echoes that + // shape on a sync-only update; the credentials group is not applied. + let body = serde_json::json!({ + "endpoint": "192.168.1.10:9000", + "credentials": { + "accessKey": "access", + "expiration": "2099-01-01T00:00:00Z" + }, + "targetbucket": "target", + "arn": "arn:rustfs:replication:us-east-1:dep:target", + "type": "replication", + "replicationSync": true + }); + + let target = serde_json::from_value::(body) + .expect("redacted mc round-trip should deserialize") + .into_update_bucket_target(&[TargetUpdateOp::Sync]) + .expect("a sync-only update must ignore the redacted credential group"); + + assert!(target.replication_sync); + assert!(target.credentials.and_then(|credentials| credentials.expiration).is_some()); + } + #[test] fn remote_target_request_ignores_client_supplied_latency() { // Latency is a server-measured runtime stat; mc echoes the @@ -2238,22 +2297,6 @@ mod tests { assert!(!target.ca_cert_pem.is_empty()); } - #[test] - fn remote_target_request_validation_does_not_echo_credential_values() { - let mut request = valid_remote_target_request(); - request["credentials"]["session_token"] = serde_json::json!("session-token-must-not-leak"); - - let request: RemoteTargetRequest = serde_json::from_value(request).expect("request should deserialize"); - let err = request - .into_bucket_target() - .expect_err("session tokens must be rejected before persistence"); - let message = err.to_string(); - - assert!(message.contains("credentials.session_token")); - assert!(!message.contains("session-token-must-not-leak")); - assert!(!message.contains("secret")); - } - #[test] fn remote_target_request_accepts_go_duration_wire_values() { // `mc replicate add` defaults `--healthcheck-seconds` to 60; madmin @@ -2514,17 +2557,6 @@ mod tests { #[test] fn remote_target_capability_fields_do_not_overlap() { - for field in REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS { - assert!( - !REMOTE_TARGET_WRITABLE_FIELDS.contains(field), - "remote target field {field} cannot be both historical-only and writable" - ); - assert!( - !REMOTE_TARGET_UNSUPPORTED_FIELDS.contains(field), - "remote target field {field} cannot be both historical-only and unsupported" - ); - } - for field in REMOTE_TARGET_UNSUPPORTED_FIELDS { assert!( !REMOTE_TARGET_WRITABLE_FIELDS.contains(field), diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index 4b134d2e5..82df34f05 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -24,9 +24,8 @@ use crate::admin::runtime_sources::{ DefaultAdminUsecase, QueryServerInfoRequest, current_endpoints_handle, default_admin_usecase, object_store_from_req, }; use crate::admin::storage_api::bucket::replication::{ - REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, REMOTE_TARGET_UNSUPPORTED_FIELDS, - REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, - REPLICATION_WRITABLE_FIELDS, + REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, + REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, }; use crate::admin::storage_api::cluster::{ CapabilityState, CapabilityStatus, ObservabilitySnapshotProvider, TopologySnapshot, TopologySnapshotProvider, @@ -730,15 +729,6 @@ impl ReplicationCapabilities { name, state: ReplicationFieldState::Supported, }) - .chain( - REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS - .iter() - .copied() - .map(|name| ReplicationFieldCapability { - name, - state: ReplicationFieldState::ReadOnlyHistorical, - }), - ) .chain( REMOTE_TARGET_UNSUPPORTED_FIELDS .iter() @@ -1306,7 +1296,7 @@ mod tests { assert_eq!(response.summary.manual_transition_jobs.state, CapabilityState::Supported); assert_eq!(response.replication.contract_version, 1); assert_eq!(response.replication.bucket_replication.contract_version, 1); - // v3: temporary-credential fields are explicitly historical-only. + // v3: temporary-credential fields are writable and used for signing. assert_eq!(response.replication.remote_targets.contract_version, 3); assert_eq!(response.replication.bucket_replication.status.state, CapabilityState::Supported); assert_eq!(response.replication.remote_targets.status.state, CapabilityState::Supported); @@ -1363,8 +1353,8 @@ mod tests { .remote_targets .fields .iter() - .any(|field| field.name == name && field.state == super::ReplicationFieldState::ReadOnlyHistorical), - "remote target field {name} must be advertised as historical-only" + .any(|field| field.name == name && field.state == super::ReplicationFieldState::Supported), + "remote target field {name} must be advertised as writable" ); } assert_eq!(response.manual_transition_jobs.contract_version, 1); @@ -1469,8 +1459,8 @@ mod tests { .as_array() .expect("remote target fields should be an array") .iter() - .any(|field| field["name"] == name && field["state"] == "read_only_historical"), - "serialized remote target field {name} must be historical-only" + .any(|field| field["name"] == name && field["state"] == "supported"), + "serialized remote target field {name} must be writable" ); } assert_eq!(value["manual_transition_jobs"]["contract_version"], 1); diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 5ea5b3f74..1c8a14c26 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -443,10 +443,10 @@ pub(crate) mod quota { pub(crate) mod replication { pub(crate) use super::ecstore_bucket::replication::{ - OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_READ_ONLY_HISTORICAL_FIELDS, - REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, - REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, assign_site_replication_rule_priorities, - merge_incoming_replication_config, replication_target_arn_deployment_id, + OperatorRuleContract, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, + REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, + REPLICATION_WRITABLE_FIELDS, assign_site_replication_rule_priorities, merge_incoming_replication_config, + replication_target_arn_deployment_id, }; pub(crate) type BucketReplicationResyncStatus = super::ecstore_bucket::replication::BucketReplicationResyncStatus; pub(crate) type BucketStats = super::ecstore_bucket::replication::BucketStats;