diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 51c7dda99..eca73bb40 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -66,8 +66,8 @@ use rustfs_madmin::{ ReplicateRemoveStatus, ResyncBucketStatus, SITE_REPL_API_VERSION, SR_IAM_ITEM_STS_ACC, SR_IAM_ITEM_STS_ACC_LEGACY, SRBucketInfo, SRBucketMeta, SRBucketStatsSummary, SRGroupInfo, SRGroupStatsSummary, SRIAMItem, SRIAMUser, SRILMExpiryStatsSummary, SRInfo, SRMetric, SRMetricsSummary, SRPeerError, SRPeerJoinReq, SRPendingOperation, SRPolicyMapping, - SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRSTSCredential, SRSessionPolicy, SRSiteSummary, SRStateEditReq, - SRStateInfo, SRStatusInfo, SRSvcAccChange, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat, + SRPolicyStatsSummary, SRRemoveReq, SRResyncOpStatus, SRSTSCredential, SRSiteSummary, SRStateEditReq, SRStateInfo, + SRStatusInfo, SRSvcAccChange, SRSvcAccCreate, SRUserStatsSummary, SiteReplicationInfo, SyncStatus, WorkerStat, }; use rustfs_policy::policy::{ Policy, @@ -98,7 +98,6 @@ use uuid::Uuid; // paths keep resolving while this file keeps only the HTTP handlers. pub(crate) use crate::site_replication::*; -const SERVICE_ACCOUNT_ENVELOPE_VERSION: u64 = 2; // Serializes peer-join admission (staleness check -> IAM upsert -> state // commit) across every node of this site; see admit_peer_join. Never an // actual object — only a namespace-lock key, like the repair execution lock. @@ -1980,7 +1979,7 @@ async fn bootstrap_existing_metadata_after_add( return errors; } }; - let plan = match site_replication_bootstrap_plan(&info) { + let plan = match build_site_replication_bootstrap_plan(&info).await { Ok(plan) => plan, Err(err) => { let mut errors = SiteReplicationErrorSummary::default(); @@ -5660,41 +5659,6 @@ fn group_info_requires_upsert(update: &rustfs_madmin::GroupAddRemove) -> bool { !update.is_remove } -pub(crate) fn encode_service_account_replication_policy( - claims: &HashMap, - session_policy: Option<&str>, -) -> S3Result<(SRSessionPolicy, Option)> { - if !claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) { - return session_policy - .map(SRSessionPolicy::from_json) - .transpose() - .map(|policy| policy.unwrap_or_default()) - .map(|policy| (policy, None)) - .map_err(|err| s3_error!(InvalidArgument, "marshal policy failed: {:?}", err)); - } - - let policy = match session_policy { - Some(policy) => serde_json::from_str::(policy) - .map_err(|err| s3_error!(InvalidArgument, "invalid service account replication policy: {:?}", err))?, - None => Policy::default(), - }; - if policy.statements.is_empty() && (!policy.id.is_empty() || !policy.version.is_empty()) - || policy.version.is_empty() && !policy.statements.is_empty() - { - return Err(s3_error!(InvalidArgument, "service account replication policy is not normalized")); - } - let policy = serde_json::to_string(&policy) - .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?; - let policy = SRSessionPolicy::from_json(&policy) - .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?; - Ok(( - policy, - Some(rustfs_madmin::SRSvcAccReplicationEnvelope { - version: SERVICE_ACCOUNT_ENVELOPE_VERSION, - }), - )) -} - #[derive(Debug)] struct ReplicatedServiceAccountPolicy { policy: Option, @@ -5979,6 +5943,8 @@ async fn apply_iam_service_account_item( .map_err(ApiError::from)?; } Err(err) if is_err_no_such_service_account(&err) => { + let access_key = create.access_key.clone(); + let status = create.status.clone(); iam_sys .new_service_account( &create.parent, @@ -5996,6 +5962,28 @@ async fn apply_iam_service_account_item( ) .await .map_err(ApiError::from)?; + // A snapshot (bootstrap / repair / retry resend) carries the + // account's current status; creation always enables, so a + // disabled account must be switched off in a second step or + // the peer keeps accepting credentials the source rejects. + if !status.is_empty() && status != "on" { + iam_sys + .update_service_account( + &access_key, + UpdateServiceAccountOpts { + session_policy: None, + secret_key: None, + name: None, + description: None, + expiration: None, + status: Some(status), + parent_user: None, + allow_site_replicator_account: false, + }, + ) + .await + .map_err(ApiError::from)?; + } } Err(err) => return Err(ApiError::from(err).into()), } @@ -7692,7 +7680,7 @@ impl Operation for SiteReplicationRepairHandler { let local_peer = current_local_peer(&req, &state); let body: SiteReplicationRepairRequest = read_site_replication_json(req, "", false).await?; let info = build_sr_info(&state, &local_peer).await?; - let plan = site_replication_bootstrap_plan(&info)?; + let plan = build_site_replication_bootstrap_plan(&info).await?; let signing_key = current_token_signing_key().ok_or_else(|| { S3Error::with_message(S3ErrorCode::InternalError, "token signing key is not initialized".to_string()) })?; @@ -7885,6 +7873,7 @@ impl Operation for SRRotateServiceAccountHandler { mod tests { use super::*; use crate::site_replication::identity::deployment_id_for_endpoint; + use rustfs_madmin::SRSessionPolicy; /// A peer the status probe could not reach must render as offline. /// diff --git a/rustfs/src/site_replication/hooks.rs b/rustfs/src/site_replication/hooks.rs index 95a830500..2651eecd3 100644 --- a/rustfs/src/site_replication/hooks.rs +++ b/rustfs/src/site_replication/hooks.rs @@ -302,7 +302,164 @@ pub(crate) fn site_replication_state_replicates_ilm_expiry(state: &SiteReplicati state.peers.values().any(|peer| peer.replicate_ilm_expiry) } -pub(crate) fn site_replication_bootstrap_plan(info: &SRInfo) -> S3Result { +/// Secret-bearing half of the IAM snapshot. `SRInfo` is served to admin +/// callers (`site-replication/info`, status, add preflight) and must stay +/// secret-free, so the bootstrap plan receives credentials through this +/// separate value, built only on the paths that deliver to peers (site add +/// bootstrap, repair, retry snapshot resend). Never persisted, never served. +#[derive(Debug, Clone, Default)] +pub(crate) struct SiteReplicationIamCredentials { + /// Built-in users (access key -> credential); temp and service accounts + /// are excluded, external/IdP users never appear here. + pub(crate) users: BTreeMap, + /// Every service account except the site replicator's own, already + /// shaped as the `service-account` create item the live hook emits. + pub(crate) service_accounts: Vec, +} + +#[derive(Debug, Clone)] +pub(crate) struct SiteReplicationUserCredential { + pub(crate) secret_key: String, + pub(crate) status: AccountStatus, + /// The user record's own update time (the axis the receiver's staleness + /// check compares against), unlike `UserInfo::updated_at` which + /// `list_users` overwrites with the policy mapping's time. + pub(crate) updated_at: Option, +} + +#[derive(Debug, Clone)] +pub(crate) struct SiteReplicationServiceAccountSnapshot { + pub(crate) create: SRSvcAccCreate, + pub(crate) envelope: Option, + pub(crate) updated_at: Option, +} + +pub(crate) const SERVICE_ACCOUNT_ENVELOPE_VERSION: u64 = 2; + +pub(crate) fn encode_service_account_replication_policy( + claims: &HashMap, + session_policy: Option<&str>, +) -> S3Result<(SRSessionPolicy, Option)> { + if !claims.contains_key(OIDC_VIRTUAL_PARENT_CLAIM) { + return session_policy + .map(SRSessionPolicy::from_json) + .transpose() + .map(|policy| policy.unwrap_or_default()) + .map(|policy| (policy, None)) + .map_err(|err| s3_error!(InvalidArgument, "marshal policy failed: {:?}", err)); + } + + let policy = match session_policy { + Some(policy) => serde_json::from_str::(policy) + .map_err(|err| s3_error!(InvalidArgument, "invalid service account replication policy: {:?}", err))?, + None => Policy::default(), + }; + if policy.statements.is_empty() && (!policy.id.is_empty() || !policy.version.is_empty()) + || policy.version.is_empty() && !policy.statements.is_empty() + { + return Err(s3_error!(InvalidArgument, "service account replication policy is not normalized")); + } + let policy = serde_json::to_string(&policy) + .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?; + let policy = SRSessionPolicy::from_json(&policy) + .map_err(|err| s3_error!(InternalError, "marshal service account replication policy failed: {:?}", err))?; + Ok(( + policy, + Some(SRSvcAccReplicationEnvelope { + version: SERVICE_ACCOUNT_ENVELOPE_VERSION, + }), + )) +} + +/// Read the credentials the IAM snapshot needs straight from the IAM store: +/// `list_users` deliberately strips secret keys and skips service accounts, +/// which is right for an admin listing and wrong for a peer snapshot (the +/// plan builder used to drop every user for lack of a secret, so a status +/// change or secret rotation committed while a peer was unreachable never +/// reached it — backlog#2289). +pub(crate) async fn build_sr_iam_credentials() -> S3Result { + let mut credentials = SiteReplicationIamCredentials::default(); + let Some(iam_sys) = current_iam_handle() else { + return Ok(credentials); + }; + + let mut users = HashMap::new(); + iam_sys.load_users(UserType::Reg, &mut users).await.map_err(ApiError::from)?; + for (access_key, identity) in users { + if identity.credentials.is_temp() || identity.credentials.is_service_account() { + continue; + } + credentials.users.insert( + access_key, + SiteReplicationUserCredential { + secret_key: identity.credentials.secret_key, + status: if identity.credentials.status == "off" { + AccountStatus::Disabled + } else { + AccountStatus::Enabled + }, + updated_at: identity.update_at, + }, + ); + } + + let mut service_accounts = HashMap::new(); + iam_sys + .load_users(UserType::Svc, &mut service_accounts) + .await + .map_err(ApiError::from)?; + let mut service_accounts: Vec<_> = service_accounts.into_iter().collect(); + service_accounts.sort_by(|(a, _), (b, _)| a.cmp(b)); + for (access_key, identity) in service_accounts { + // The replicator account is installed by join / rotate, never by a snapshot. + if access_key == SITE_REPLICATOR_SERVICE_ACCOUNT || !identity.credentials.is_service_account() { + continue; + } + let claims = iam_sys.get_claims_for_svc_acc(&access_key).await.map_err(ApiError::from)?; + let (account, session_policy) = iam_sys.get_service_account(&access_key).await.map_err(ApiError::from)?; + let session_policy = session_policy + .map(|policy| serde_json::to_string(&policy)) + .transpose() + .map_err(|err| { + S3Error::with_message( + S3ErrorCode::InternalError, + format!("marshal service account session policy failed: {err:?}"), + ) + })?; + let (session_policy, envelope) = encode_service_account_replication_policy(&claims, session_policy.as_deref())?; + credentials.service_accounts.push(SiteReplicationServiceAccountSnapshot { + create: SRSvcAccCreate { + parent: identity.credentials.parent_user, + access_key, + secret_key: identity.credentials.secret_key, + groups: identity.credentials.groups.unwrap_or_default(), + claims, + session_policy, + status: identity.credentials.status, + name: account.name.unwrap_or_default(), + description: account.description.unwrap_or_default(), + expiration: account.expiration, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }, + envelope, + updated_at: identity.update_at, + }); + } + + Ok(credentials) +} + +/// The bootstrap plan for peer delivery: `info` (secret-free) plus the IAM +/// credentials read at this moment. +pub(crate) async fn build_site_replication_bootstrap_plan(info: &SRInfo) -> S3Result { + let credentials = build_sr_iam_credentials().await?; + site_replication_bootstrap_plan(info, &credentials) +} + +pub(crate) fn site_replication_bootstrap_plan( + info: &SRInfo, + credentials: &SiteReplicationIamCredentials, +) -> S3Result { let mut plan = SiteReplicationBootstrapPlan::default(); let replicate_ilm_expiry = site_replication_info_replicates_ilm_expiry(info); @@ -318,24 +475,57 @@ pub(crate) fn site_replication_bootstrap_plan(info: &SRInfo) -> S3Result Option { @@ -972,6 +973,11 @@ pub(crate) fn iam_snapshot_key(item: &SRIAMItem) -> Option { user_type: mapping.user_type, is_group: mapping.is_group, }), + "service-account" => item + .svc_acc_change + .as_ref() + .and_then(|change| change.create.as_ref()) + .map(|create| IamSnapshotKey::ServiceAccount(create.access_key.clone())), _ => None, } } @@ -1006,6 +1012,24 @@ pub(crate) fn iam_snapshot_tombstones(item: &SRIAMItem, observed_at: OffsetDateT mapping.policy.clear(); } } + "service-account" => { + let Some(access_key) = item + .svc_acc_change + .as_ref() + .and_then(|change| change.create.as_ref()) + .map(|create| create.access_key.clone()) + else { + return Vec::new(); + }; + tombstone.svc_acc_change = Some(SRSvcAccChange { + delete: Some(SRSvcAccDelete { + access_key, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }), + api_version: Some(SITE_REPL_API_VERSION.to_string()), + ..Default::default() + }); + } _ => return Vec::new(), } vec![tombstone] @@ -1701,7 +1725,7 @@ pub(crate) async fn drain_site_replication_retry_queue_locked( // tick and only when a snapshot resend is actually due. let plan = if needs_plan { let info = build_sr_info(&runtime.state, &runtime.local_peer).await?; - Some(site_replication_bootstrap_plan(&info)?) + Some(build_site_replication_bootstrap_plan(&info).await?) } else { None }; @@ -1841,7 +1865,7 @@ pub(crate) async fn drain_one_site_replication_retry_event( } } let fresh_info = build_sr_info(&runtime.state, &runtime.local_peer).await?; - let fresh_plan = site_replication_bootstrap_plan(&fresh_info)?; + let fresh_plan = build_site_replication_bootstrap_plan(&fresh_info).await?; let fresh_snapshot = RetrySnapshot::from_plan(&action, &fresh_plan).expect("snapshot action has a snapshot"); if fresh_snapshot.fingerprint()? == current_fingerprint { if is_iam { diff --git a/rustfs/src/site_replication/tests.rs b/rustfs/src/site_replication/tests.rs index 3b2c48666..a9491e1b9 100644 --- a/rustfs/src/site_replication/tests.rs +++ b/rustfs/src/site_replication/tests.rs @@ -1679,7 +1679,8 @@ fn test_site_replication_bootstrap_plan_includes_replayable_snapshot_items() { }, ); - let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build"); + let plan = + site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build"); assert_eq!(plan.iam_items.iter().map(|item| item.r#type.as_str()).collect::>(), { vec!["policy", "iam-user", "group-info", "policy-mapping"] @@ -1717,7 +1718,8 @@ fn test_site_replication_bootstrap_plan_skips_lifecycle_by_default() { }, ); - let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build"); + let plan = + site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build"); assert!(!plan.bucket_items.iter().any(|item| item.r#type == "lc-config")); } @@ -1748,7 +1750,8 @@ fn test_site_replication_bootstrap_plan_emits_timestamped_lifecycle_delete() { }, ); - let plan = site_replication_bootstrap_plan(&info).expect("bootstrap plan should build"); + let plan = + site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("bootstrap plan should build"); let item = plan .bucket_items @@ -1935,8 +1938,8 @@ fn test_site_replication_repair_preflight_token_is_deterministic_for_equal_state }, ); - let plan_a = site_replication_bootstrap_plan(&info).expect("first plan"); - let plan_b = site_replication_bootstrap_plan(&info).expect("second plan"); + let plan_a = site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("first plan"); + let plan_b = site_replication_bootstrap_plan(&info, &SiteReplicationIamCredentials::default()).expect("second plan"); let token_a = site_replication_repair_preflight_token(&state, &plan_a, b"test-signing-key").expect("first token"); let token_b = site_replication_repair_preflight_token(&state, &plan_b, b"test-signing-key").expect("second token"); @@ -3219,3 +3222,148 @@ fn test_reconcile_adds_missing_peer_rules_to_existing_config() { assert!(rule_ids.contains(&"site-repl-dep-b")); assert!(rule_ids.contains(&"site-repl-dep-c")); } + +/// backlog#2289: the IAM snapshot (retry resend, repair, site-add bootstrap) +/// used to be built from `list_users`, whose `UserInfo` never carries a +/// secret key, so the plan dropped every user and a status change or secret +/// rotation committed while a peer was unreachable never reached it. The +/// credentials now come from a separate store read; SRInfo stays secret-free. +#[test] +fn test_bootstrap_plan_carries_users_from_the_credential_snapshot() { + let mut info = SRInfo::default(); + // Exactly what `list_users` builds: status, policy, updated_at — never secret_key. + info.user_info_map.insert( + "alice".to_string(), + rustfs_madmin::UserInfo { + status: rustfs_madmin::AccountStatus::Disabled, + policy_name: Some("readwrite".to_string()), + updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp")), + ..Default::default() + }, + ); + info.user_info_map.insert( + "external-idp-user".to_string(), + rustfs_madmin::UserInfo { + status: rustfs_madmin::AccountStatus::Enabled, + ..Default::default() + }, + ); + let user_updated_at = OffsetDateTime::from_unix_timestamp(1_700_000_500).expect("timestamp"); + let mut credentials = SiteReplicationIamCredentials::default(); + credentials.users.insert( + "alice".to_string(), + SiteReplicationUserCredential { + secret_key: "alice-secret".to_string(), + status: rustfs_madmin::AccountStatus::Disabled, + updated_at: Some(user_updated_at), + }, + ); + + let plan = site_replication_bootstrap_plan(&info, &credentials).expect("bootstrap plan should build"); + + let users: Vec<_> = plan.iam_items.iter().filter(|item| item.r#type == "iam-user").collect(); + assert_eq!(users.len(), 1, "only the user with a credential travels: {:?}", plan.iam_items); + let alice = users[0].iam_user.as_ref().expect("iam user body"); + assert_eq!(alice.access_key, "alice"); + let req = alice.user_req.as_ref().expect("user request"); + assert_eq!(req.secret_key, "alice-secret"); + assert_eq!(req.status, rustfs_madmin::AccountStatus::Disabled); + assert_eq!(req.policy.as_deref(), Some("readwrite")); + // the user record's own axis, not the policy-mapping time list_users reports + assert_eq!(users[0].updated_at, Some(user_updated_at)); +} + +fn service_account_snapshot(access_key: &str, parent: &str, status: &str) -> SiteReplicationServiceAccountSnapshot { + SiteReplicationServiceAccountSnapshot { + create: rustfs_madmin::SRSvcAccCreate { + parent: parent.to_string(), + access_key: access_key.to_string(), + secret_key: format!("{access_key}-secret"), + groups: Vec::new(), + claims: HashMap::new(), + session_policy: SRSessionPolicy::default(), + status: status.to_string(), + name: String::new(), + description: String::new(), + expiration: None, + api_version: Some(SITE_REPL_API_VERSION.to_string()), + }, + envelope: None, + updated_at: Some(OffsetDateTime::from_unix_timestamp(1_700_000_600).expect("timestamp")), + } +} + +/// backlog#2289: service accounts were absent from every snapshot (the +/// listing filters them). They now travel as the create item the live hook +/// emits — after their parents — carrying secret and status. +#[test] +fn test_bootstrap_plan_emits_service_accounts_after_their_parents() { + let mut info = SRInfo::default(); + info.user_info_map + .insert("alice".to_string(), rustfs_madmin::UserInfo::default()); + let mut credentials = SiteReplicationIamCredentials::default(); + credentials.users.insert( + "alice".to_string(), + SiteReplicationUserCredential { + secret_key: "alice-secret".to_string(), + status: rustfs_madmin::AccountStatus::Enabled, + updated_at: None, + }, + ); + credentials + .service_accounts + .push(service_account_snapshot("alice-svc", "alice", "off")); + + let plan = site_replication_bootstrap_plan(&info, &credentials).expect("bootstrap plan should build"); + + let types: Vec<_> = plan.iam_items.iter().map(|item| item.r#type.as_str()).collect(); + assert_eq!(types, vec!["iam-user", "service-account"]); + let change = plan.iam_items[1].svc_acc_change.as_ref().expect("service account change"); + let create = change.create.as_ref().expect("create body"); + assert_eq!((create.access_key.as_str(), create.parent.as_str()), ("alice-svc", "alice")); + assert_eq!(create.secret_key, "alice-svc-secret"); + assert_eq!(create.status, "off", "a disabled account must arrive disabled"); + assert!(change.delete.is_none() && change.update.is_none()); +} + +/// A service account present in the previous snapshot but gone from the +/// fresh one is replayed as an explicit delete, like the other IAM kinds. +#[test] +fn test_retry_snapshot_tombstones_removed_service_accounts() { + let observed_at = OffsetDateTime::from_unix_timestamp(1_700_001_000).expect("timestamp"); + let mut info = SRInfo::default(); + info.user_info_map + .insert("alice".to_string(), rustfs_madmin::UserInfo::default()); + let mut credentials = SiteReplicationIamCredentials::default(); + credentials.users.insert( + "alice".to_string(), + SiteReplicationUserCredential { + secret_key: "alice-secret".to_string(), + status: rustfs_madmin::AccountStatus::Enabled, + updated_at: None, + }, + ); + let mut with_account = credentials.clone(); + with_account + .service_accounts + .push(service_account_snapshot("alice-svc", "alice", "on")); + let previous = site_replication_bootstrap_plan(&info, &with_account).expect("previous plan"); + let fresh = site_replication_bootstrap_plan(&info, &credentials).expect("fresh plan"); + + let replay = RetrySnapshot::replay_after_change( + &RetrySnapshot::Iam(previous.iam_items), + &RetrySnapshot::Iam(fresh.iam_items), + observed_at, + ); + let RetrySnapshot::Iam(items) = replay else { + panic!("IAM snapshot expected"); + }; + let tombstone = items + .iter() + .find(|item| item.r#type == "service-account") + .expect("service account tombstone"); + let change = tombstone.svc_acc_change.as_ref().expect("change"); + assert_eq!(change.delete.as_ref().map(|delete| delete.access_key.as_str()), Some("alice-svc")); + assert!(change.create.is_none()); + assert_eq!(tombstone.updated_at, Some(observed_at)); +}