From c40988dae7795cf214fe8acf523174cc41a7b3f3 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 14:16:19 +0800 Subject: [PATCH 1/6] feat(scanner): replay recovery intents at startup (#7354) Discover durable scanner recovery intents during startup and sequentially re-drive accepted or running usage full-rebuild work. Disabled scanner startup also replays existing durable intents without enabling the normal scanner loop. Co-authored-by: zhi22915 --- crates/scanner/src/scanner.rs | 51 +++++- crates/scanner/src/scanner/cycle_state.rs | 124 +++++++++++++ .../src/scanner/tests/recovery_control.rs | 170 ++++++++++++++++++ 3 files changed, 339 insertions(+), 6 deletions(-) diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index f2337c3ea..b8c16f994 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -70,12 +70,13 @@ use crate::storage_api::scan::{ SCANNER_ACTIVITY_PROTOCOL_VERSION, }; use crate::{ - ECStore, EcstoreError, RUSTFS_META_BUCKET, SCANNER_PUBLICATION_EPOCH_CHANGED, ScannerLifecycleConfigExt as _, - ScannerReplicationConfigExt as _, delete_config_with_publication_admission_for_epoch, get_lifecycle_config, - get_replication_config, invalidate_admin_data_usage_snapshot_cache, invalidate_data_usage_snapshot_cache, read_config, - replace_bucket_usage_memory_from_info, save_config, save_config_shared_with_preconditions_and_lease_fence_and_scope, - save_config_with_preconditions, save_config_with_publication_admission_for_epoch, scanner_publication_admission_for_epoch, - scanner_publication_epoch, scanner_publication_epoch_changed, + DiskError, ECStore, EcstoreError, ListPathRawOptions, RUSTFS_META_BUCKET, SCANNER_PUBLICATION_EPOCH_CHANGED, + ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _, delete_config_with_publication_admission_for_epoch, + get_lifecycle_config, get_replication_config, invalidate_admin_data_usage_snapshot_cache, + invalidate_data_usage_snapshot_cache, list_path_raw, read_config, replace_bucket_usage_memory_from_info, save_config, + save_config_shared_with_preconditions_and_lease_fence_and_scope, save_config_with_preconditions, + save_config_with_publication_admission_for_epoch, scanner_publication_admission_for_epoch, scanner_publication_epoch, + scanner_publication_epoch_changed, }; const LOG_COMPONENT_SCANNER: &str = "scanner"; @@ -947,6 +948,22 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { init_data_scanner_with_storage(ctx, storeapi).await; } +async fn run_scanner_usage_recovery_intents_for_startup( + ctx: CancellationToken, + storeapi: Arc, +) -> Result { + let intent_ids = scanner_usage_recovery_intents_for_startup(&ctx, storeapi.clone()).await?; + let mut attempted = 0usize; + for intent_id in intent_ids { + if ctx.is_cancelled() { + break; + } + run_scanner_usage_recovery_intent(ctx.child_token(), storeapi.clone(), intent_id).await?; + attempted = attempted.saturating_add(1); + } + Ok(attempted) +} + /// Start normal scanning when enabled, or one resume-only cleanup attempt. /// The disabled branch returns a finite task for the startup owner to join; /// it never enables ordinary namespace scanning or accepts a new reset intent. @@ -956,10 +973,32 @@ pub async fn init_scanner_with_recovery( enabled: bool, ) -> Option> { if enabled { + if let Err(error) = run_scanner_usage_recovery_intents_for_startup(ctx.clone(), storeapi.clone()).await { + warn!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + state = "recovery_intent_startup_discovery_failed", + error = %error, + "Scanner recovery intent startup discovery failed" + ); + } init_data_scanner(ctx, storeapi).await; return None; } Some(tokio::spawn(async move { + if let Err(error) = run_scanner_usage_recovery_intents_for_startup(ctx.clone(), storeapi.clone()).await { + warn!( + target: "rustfs::scanner", + event = EVENT_SCANNER_PERSIST_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_RUNTIME, + state = "recovery_intent_startup_discovery_failed", + error = %error, + "Disabled scanner recovery intent startup discovery failed" + ); + } if let Err(error) = resume_scanner_cycle_cleanup(ctx, storeapi).await { warn!( target: "rustfs::scanner", diff --git a/crates/scanner/src/scanner/cycle_state.rs b/crates/scanner/src/scanner/cycle_state.rs index 646fbeead..f20374ae0 100644 --- a/crates/scanner/src/scanner/cycle_state.rs +++ b/crates/scanner/src/scanner/cycle_state.rs @@ -18,6 +18,8 @@ use crate::data_usage_define::{ DATA_USAGE_BLOOM_RECOVERY_PATH, DATA_USAGE_RECOVERY_PATH, usage_floor_primary_read_error_allows_backup, }; use crate::storage_api::owner::ObjectIO as _; +use rustfs_filemeta::MetaCacheEntry; +use std::collections::BTreeSet; use std::sync::atomic::AtomicU64; use tokio::io::AsyncReadExt as _; @@ -42,6 +44,9 @@ const SCANNER_RECOVERY_INTENT_STATE_RUNNING: &str = "running"; const SCANNER_RECOVERY_INTENT_STATE_COMPLETED: &str = "completed"; const SCANNER_RECOVERY_INTENT_STATE_FAILED: &str = "failed"; pub const SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD: &str = "scanner-usage-full-rebuild"; +const MAX_SCANNER_RECOVERY_INTENT_STARTUP_CANDIDATES: usize = 4096; +const SCANNER_RECOVERY_INTENT_STARTUP_PAGE_SIZE: usize = 128; +const MAX_SCANNER_RECOVERY_INTENT_STARTUP_REPLAY: usize = 64; #[cfg(test)] pub(super) mod cleanup_io_fault { @@ -465,6 +470,14 @@ fn scanner_recovery_intent_path(intent_id: &str) -> Result Ok(format!("{SCANNER_RECOVERY_INTENT_PREFIX}/{intent_id}.json")) } +fn scanner_recovery_intent_id_from_entry_name(entry_name: &str) -> Option { + let relative = entry_name + .strip_prefix(SCANNER_RECOVERY_INTENT_PREFIX) + .and_then(|name| name.strip_prefix('/'))?; + let intent_id = relative.strip_suffix(".json")?; + (!intent_id.contains('/') && is_canonical_sha256(intent_id)).then(|| intent_id.to_string()) +} + pub fn scanner_recovery_actor_sha256(actor: &str) -> String { sha256_hex(&[b"scanner-recovery-actor-v1", actor.as_bytes()]) } @@ -623,6 +636,117 @@ pub async fn get_scanner_usage_recovery_intent( read_recovery_intent_record(storeapi, &path).await } +fn scanner_recovery_intent_is_replayable(record: &ScannerRecoveryIntentRecord) -> bool { + matches!( + record.state.as_str(), + SCANNER_RECOVERY_INTENT_STATE_ACCEPTED | SCANNER_RECOVERY_INTENT_STATE_RUNNING + ) +} + +pub(super) async fn scanner_usage_recovery_intents_for_startup( + ctx: &CancellationToken, + storeapi: Arc, +) -> Result, ScannerError> { + let discovered = Arc::new(StdMutex::new(BTreeSet::::new())); + for set in storeapi.all_set_disks() { + if ctx.is_cancelled() { + break; + } + let disks = set.get_local_disks().await; + if disks.is_empty() { + continue; + } + let read_quorum = disks.len().saturating_sub(set.default_parity_count).clamp(1, disks.len()); + let mut forward_to = None; + loop { + let page_entries = Arc::new(StdMutex::new(Vec::::new())); + let page_entries_for_set = page_entries.clone(); + let list_result = list_path_raw( + ctx.child_token(), + ListPathRawOptions { + disks: disks.clone(), + bucket: RUSTFS_META_BUCKET.to_string(), + path: SCANNER_RECOVERY_INTENT_PREFIX.to_string(), + recursive: true, + skip_hidden_prefix_check: true, + forward_to: forward_to.clone(), + min_disks: read_quorum, + report_not_found: true, + per_disk_limit: i32::try_from(SCANNER_RECOVERY_INTENT_STARTUP_PAGE_SIZE).unwrap_or(i32::MAX), + agreed: Some(Box::new(move |entry: MetaCacheEntry| { + let page_entries = page_entries_for_set.clone(); + Box::pin(async move { + if scanner_recovery_intent_id_from_entry_name(&entry.name).is_some() { + page_entries + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .push(entry.name); + } + }) + })), + ..Default::default() + }, + ) + .await; + match list_result { + Ok(()) => {} + Err(DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound) => break, + Err(err) => { + return Err(ScannerError::Other(format!("failed to list scanner recovery intents for startup: {err}"))); + } + } + + let page = { + let mut guard = page_entries.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + guard.sort(); + guard.dedup(); + std::mem::take(&mut *guard) + }; + if page.is_empty() { + break; + } + + { + let mut all = discovered.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + for entry_name in &page { + if let Some(intent_id) = scanner_recovery_intent_id_from_entry_name(entry_name) { + all.insert(intent_id); + } + if all.len() >= MAX_SCANNER_RECOVERY_INTENT_STARTUP_CANDIDATES { + break; + } + } + if all.len() >= MAX_SCANNER_RECOVERY_INTENT_STARTUP_CANDIDATES { + break; + } + } + + if page.len() < SCANNER_RECOVERY_INTENT_STARTUP_PAGE_SIZE { + break; + } + forward_to = page.last().cloned(); + } + } + + let intent_ids = { + let guard = discovered.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + guard.iter().cloned().collect::>() + }; + let mut replayable = Vec::new(); + for intent_id in intent_ids { + let Some(record) = get_scanner_usage_recovery_intent(storeapi.clone(), &intent_id).await? else { + continue; + }; + if scanner_recovery_intent_is_replayable(&record) { + replayable.push(record.intent_id); + if replayable.len() >= MAX_SCANNER_RECOVERY_INTENT_STARTUP_REPLAY { + break; + } + } + } + Ok(replayable) +} + pub async fn accept_scanner_usage_recovery_intent( storeapi: Arc, request: ScannerRecoveryIntentRequest, diff --git a/crates/scanner/src/scanner/tests/recovery_control.rs b/crates/scanner/src/scanner/tests/recovery_control.rs index 5f927011a..dacd83a64 100644 --- a/crates/scanner/src/scanner/tests/recovery_control.rs +++ b/crates/scanner/src/scanner/tests/recovery_control.rs @@ -106,6 +106,20 @@ fn recovery_intent_request(key: &str, actor: &str) -> ScannerRecoveryIntentReque } } +fn synthetic_recovery_intent_record(intent_id: String, state: &str) -> ScannerRecoveryIntentRecord { + ScannerRecoveryIntentRecord { + schema_version: 1, + intent_id, + action: SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD.to_string(), + mode: "full-rebuild".to_string(), + state: state.to_string(), + actor_sha256: "a".repeat(64), + idempotency_key_sha256: "b".repeat(64), + request_sha256: "c".repeat(64), + accepted_at_unix_secs: 1, + } +} + async fn assert_reset_fences(store: &Arc) { let data = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str()) .await @@ -311,6 +325,162 @@ async fn scanner_recovery_intent_executor_persists_failed_progress() { assert_eq!(failed.intent_id, record.intent_id); } +#[tokio::test] +#[serial] +async fn scanner_recovery_intent_startup_discovers_only_non_terminal_records() { + let (_dir, store) = setup_scanner_cycle_store().await; + let accepted = match accept_scanner_usage_recovery_intent( + store.clone(), + recovery_intent_request("intent-key-0001-startup-accepted", "operator-a"), + ) + .await + .expect("accepted startup intent should persist") + { + ScannerRecoveryIntentAcceptResult::Accepted { record } => record, + other => panic!("first request must create accepted startup intent: {other:?}"), + }; + let mut running = match accept_scanner_usage_recovery_intent( + store.clone(), + recovery_intent_request("intent-key-0001-startup-running", "operator-a"), + ) + .await + .expect("running startup intent should persist") + { + ScannerRecoveryIntentAcceptResult::Accepted { record } => record, + other => panic!("first request must create running startup intent: {other:?}"), + }; + running.state = "running".to_string(); + let running_path = format!(".usage.v2.recovery-intents/{}.json", running.intent_id); + save_config( + store.clone(), + &running_path, + serde_json::to_vec(&running).expect("running intent should encode"), + ) + .await + .expect("running intent override should persist"); + + for (key, state) in [ + ("intent-key-0001-startup-completed", "completed"), + ("intent-key-0001-startup-failed", "failed"), + ] { + let mut terminal = match accept_scanner_usage_recovery_intent(store.clone(), recovery_intent_request(key, "operator-a")) + .await + .expect("terminal startup intent should persist") + { + ScannerRecoveryIntentAcceptResult::Accepted { record } => record, + other => panic!("first request must create terminal startup intent: {other:?}"), + }; + terminal.state = state.to_string(); + let path = format!(".usage.v2.recovery-intents/{}.json", terminal.intent_id); + save_config( + store.clone(), + &path, + serde_json::to_vec(&terminal).expect("terminal intent should encode"), + ) + .await + .expect("terminal intent override should persist"); + } + save_config(store.clone(), ".usage.v2.recovery-intents/not-a-sha.json", b"{}".to_vec()) + .await + .expect("foreign startup key should persist"); + + let restarted = restart_scanner_cycle_store_from(&store).await; + let replayable = scanner_usage_recovery_intents_for_startup(&CancellationToken::new(), restarted) + .await + .expect("startup discovery should tolerate terminal and foreign records"); + let replayable = replayable.into_iter().collect::>(); + assert_eq!( + replayable, + [accepted.intent_id, running.intent_id].into_iter().collect(), + "startup discovery must only re-drive accepted/running durable intents" + ); +} + +#[tokio::test] +#[serial] +async fn scanner_recovery_intent_startup_pages_past_terminal_records() { + let (_dir, store) = setup_scanner_cycle_store().await; + for index in 0..70 { + let intent_id = format!("{index:064x}"); + let path = format!(".usage.v2.recovery-intents/{intent_id}.json"); + let terminal = synthetic_recovery_intent_record(intent_id, "completed"); + save_config( + store.clone(), + &path, + serde_json::to_vec(&terminal).expect("terminal paging fixture should encode"), + ) + .await + .expect("terminal paging fixture should persist"); + } + let pending_id = "f".repeat(64); + let pending = synthetic_recovery_intent_record(pending_id.clone(), "accepted"); + save_config( + store.clone(), + &format!(".usage.v2.recovery-intents/{pending_id}.json"), + serde_json::to_vec(&pending).expect("pending paging fixture should encode"), + ) + .await + .expect("pending paging fixture should persist"); + + let restarted = restart_scanner_cycle_store_from(&store).await; + let replayable = scanner_usage_recovery_intents_for_startup(&CancellationToken::new(), restarted) + .await + .expect("startup discovery should page past terminal records"); + assert_eq!(replayable, vec![pending_id]); +} + +#[tokio::test] +#[serial] +async fn scanner_recovery_intent_startup_rejects_corrupt_pending_record() { + let (_dir, store) = setup_scanner_cycle_store().await; + let record = match accept_scanner_usage_recovery_intent( + store.clone(), + recovery_intent_request("intent-key-0001-startup-corrupt", "operator-a"), + ) + .await + .expect("corrupt startup fixture intent should persist") + { + ScannerRecoveryIntentAcceptResult::Accepted { record } => record, + other => panic!("first request must create corrupt startup fixture: {other:?}"), + }; + let path = format!(".usage.v2.recovery-intents/{}.json", record.intent_id); + save_config(store.clone(), &path, b"{corrupt".to_vec()) + .await + .expect("corrupt intent payload should persist"); + + let restarted = restart_scanner_cycle_store_from(&store).await; + let error = scanner_usage_recovery_intents_for_startup(&CancellationToken::new(), restarted) + .await + .expect_err("startup discovery must not silently drop corrupt pending intent records"); + assert!(error.to_string().contains("scanner recovery intent is invalid"), "{error}"); +} + +#[tokio::test] +#[serial] +async fn scanner_recovery_intent_disabled_startup_executes_persisted_non_terminal_intent() { + let (_dir, store) = setup_scanner_cycle_store().await; + let record = match accept_scanner_usage_recovery_intent( + store.clone(), + recovery_intent_request("intent-key-0001-startup-exec", "operator-a"), + ) + .await + .expect("startup execution intent should persist") + { + ScannerRecoveryIntentAcceptResult::Accepted { record } => record, + other => panic!("first request must create startup execution intent: {other:?}"), + }; + + let restarted = restart_scanner_cycle_store_from(&store).await; + run_disabled_startup(CancellationToken::new(), restarted.clone()).await; + + let completed = get_scanner_usage_recovery_intent(restarted, &record.intent_id) + .await + .expect("startup-executed intent should read") + .expect("startup-executed intent should remain durable"); + assert_eq!(completed.state, "completed"); + assert_eq!(completed.intent_id, record.intent_id); +} + #[tokio::test] #[serial] async fn scanner_recovery_intent_rejects_same_namespace_conflict() { From 7c7b48aaca200b9f8a9224420fdc3db3742f42f3 Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 14:16:31 +0800 Subject: [PATCH 2/6] test(heal): cover MRF replay across process restart (#7355) Add an ECStore-backed child-process fixture that publishes the MRF journal in one OS process, then verifies a restarted manager can replay the complete snapshot while retaining the journal when bounded admission accepts only a prefix. Co-authored-by: zhi22915 --- crates/heal/tests/mrf_pipeline_test.rs | 87 ++++++++++++++++++++++++-- 1 file changed, 82 insertions(+), 5 deletions(-) diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs index fdcf7848c..1b54b4899 100644 --- a/crates/heal/tests/mrf_pipeline_test.rs +++ b/crates/heal/tests/mrf_pipeline_test.rs @@ -29,7 +29,7 @@ use rustfs_heal::heal::{ storage::{ECStoreHealStorage, HealStorageAPI}, }; use serial_test::serial; -use std::{path::Path, sync::Arc, time::Duration}; +use std::{path::Path, process::Command, sync::Arc, time::Duration}; mod storage_api; @@ -40,10 +40,15 @@ const JOURNAL_REL: &str = "buckets/.heal/mrf/journal.bin"; const SCOPED_JOURNAL_REL: &str = "buckets/.heal/mrf/journal-scoped.bin"; async fn heal_env() -> (Vec, Arc) { - let env = rustfs_test_utils::TestECStoreEnv::builder() - .prefix("rustfs_heal_mrf_test") - .build() - .await; + heal_env_at(None).await +} + +async fn heal_env_at(base_dir: Option<&Path>) -> (Vec, Arc) { + let mut builder = rustfs_test_utils::TestECStoreEnv::builder().prefix("rustfs_heal_mrf_test"); + if let Some(base_dir) = base_dir { + builder = builder.base_dir(base_dir); + } + let env = builder.build().await; let heal_storage: Arc = Arc::new(ECStoreHealStorage::new(env.ecstore.clone())); (env.disk_paths, heal_storage) } @@ -117,6 +122,12 @@ fn write_journal_to_disks(disk_paths: &[std::path::PathBuf], data: &[u8]) { write_journal_path_to_disks(disk_paths, JOURNAL_REL, data); } +fn journal_exists_on_all_disks(disk_paths: &[std::path::PathBuf], relative_path: &str) -> bool { + disk_paths + .iter() + .all(|path| Path::new(path).join(META_BUCKET).join(relative_path).exists()) +} + async fn wait_until(deadline: Duration, mut probe: F) -> bool where F: FnMut() -> Fut, @@ -313,3 +324,69 @@ async fn journal_replay_retains_file_when_manager_is_full() { "the anchor remains until a successor snapshot can safely replace it" ); } + +#[test] +fn mrf_journal_child_process_fixture() { + let Ok(root) = std::env::var("RUSTFS_MRF_REPLAY_CHILD_ROOT") else { + return; + }; + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("child runtime should build"); + runtime.block_on(async { + let (disk_paths, _storage) = heal_env_at(Some(Path::new(&root))).await; + let mut journal = journal_record(1, "child-restart-bucket", "first-object", None, 0); + journal.extend(journal_record(1, "child-restart-bucket", "second-object", None, 0)); + write_journal_path_to_disks(&disk_paths, SCOPED_JOURNAL_REL, &journal); + write_journal_path_to_disks(&disk_paths, JOURNAL_REL, &journal); + assert!( + journal_exists_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL), + "child process must publish the authoritative MRF journal before exiting" + ); + }); + std::process::exit(77); +} + +/// A journal published by a different OS process must remain a durable anchor +/// when the restarted process can only admit a prefix of the replayed intents. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn journal_replay_retains_child_process_anchor_when_manager_is_full() { + let temp_dir = tempfile::tempdir().expect("child process MRF root"); + let status = Command::new(std::env::current_exe().expect("test binary path")) + .arg("mrf_journal_child_process_fixture") + .arg("--exact") + .arg("--nocapture") + .env("RUSTFS_MRF_REPLAY_CHILD_ROOT", temp_dir.path()) + .status() + .expect("child MRF fixture should start"); + assert_eq!(status.code(), Some(77), "child process did not reach the MRF journal boundary"); + + let (disk_paths, storage) = heal_env_at(Some(temp_dir.path())).await; + assert!( + journal_exists_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL), + "restarted process must see the authoritative MRF journal left by the child" + ); + + let restarted = Arc::new(HealManager::new( + storage, + Some(HealConfig { + queue_size: 1, + heal_interval: Duration::from_secs(3600), + enable_auto_heal: false, + ..Default::default() + }), + )); + let replayed = mrf_queue::replay_journal_once(&restarted).await; + assert_eq!(replayed, 2, "the restarted process must decode the complete child journal"); + assert_eq!( + restarted.operations_snapshot().await.queued_by_source.mrf, + 1, + "bounded admission may accept only the prefix, but must not lose the replayed tail" + ); + assert!( + journal_exists_on_all_disks(&disk_paths, SCOPED_JOURNAL_REL), + "replay must retain the child-published journal until a successor snapshot can replace it" + ); +} From 638da7a2e3df3145232af9d9efe54e75bfd6cb4b Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 14:27:53 +0800 Subject: [PATCH 3/6] fix(scanner): confirm lost dirty usage acknowledgements (#7357) When the remote dirty-usage ACK response is lost, re-probe scanner activity once and accept the ACK only if every target host still reports the same scanner instance with no dirty usage pending. Duplicate targets, restarted peers, unverified activity, and concurrent writes remain pending. Co-authored-by: zhi22915 --- crates/scanner/src/scanner.rs | 3 + crates/scanner/src/scanner/activity.rs | 33 +++++++++- crates/scanner/src/scanner/tests.rs | 91 +++++++++++++++++++++++++- 3 files changed, 124 insertions(+), 3 deletions(-) diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index b8c16f994..d96f90353 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -2255,11 +2255,14 @@ where false } else if let Some(notification_system) = storeapi.scanner_notification_system() { let acknowledgement_count = remote_dirty_usage_acknowledgements.len(); + let acknowledgement_proof = remote_dirty_usage_acknowledgements.clone(); let acknowledgements = remote_dirty_usage_acknowledgements.into_iter().map(Into::into).collect(); remote_dirty_usage_acknowledgement_pending( cycle_info.current, acknowledgement_count, + &acknowledgement_proof, notification_system.acknowledge_scanner_dirty_usage(acknowledgements), + || probe_scanner_activity(storeapi.as_ref(), true), ) .await } else { diff --git a/crates/scanner/src/scanner/activity.rs b/crates/scanner/src/scanner/activity.rs index 582649147..0b5c38ab5 100644 --- a/crates/scanner/src/scanner/activity.rs +++ b/crates/scanner/src/scanner/activity.rs @@ -15,6 +15,7 @@ use super::*; use crate::storage_api::ScannerStorage; use crate::storage_api::scan::SCANNER_ACTIVITY_V6_PROTOCOL_VERSION; +use std::collections::HashSet; #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(super) enum ScannerCycleWakeReason { @@ -51,18 +52,25 @@ pub(crate) fn scanner_cycle_outcome_with_pending_maintenance( } } -pub(super) async fn remote_dirty_usage_acknowledgement_pending( +pub(super) async fn remote_dirty_usage_acknowledgement_pending( cycle: u64, acknowledgement_count: usize, + acknowledgements: &[ScannerDirtyUsageAcknowledgement], acknowledgement: F, + confirm_after_error: C, ) -> bool where F: Future>, E: std::fmt::Display, + C: FnOnce() -> CF, + CF: Future>, { match acknowledgement.await { Ok(dirty_usage_pending) => dirty_usage_pending, Err(err) => { + if remote_dirty_usage_acknowledgement_loss_reconciled(acknowledgements, confirm_after_error().await) { + return false; + } warn!( target: "rustfs::scanner", event = EVENT_SCANNER_PERSIST_STATE, @@ -79,6 +87,29 @@ where } } +pub(super) fn remote_dirty_usage_acknowledgement_loss_reconciled( + acknowledgements: &[ScannerDirtyUsageAcknowledgement], + activity_after_error: Result, +) -> bool { + if acknowledgements.is_empty() { + return false; + } + let Ok(activity_after_error) = activity_after_error else { + return false; + }; + if !scanner_activity_allows_usage_publication(&activity_after_error) { + return false; + } + let mut acknowledged_hosts = HashSet::with_capacity(acknowledgements.len()); + acknowledgements.iter().all(|acknowledgement| { + if !acknowledged_hosts.insert(acknowledgement.host.as_str()) { + return false; + } + scanner_activity_dirty_usage_state_for_host(&activity_after_error, &acknowledgement.host) + .is_some_and(|(instance_id, _generation, pending)| instance_id == acknowledgement.instance_id && !pending) + }) +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(super) struct ScannerCleanIdleBackoff { pub(super) interval_multiplier: u32, diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 5ee1840a0..331c2c107 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -7498,13 +7498,28 @@ fn finalizing_post_scan_observation_advances_partially_without_dirty_ack() { #[tokio::test] async fn scanner_cycle_keeps_remote_pending_acknowledgement() { - let pending = remote_dirty_usage_acknowledgement_pending(7, 1, std::future::ready(Ok::(true))).await; + let acknowledgements = Vec::new(); + let pending = remote_dirty_usage_acknowledgement_pending( + 7, + 1, + &acknowledgements, + std::future::ready(Ok::(true)), + || async { Ok(BTreeMap::new()) }, + ) + .await; assert_eq!( scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, pending), ScannerCycleOutcome::CompletedWithPendingMaintenance ); - let cleared = remote_dirty_usage_acknowledgement_pending(7, 1, std::future::ready(Ok::(false))).await; + let cleared = remote_dirty_usage_acknowledgement_pending( + 7, + 1, + &acknowledgements, + std::future::ready(Ok::(false)), + || async { Ok(BTreeMap::new()) }, + ) + .await; assert_eq!( scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, cleared), ScannerCycleOutcome::Completed @@ -7513,7 +7528,9 @@ async fn scanner_cycle_keeps_remote_pending_acknowledgement() { let failed = remote_dirty_usage_acknowledgement_pending( 7, 1, + &acknowledgements, std::future::ready(Err::(std::io::Error::other("injected acknowledgement failure"))), + || async { Err("confirmation probe failed".to_string()) }, ) .await; assert_eq!( @@ -7522,6 +7539,76 @@ async fn scanner_cycle_keeps_remote_pending_acknowledgement() { ); } +#[tokio::test] +async fn scanner_cycle_confirms_lost_remote_ack_from_activity_snapshot() { + let acknowledgement = ScannerDirtyUsageAcknowledgement { + host: "node-2".to_string(), + instance_id: "epoch-a".to_string(), + kind: ScannerDirtyUsageAcknowledgementKind::Generation(5), + }; + let cleared_activity = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]); + let response_lost = remote_dirty_usage_acknowledgement_pending( + 8, + 1, + std::slice::from_ref(&acknowledgement), + std::future::ready(Err::(std::io::Error::other("response lost after peer ack"))), + || async { Ok(cleared_activity) }, + ) + .await; + assert_eq!( + scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, response_lost), + ScannerCycleOutcome::Completed, + "a same-instance activity confirmation with no dirty work closes the uncertain ACK" + ); + + let duplicate_acknowledgements = vec![acknowledgement.clone(), acknowledgement.clone()]; + let duplicate_target = remote_dirty_usage_acknowledgement_pending( + 8, + duplicate_acknowledgements.len(), + &duplicate_acknowledgements, + std::future::ready(Err::(std::io::Error::other("duplicate target rejected before peer ack"))), + || async { Ok(BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))])) }, + ) + .await; + assert_eq!( + scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, duplicate_target), + ScannerCycleOutcome::CompletedWithPendingMaintenance, + "a request rejected before peer delivery cannot be recovered by a clean activity snapshot" + ); + + let restarted_activity = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-b", 7, 3))]); + let peer_restarted = remote_dirty_usage_acknowledgement_pending( + 8, + 1, + std::slice::from_ref(&acknowledgement), + std::future::ready(Err::(std::io::Error::other("response lost before restart was observed"))), + || async { Ok(restarted_activity) }, + ) + .await; + assert_eq!( + scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, peer_restarted), + ScannerCycleOutcome::CompletedWithPendingMaintenance, + "a new peer instance cannot confirm whether the old ACK reached durable dirty state" + ); + + let mut written_activity = scanner_node_activity("epoch-a", 7, 3); + written_activity.dirty_usage_generation = 6; + written_activity.dirty_usage_pending = true; + let concurrent_write = remote_dirty_usage_acknowledgement_pending( + 8, + 1, + &[acknowledgement], + std::future::ready(Err::(std::io::Error::other("response lost before concurrent write"))), + || async { Ok(BTreeMap::from([("node-2".to_string(), written_activity)])) }, + ) + .await; + assert_eq!( + scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, concurrent_write), + ScannerCycleOutcome::CompletedWithPendingMaintenance, + "new dirty usage on the same peer must keep maintenance pending" + ); +} + #[test] #[serial] fn finalizing_an_already_durable_enum_without_proof_keeps_dirty_pending() { From 2d6f9417ffea2ec72f137f5655341b0b65a2ff6b Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 14:27:59 +0800 Subject: [PATCH 4/6] feat(scanner): persist raw enumeration cancellation cursor (#7358) Return a partial data-usage cache when raw filesystem enumeration is cancelled before object progress can be written. The partial cache now carries a validated raw enumeration cursor for V2 checkpoint scans while keeping the snapshot incomplete and clearing older frontier/checkpoint metadata. Co-authored-by: zhi22915 --- crates/scanner/src/scanner_folder.rs | 115 +++++++++++++++++- crates/scanner/src/scanner_folder/tests.rs | 88 ++++++++++++++ .../tests/enumeration_restart.rs | 39 ++++-- 3 files changed, 227 insertions(+), 15 deletions(-) diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 05e04b0ac..3518fa002 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -21,8 +21,8 @@ use std::time::{Duration, Instant, SystemTime}; use crate::ReplTargetSizeSummary; use crate::data_usage_define::{ DATA_USAGE_SCAN_CHECKPOINT_VERSION, DataUsageCache, DataUsageCacheInfo, DataUsageEntry, DataUsageHash, DataUsageHashMap, - DataUsageScanCheckpoint, DataUsageScanCheckpointReason, PendingScannerHeal, PendingScannerHealKind, ScannerSizeSummaryExt, - SizeReconciliationEntry, SizeSummary, hash_path, + DataUsageRawEnumerationCursor, DataUsageScanCheckpoint, DataUsageScanCheckpointReason, PendingScannerHeal, + PendingScannerHealKind, ScannerSizeSummaryExt, SizeReconciliationEntry, SizeSummary, hash_path, }; use crate::error::ScannerError; use crate::runtime_config::{ @@ -55,6 +55,7 @@ use rustfs_scanner_metrics::metrics::{ UpdateCurrentPathFn, current_path_updater, global_metrics, }; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; +use sha2::{Digest, Sha256}; use time::OffsetDateTime; use tokio::select; use tokio::sync::mpsc; @@ -733,6 +734,7 @@ pub struct FolderScanner { coverage_frontier: Option, resume_frontier: Option, coverage_gap: bool, + raw_enumeration_progress: Vec, pending_heal_sync_deferred: bool, pending_heal_batch_dirty: bool, #[cfg(test)] @@ -744,6 +746,50 @@ pub struct FolderScanner { list_path_raw_options_observer: Option>, } +struct RawEnumerationProgress { + parent: String, + last_entry: Option, + entries_seen: u64, + digest: Sha256, +} + +impl RawEnumerationProgress { + fn new(parent: &str) -> Self { + let mut digest = Sha256::new(); + update_raw_enumeration_digest(&mut digest, b"parent", parent.as_bytes()); + Self { + parent: parent.to_string(), + last_entry: None, + entries_seen: 0, + digest, + } + } + + fn record_entry(&mut self, entry: &str) { + update_raw_enumeration_digest(&mut self.digest, b"entry", entry.as_bytes()); + self.last_entry = Some(entry.to_string()); + self.entries_seen = self.entries_seen.saturating_add(1); + } + + fn into_cursor(self) -> Option { + if self.entries_seen == 0 { + return None; + } + Some(DataUsageRawEnumerationCursor::new( + self.parent, + self.last_entry, + self.entries_seen, + self.digest.finalize().into(), + )) + } +} + +fn update_raw_enumeration_digest(digest: &mut Sha256, label: &[u8], value: &[u8]) { + digest.update(label); + digest.update(u64::try_from(value.len()).unwrap_or(u64::MAX).to_le_bytes()); + digest.update(value); +} + fn size_reconciliation_entry_bytes(entry: &SizeReconciliationEntry) -> usize { entry.key.len() + entry.bucket.len() @@ -999,6 +1045,41 @@ impl FolderScanner { } } + fn record_raw_enumeration_entry(&mut self, parent: &str, entry: &str) { + if self.old_cache.info.scan_progress.is_none() { + return; + } + if let Some(position) = self + .raw_enumeration_progress + .iter() + .position(|progress| progress.parent == parent) + { + self.raw_enumeration_progress.truncate(position + 1); + } else { + self.raw_enumeration_progress.push(RawEnumerationProgress::new(parent)); + } + if let Some(progress) = self.raw_enumeration_progress.last_mut() { + progress.record_entry(entry); + } + } + + fn finish_raw_enumeration_parent(&mut self, parent: &str) { + self.raw_enumeration_progress.retain(|progress| { + progress.parent != parent + && !progress + .parent + .strip_prefix(parent) + .is_some_and(|suffix| suffix.starts_with(SLASH_SEPARATOR)) + }); + } + + fn take_raw_enumeration_cursor(&mut self) -> Option { + self.raw_enumeration_progress + .drain(..) + .next() + .and_then(RawEnumerationProgress::into_cursor) + } + fn carry_forward_old_children(&mut self, parent_hash: &DataUsageHash, entry: &mut DataUsageEntry) { if entry.compacted { // Compacted entries store child totals directly; child links would be flattened twice. @@ -1329,11 +1410,15 @@ impl FolderScanner { }; let mut pending_entry_progress = 0_u64; let mut last_entry_progress = Instant::now(); + let mut raw_enumeration_complete = false; loop { let entry = match dir_reader.next_entry().await { Ok(Some(entry)) => entry, - Ok(None) => break, + Ok(None) => { + raw_enumeration_complete = true; + break; + } Err(e) if e.kind() == ErrorKind::NotFound => { debug!( target: "rustfs::scanner::folder", @@ -1345,6 +1430,7 @@ impl FolderScanner { error = %e, "Scanner folder state updated" ); + raw_enumeration_complete = true; break; } Err(e) if e.kind() == ErrorKind::NotADirectory => { @@ -1358,6 +1444,7 @@ impl FolderScanner { error = %e, "Scanner folder state updated" ); + raw_enumeration_complete = true; break; } Err(e) => return Err(ScannerError::Io(e)), @@ -1376,6 +1463,7 @@ impl FolderScanner { if file_name.is_empty() || file_name == "." || file_name == ".." { continue; } + self.record_raw_enumeration_entry(&folder.name, &file_name); let is_storage_format_entry = file_name == STORAGE_FORMAT_FILE; let file_path = entry.path().to_string_lossy().to_string(); @@ -1686,6 +1774,9 @@ impl FolderScanner { } } self.budget.record_entries_visited(pending_entry_progress); + if raw_enumeration_complete { + self.finish_raw_enumeration_parent(&folder.name); + } let mut found_erasure_data_directory = false; if self.is_erasure_mode && !found_object_metadata { @@ -2533,6 +2624,7 @@ pub(crate) async fn scan_data_folder_scoped( coverage_gap: false, pending_heal_sync_deferred: false, pending_heal_batch_dirty: false, + raw_enumeration_progress: Vec::new(), #[cfg(test)] pending_heal_sync_count: 0, pending_size_reconciliation_keys: HashSet::new(), @@ -2593,6 +2685,7 @@ pub(crate) async fn scan_data_folder_scoped( let had_scan_checkpoint = cache.info.scan_checkpoint.is_some() || new_cache.info.scan_checkpoint.is_some(); new_cache.info.scan_resume_after = None; new_cache.info.scan_checkpoint = None; + new_cache.info.scan_raw_enumeration_cursor = None; new_cache.info.scan_coverage_receipt = None; if had_scan_checkpoint { global_metrics().record_scanner_checkpoint_cleared(); @@ -2610,6 +2703,9 @@ pub(crate) async fn scan_data_folder_scoped( let root_hash = hash_path(&cache.info.name); let root_has_progress = data_usage_root_has_progress(&root); let pending_heals_changed = scanner.pending_heals_changed; + let raw_enumeration_cursor = scanner.take_raw_enumeration_cursor(); + let carry_forward_cache = + (raw_enumeration_cursor.is_some() && !root_has_progress).then(|| scanner.old_cache.cache.clone()); if root_has_progress { scanner.carry_forward_old_children(&root_hash, &mut root); } @@ -2617,8 +2713,19 @@ pub(crate) async fn scan_data_folder_scoped( let new_cache = scanner.as_mut_new_cache(); if root_has_progress { new_cache.replace_hashed(&root_hash, &None, &root); + } else if let Some(cache) = carry_forward_cache { + new_cache.cache = cache; } - if partial_cache_is_useful(&root, pending_heals_changed) || !new_cache.info.size_reconciliation.is_empty() { + if raw_enumeration_cursor.is_some() { + new_cache.info.scan_raw_enumeration_cursor = raw_enumeration_cursor; + new_cache.info.scan_checkpoint = None; + new_cache.info.scan_resume_after = None; + new_cache.info.scan_coverage_receipt = None; + } + if partial_cache_is_useful(&root, pending_heals_changed) + || new_cache.info.scan_raw_enumeration_cursor.is_some() + || !new_cache.info.size_reconciliation.is_empty() + { if new_cache.root().is_some() { new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN); } diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index 962090a00..cc1add61f 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/crates/scanner/src/scanner_folder/tests.rs @@ -353,6 +353,7 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) { coverage_frontier: None, resume_frontier: None, coverage_gap: false, + raw_enumeration_progress: Vec::new(), pending_heal_sync_deferred: false, pending_heal_batch_dirty: false, pending_heal_sync_count: 0, @@ -2637,6 +2638,93 @@ async fn test_scan_data_folder_returns_partial_cache_on_budget_cancel() { assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Directories)); } +#[tokio::test] +#[serial] +async fn test_scan_data_folder_returns_raw_cursor_on_enumeration_cancel_without_root_progress() { + let (scanner, temp_dir) = build_test_scanner().await; + let _guard = TestGuard { + temp_dir: Some(temp_dir.clone()), + }; + + let bucket_dir = temp_dir.join("bucket"); + tokio::fs::create_dir_all(&bucket_dir) + .await + .expect("failed to create bucket directory"); + for entry in ["entry-a", "entry-b", "entry-c"] { + tokio::fs::write(bucket_dir.join(entry), b"data") + .await + .expect("failed to create raw directory entry"); + } + + let plan = crate::data_usage_define::DataUsageScanPlanDigest([11; 32]); + let source = crate::data_usage_define::DataUsageCacheSource::new(1, 0); + let identity = crate::data_usage_define::DataUsageScanIdentity { + version: 1, + bucket_incarnation: Uuid::from_u128(7), + set_layout: crate::data_usage_define::DataUsageScanPlanDigest([12; 32]), + publication_epoch: 3, + tier_registry_generation: 0, + scan_mode: HealScanMode::Normal, + }; + let mut cache = DataUsageCache { + info: crate::data_usage_define::DataUsageCacheInfo { + name: "bucket".to_string(), + next_cycle: 7, + ..Default::default() + }, + ..Default::default() + }; + assert_eq!( + cache.prepare_bucket_checkpoint("bucket", 7, 3, source, plan, identity), + crate::data_usage_define::DataUsageCachePrepareOutcome::Reset + ); + + let parent = CancellationToken::new(); + let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default()); + let _raw_entry_budget = enumeration_restart::install_raw_entry_budget(scanner.local_disk.path(), 1); + + let result = scan_data_folder( + budget.token(), + budget.clone(), + vec![scanner.local_disk.clone()], + scanner.local_disk.clone(), + cache, + None, + HealScanMode::Normal, + SCANNER_SLEEPER.clone(), + ) + .await; + + let partial_cache = match result { + Err(ScannerError::PartialCache(partial_cache)) => partial_cache, + other => panic!("expected raw enumeration partial cache after cancellation, got {other:?}"), + }; + + assert!( + partial_cache + .root() + .is_none_or(|root| root.objects == 0 && root.versions == 0 && root.size == 0), + "raw cursor writer must not invent object progress" + ); + assert!(partial_cache.info.last_update.is_some()); + assert_eq!(partial_cache.info.next_cycle, 7); + assert!(!partial_cache.info.snapshot_complete); + assert!(partial_cache.info.scan_checkpoint.is_none()); + assert!(partial_cache.info.scan_resume_after.is_none()); + + let raw_cursor = partial_cache + .info + .scan_raw_enumeration_cursor + .as_ref() + .expect("raw enumeration cancellation should persist a cursor"); + assert_eq!(raw_cursor.parent, "bucket"); + assert_eq!(raw_cursor.entries_seen, 1); + assert!(raw_cursor.last_entry.is_some()); + assert_ne!(raw_cursor.page_digest, [0; 32]); + assert_eq!(partial_cache.validated_raw_enumeration_cursor(), Some(raw_cursor)); + assert_eq!(budget.reason(), Some(crate::scanner_budget::ScannerCycleBudgetReason::Runtime)); +} + #[tokio::test] #[serial] async fn test_scan_data_folder_reports_invalid_checkpoint_ignored_once() { diff --git a/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs b/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs index ccb37882b..f04759491 100644 --- a/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs +++ b/crates/scanner/src/scanner_folder/tests/enumeration_restart.rs @@ -40,7 +40,7 @@ pub(in crate::scanner_folder) fn observe_raw_entry(dir: &str, name: &std::ffi::O } } -struct ObservationGuard; +pub(in crate::scanner_folder) struct ObservationGuard; impl Drop for ObservationGuard { fn drop(&mut self) { @@ -48,6 +48,18 @@ impl Drop for ObservationGuard { } } +pub(in crate::scanner_folder) fn install_raw_entry_budget(root: PathBuf, limit: u64) -> ObservationGuard { + *OBSERVATION.lock().expect("install raw-entry observation") = Some(Observation { + root, + limit, + entries: 0, + name_bytes: 0, + first_entry: None, + last_entry: None, + }); + ObservationGuard +} + #[derive(serde::Deserialize)] #[serde(deny_unknown_fields)] struct Request { @@ -84,8 +96,21 @@ async fn round(request: &Request) -> serde_json::Value { } let mut initial = DataUsageCache::default(); initial.info.name = "bucket".to_string(); + let source = crate::data_usage_define::DataUsageCacheSource::new(0, 0); + let plan = crate::data_usage_define::DataUsageScanPlanDigest([31; 32]); + let identity = crate::data_usage_define::DataUsageScanIdentity { + version: 1, + bucket_incarnation: Uuid::from_u128(31), + set_layout: crate::data_usage_define::DataUsageScanPlanDigest([32; 32]), + publication_epoch: 1, + tier_registry_generation: 0, + scan_mode: HealScanMode::Normal, + }; + assert_eq!( + initial.prepare_bucket_checkpoint("bucket", 1, 0, source, plan, identity), + crate::data_usage_define::DataUsageCachePrepareOutcome::Reset + ); initial.info.skip_healing = true; - initial.info.snapshot_complete = false; initial.replace("bucket", "", DataUsageEntry::default()); tokio::fs::write(&cache_path, initial.marshal_msg().expect("initial cache codec")) .await @@ -106,15 +131,7 @@ async fn round(request: &Request) -> serde_json::Value { .expect("open synthetic disk in this process"); let parent = CancellationToken::new(); let budget = ScannerCycleBudget::new_with_progress_tracking(&parent, Default::default()); - *OBSERVATION.lock().expect("install observation") = Some(Observation { - root: disk.path(), - limit: request.raw_entry_budget, - entries: 0, - name_bytes: 0, - first_entry: None, - last_entry: None, - }); - let _observation_guard = ObservationGuard; + let _observation_guard = install_raw_entry_budget(disk.path(), request.raw_entry_budget); let result = scan_data_folder( budget.token(), budget.clone(), From ad0c44dc63ab5eee5e037ebbc56e395213920a9e Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 7 Sep 2026 14:31:25 +0800 Subject: [PATCH 5/6] fix(ecstore): keep pool meta read probes retryable (#7359) Evaluate read-only pool metadata planning probes through an isolated write-state clone so transient unreadable replicas fail the current admission without permanently latching the shared writer gate. Co-authored-by: zhi22915 --- crates/ecstore/src/core/pools.rs | 69 ++++++++++++++++++++++++++++++-- 1 file changed, 66 insertions(+), 3 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index c9733e787..d98c29c2d 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -5137,6 +5137,23 @@ where } } +fn select_pool_meta_replicas_for_read_probe( + write_state: &PoolMetaWriteState, + replicas: Vec, + operation: &str, +) -> Result +where + R: Into, +{ + // Read-only planning probes must fail the current request on unsafe pool + // metadata, but they must not permanently poison the shared writer gate. + let mut probe_state = write_state.clone(); + let selection = select_pool_meta_replicas_observing(&mut probe_state, replicas)?; + probe_state.observe_replicas(selection.replica_state); + probe_state.ensure_write_safe(operation)?; + Ok(selection) +} + async fn load_pool_meta_replicas(pools: Vec>, no_lock: bool) -> Result where S: EcstoreObjectIO, @@ -5156,6 +5173,19 @@ where select_pool_meta_replicas_observing(write_state, replicas) } +async fn load_pool_meta_replicas_for_read_probe( + pools: Vec>, + no_lock: bool, + write_state: &PoolMetaWriteState, + operation: &str, +) -> Result +where + S: EcstoreObjectIO, +{ + let replicas = read_pool_meta_replicas(pools, no_lock).await; + select_pool_meta_replicas_for_read_probe(write_state, replicas, operation) +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] struct PersistedPoolMetaV3 { @@ -8997,9 +9027,7 @@ impl ECStore { })?; let pool_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?; let pool_meta_guard = pool_meta_lock.get_read_lock(get_lock_acquire_timeout()).await?; - let selection = load_pool_meta_replicas_observing(self.pools.clone(), true, write_state).await?; - write_state.observe_replicas(selection.replica_state); - write_state.ensure_write_safe(operation)?; + let selection = load_pool_meta_replicas_for_read_probe(self.pools.clone(), true, write_state, operation).await?; Ok((pool_meta_guard, selection.meta)) } @@ -17952,6 +17980,41 @@ mod tests { ); } + #[test] + fn pool_meta_read_probe_does_not_latch_writer_state() { + let write_state = PoolMetaWriteState::default(); + select_pool_meta_replicas_for_read_probe( + &write_state, + vec![PoolMetaReplica::Unreadable("transient read failure".to_string())], + "capacity probe", + ) + .expect_err("an unreadable probe replica must fail the current admission"); + + write_state + .ensure_write_safe("ordinary object write") + .expect("a read-only capacity probe must not permanently latch the pool metadata writer"); + } + + #[tokio::test] + #[serial_test::serial] + async fn pool_meta_read_guard_does_not_latch_after_unreadable_replica() { + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await; + for set in &store.pools[1].disk_set { + let mut disks = set.disks.write().await; + let disk_count = disks.len(); + *disks = vec![None; disk_count]; + } + + let mut write_state = store.pool_meta_save_gate.lock().await; + store + .acquire_pool_meta_read_guard(&mut write_state, "capacity probe") + .await + .expect_err("an unreadable metadata replica must reject this probe"); + write_state + .ensure_write_safe("ordinary object write") + .expect("a failed read-only probe must remain retryable"); + } + #[test] fn pool_meta_write_state_blocks_on_any_recovery_required_selection() { fn assert_selection_blocks(replicas: Vec) { From c9c6bb7a24eb9fa8f158c350efde789b50dbf514 Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Mon, 7 Sep 2026 14:38:04 +0800 Subject: [PATCH 6/6] fix(oidc): support workload discovery and JWKS negotiation (#7348) * fix(oidc): support workload discovery and JWKS negotiation * test(oidc): cover existing Console authorization code flows * test(oidc): cover admin validation and document workload setup --- Cargo.lock | 1 + Cargo.toml | 1 + crates/iam/Cargo.toml | 1 + crates/iam/src/oidc.rs | 708 ++++++++++++++++-- docs/operations/oidc-console-integration.md | 2 +- docs/operations/oidc-provider-requirements.md | 12 + rustfs/src/admin/handlers/oidc.rs | 97 ++- 7 files changed, 766 insertions(+), 56 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f02676ff7..80f15677c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9999,6 +9999,7 @@ dependencies = [ "rustls-pki-types", "serde", "serde_json", + "serde_with", "serial_test", "sha1 0.11.0", "sha2 0.11.0", diff --git a/Cargo.toml b/Cargo.toml index 01b10ff45..c12edcaba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -191,6 +191,7 @@ rmp = { version = "0.8.15" } rmp-serde = { version = "1.3.1" } serde = { version = "1.0.229" } serde_ignored = { version = "0.1" } +serde_with = { version = "3", default-features = false, features = ["macros", "std"] } serde_json = { version = "1.0.151" } serde_urlencoded = "0.7.1" diff --git a/crates/iam/Cargo.toml b/crates/iam/Cargo.toml index 5d8116af6..49ad6203b 100644 --- a/crates/iam/Cargo.toml +++ b/crates/iam/Cargo.toml @@ -86,6 +86,7 @@ rustfs-ecstore = { workspace = true } rustfs-storage-api = { workspace = true } rustfs-policy.workspace = true serde_json = { workspace = true, features = ["raw_value"] } +serde_with = { workspace = true } async-trait.workspace = true thiserror.workspace = true arc-swap = { workspace = true } diff --git a/crates/iam/src/oidc.rs b/crates/iam/src/oidc.rs index fc287bb4a..27fc2db50 100644 --- a/crates/iam/src/oidc.rs +++ b/crates/iam/src/oidc.rs @@ -19,11 +19,13 @@ //! and ID token verification. use crate::oidc_state::{OidcAuthSession, OidcLogoutSession, OidcStateStore}; -use openidconnect::core::{CoreAuthenticationFlow, CoreClient, CoreIdToken, CoreJsonWebKeySet}; +use openidconnect::core::{ + CoreAuthenticationFlow, CoreClient, CoreIdToken, CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm, +}; use openidconnect::{ AsyncHttpClient, Audience, AuthType, AuthorizationCode, ClientId, ClientSecret, CsrfToken, DiscoveryError, IssuerUrl, JsonWebKeySetUrl, LogoutRequest, Nonce, PkceCodeChallenge, PkceCodeVerifier, PostLogoutRedirectUrl, - ProviderMetadataWithLogout, RedirectUrl, RequestTokenError, Scope, + ProviderMetadataWithLogout, RedirectUrl, RequestTokenError, Scope, TokenUrl, }; use reqwest::{Certificate, Client}; use rustfs_config::oidc::*; @@ -52,6 +54,7 @@ const EVENT_OIDC_HTTP: &str = "oidc_http"; const OIDC_JWKS_REFRESH_INTERVAL: StdDuration = StdDuration::from_secs(24 * 60 * 60); const OIDC_DISCOVERY_TRANSPORT_RETRIES: usize = 3; const OIDC_DISCOVERY_TRANSPORT_RETRY_DELAY: StdDuration = StdDuration::from_millis(50); +const OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY: &str = "JWKS request blocked by outbound policy"; const OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY: &str = "OIDC provider discovery blocked by outbound policy"; const OIDC_HTTP_REQUEST_TIMEOUT: StdDuration = StdDuration::from_secs(10); const OIDC_HTTP_CONNECT_TIMEOUT: StdDuration = StdDuration::from_secs(3); @@ -753,7 +756,7 @@ pub struct SourcedOidcProviderConfig { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct OidcProviderValidationResult { pub issuer: String, - pub authorization_endpoint: String, + pub authorization_endpoint: Option, pub token_endpoint: Option, } @@ -783,10 +786,142 @@ pub struct OidcClaims { /// on-the-fly from metadata when needed. #[derive(Clone)] struct ProviderState { - metadata: ProviderMetadataWithLogout, + metadata: DiscoveredProviderMetadata, discovered_at: Instant, } +// Workload issuers do not implement the browser authorization flow. Keep their +// verification metadata separate rather than inventing an authorization URL. +#[serde_with::serde_as] +#[derive(Clone, Deserialize)] +struct WorkloadProviderMetadata { + issuer: IssuerUrl, + jwks_uri: JsonWebKeySetUrl, + token_endpoint: Option, + #[serde_as(as = "serde_with::VecSkipError<_>")] + id_token_signing_alg_values_supported: Vec, + #[serde(skip)] + jwks: CoreJsonWebKeySet, + // Discovery is extensible; report unsupported fields without logging values. + #[serde(flatten)] + additional_fields: HashMap, +} + +#[derive(Clone)] +enum DiscoveredProviderMetadata { + Console(Box), + Workload(Box), +} + +impl DiscoveredProviderMetadata { + fn parse(body: &[u8], hide_from_ui: bool) -> Result { + let document: serde_json::Value = serde_json::from_slice(body).map_err(|err| err.to_string())?; + if hide_from_ui + && document + .as_object() + .is_some_and(|fields| !fields.contains_key("authorization_endpoint")) + { + let mut metadata: WorkloadProviderMetadata = serde_json::from_slice(body).map_err(|err| err.to_string())?; + if !metadata.additional_fields.is_empty() { + warn!( + event = EVENT_OIDC_DIAGNOSTICS, + component = LOG_COMPONENT_IAM, + subsystem = LOG_SUBSYSTEM_OIDC, + result = "workload_discovery_additional_fields", + field_count = metadata.additional_fields.len(), + "workload discovery contains additional fields" + ); + metadata.additional_fields.clear(); + } + Ok(Self::Workload(Box::new(metadata))) + } else { + serde_json::from_slice(body) + .map(|metadata| Self::Console(Box::new(metadata))) + .map_err(|err| err.to_string()) + } + } + + fn console(&self) -> Result<&ProviderMetadataWithLogout, String> { + match self { + Self::Console(metadata) => Ok(metadata), + Self::Workload(_) => Err("OIDC provider has no authorization endpoint; only web identity is supported".into()), + } + } + + fn issuer(&self) -> &IssuerUrl { + match self { + Self::Console(metadata) => metadata.issuer(), + Self::Workload(metadata) => &metadata.issuer, + } + } + + fn jwks_uri(&self) -> &JsonWebKeySetUrl { + match self { + Self::Console(metadata) => metadata.jwks_uri(), + Self::Workload(metadata) => &metadata.jwks_uri, + } + } + + fn set_jwks(self, jwks: CoreJsonWebKeySet) -> Self { + match self { + Self::Console(metadata) => Self::Console(Box::new(metadata.set_jwks(jwks))), + Self::Workload(mut metadata) => { + metadata.jwks = jwks; + Self::Workload(metadata) + } + } + } + + fn authorization_endpoint(&self) -> Option { + match self { + Self::Console(metadata) => Some(metadata.authorization_endpoint().to_string()), + Self::Workload(_) => None, + } + } + + fn token_endpoint(&self) -> Option<&TokenUrl> { + match self { + Self::Console(metadata) => metadata.token_endpoint(), + Self::Workload(metadata) => metadata.token_endpoint.as_ref(), + } + } + + fn verifier(&self, config: &OidcProviderConfig) -> CoreIdTokenVerifier<'static> { + let client_id = ClientId::new(config.client_id.clone()); + let secret = config.client_secret.as_ref().map(|secret| ClientSecret::new(secret.clone())); + let (issuer, jwks, algorithms) = match self { + Self::Console(metadata) => (metadata.issuer(), metadata.jwks(), metadata.id_token_signing_alg_values_supported()), + Self::Workload(metadata) => (&metadata.issuer, &metadata.jwks, &metadata.id_token_signing_alg_values_supported), + }; + let verifier = match secret { + Some(secret) => CoreIdTokenVerifier::new_confidential_client(client_id, secret, issuer.clone(), jwks.clone()), + None => CoreIdTokenVerifier::new_public_client(client_id, issuer.clone(), jwks.clone()), + }; + verifier.set_allowed_algs(algorithms.clone()) + } +} + +// This adapter is used only for discovery/JWKS fetches, never token exchange. +struct JwksAcceptClient<'a> { + inner: &'a ReqwestHttpClient, + discovery_url: Option, +} + +impl<'c> AsyncHttpClient<'c> for JwksAcceptClient<'_> { + type Error = OidcHttpError; + type Future = >::Future; + + fn call(&'c self, mut request: http::Request>) -> Self::Future { + if !self.discovery_url.as_ref().is_some_and(|url| request.uri() == url.as_str()) { + request.headers_mut().insert( + http::header::ACCEPT, + http::HeaderValue::from_static("application/json, application/jwk-set+json"), + ); + } + self.inner.call(request) + } +} + impl ProviderState { fn is_stale(&self) -> bool { self.discovered_at.elapsed() >= OIDC_JWKS_REFRESH_INTERVAL @@ -932,7 +1067,7 @@ impl OidcSys { let redirect = RedirectUrl::new(redirect_uri.to_string()).map_err(|e| format!("invalid redirect URI: {e}"))?; let client = CoreClient::from_provider_metadata( - state.metadata.clone(), + state.metadata.console()?.clone(), ClientId::new(config.client_id.clone()), config.client_secret.as_ref().map(|s| ClientSecret::new(s.clone())), ) @@ -994,7 +1129,7 @@ impl OidcSys { // Construct CoreClient on-the-fly with JWKS from discovery let client = CoreClient::from_provider_metadata( - provider_state.metadata.clone(), + provider_state.metadata.console()?.clone(), ClientId::new(config.client_id.clone()), config.client_secret.as_ref().map(|s| ClientSecret::new(s.clone())), ) @@ -1230,7 +1365,7 @@ impl OidcSys { ); let client = CoreClient::from_provider_metadata( - refreshed_state.metadata, + refreshed_state.metadata.console()?.clone(), ClientId::new(config.client_id.clone()), config.client_secret.as_ref().map(|s| ClientSecret::new(s.clone())), ) @@ -1320,7 +1455,7 @@ impl OidcSys { .get(&session.provider_id) .ok_or_else(|| format!("unknown OIDC provider: {}", session.provider_id))?; let state = self.ensure_provider_state(&session.provider_id, config).await?; - let Some(end_session_endpoint) = state.metadata.additional_metadata().end_session_endpoint.clone() else { + let Some(end_session_endpoint) = state.metadata.console()?.additional_metadata().end_session_endpoint.clone() else { return Ok(None); }; @@ -1460,14 +1595,6 @@ impl OidcSys { state = self.ensure_provider_state_if_stale(&provider_id, &config, &state).await?; - // Reconstruct CoreClient from provider metadata - let client = CoreClient::from_provider_metadata( - state.metadata.clone(), - ClientId::new(config.client_id.clone()), - config.client_secret.as_ref().map(|s| ClientSecret::new(s.clone())), - ) - .set_auth_type(AuthType::RequestBody); - // Parse raw JWT string into CoreIdToken let id_token: CoreIdToken = jwt .parse() @@ -1475,8 +1602,9 @@ impl OidcSys { // Verify the token (signature, issuer, audience, expiry) — skip nonce // (nonce is only required for the authorization code flow) - let verifier = client - .id_token_verifier() + let verifier = state + .metadata + .verifier(&config) .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud)); if let Err(e) = id_token.claims(&verifier, |_: Option<&Nonce>| Ok(())) { state = self @@ -1486,14 +1614,9 @@ impl OidcSys { format!("ID token verification failed: {e}; failed to refresh provider metadata: {refresh_err}") })?; - let client = CoreClient::from_provider_metadata( - state.metadata, - ClientId::new(config.client_id.clone()), - config.client_secret.as_ref().map(|s| ClientSecret::new(s.clone())), - ) - .set_auth_type(AuthType::RequestBody); - let verifier = client - .id_token_verifier() + let verifier = state + .metadata + .verifier(&config) .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud)); id_token .claims(&verifier, |_: Option<&Nonce>| Ok(())) @@ -1868,18 +1991,39 @@ impl OidcSys { let issuer_url = IssuerUrl::new(candidate_issuer.clone()).map_err(|e| format!("invalid issuer URL: {e}"))?; for attempt in 0..OIDC_DISCOVERY_TRANSPORT_RETRIES { - match ProviderMetadataWithLogout::discover_async(issuer_url.clone(), http_client).await { - Ok(metadata) => { - return Ok(ProviderState { - metadata, + let discovered = if config.hide_from_ui { + Self::discover_provider_from_config_url(config, candidate_issuer, http_client).await + } else { + let client = JwksAcceptClient { + inner: http_client, + discovery_url: Some( + issuer_url + .join(".well-known/openid-configuration") + .map_err(|err| err.to_string())?, + ), + }; + ProviderMetadataWithLogout::discover_async(issuer_url.clone(), &client) + .await + .map(|metadata| ProviderState { + metadata: DiscoveredProviderMetadata::Console(Box::new(metadata)), discovered_at: Instant::now(), - }); - } - Err(DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason))) => { - return Err(format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}")); + }) + .map_err(|err| match err { + DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason)) => { + format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}") + } + err => format!("discovery failed: {err}"), + }) + }; + match discovered { + Ok(state) => return Ok(state), + Err(error) + if error.starts_with(OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY) + || error.starts_with(OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY) => + { + return Err(error); } Err(error) => { - let error = format!("discovery failed: {error}"); let is_transient_transport = error.contains("Request failed"); let should_retry = is_transient_transport && attempt + 1 < OIDC_DISCOVERY_TRANSPORT_RETRIES; if should_retry { @@ -1933,7 +2077,14 @@ impl OidcSys { http_client: &ReqwestHttpClient, ) -> Result { let issuer_url = IssuerUrl::new(issuer.trim().to_string()).map_err(|e| format!("invalid issuer URL: {e}"))?; - let discovery_url = discovery_url_from_config_url(&config.config_url)?; + let explicit_issuer = config.issuer.as_deref().is_some_and(|issuer| !issuer.trim().is_empty()); + let discovery_url = if explicit_issuer { + discovery_url_from_config_url(&config.config_url)? + } else { + issuer_url + .join(".well-known/openid-configuration") + .map_err(|err| err.to_string())? + }; let request = http::Request::builder() .uri(discovery_url.to_string()) .method(http::Method::GET) @@ -1946,13 +2097,25 @@ impl OidcSys { Err(OidcHttpError::ForbiddenOutbound(reason)) => { return Err(format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}")); } - Err(err) => return Err(format!("discovery request failed: {err}")), + Err(err) => return Err(format!("discovery request failed: Request failed: {err}")), }; if response.status() != http::StatusCode::OK { return Err(format!("discovery failed: HTTP status code {} at {}", response.status(), discovery_url)); } - let provider_metadata = serde_json::from_slice::(response.body()) + if !explicit_issuer + && let Some(content_type) = response.headers().get(http::header::CONTENT_TYPE) + && !content_type.to_str().ok().is_some_and(|value| { + value + .split(';') + .next() + .is_some_and(|essence| essence.eq_ignore_ascii_case("application/json")) + }) + { + return Err("Unexpected response Content-Type: expected application/json".into()); + } + + let provider_metadata = DiscoveredProviderMetadata::parse(response.body(), config.hide_from_ui) .map_err(|err| format!("failed to parse discovery response: {err}"))?; if provider_metadata.issuer() != &issuer_url { return Err(format!( @@ -1962,11 +2125,23 @@ impl OidcSys { )); } - let jwks_url = jwks_url_from_config_url(&config.config_url, &issuer_url, provider_metadata.jwks_uri())?; - let jwks = match CoreJsonWebKeySet::fetch_async(&jwks_url, http_client).await { + let jwks_url = if explicit_issuer { + jwks_url_from_config_url(&config.config_url, &issuer_url, provider_metadata.jwks_uri())? + } else { + provider_metadata.jwks_uri().clone() + }; + let jwks = match CoreJsonWebKeySet::fetch_async( + &jwks_url, + &JwksAcceptClient { + inner: http_client, + discovery_url: None, + }, + ) + .await + { Ok(jwks) => jwks, Err(DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason))) => { - return Err(format!("JWKS request blocked by outbound policy: {reason}")); + return Err(format!("{OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY}: {reason}")); } Err(err) => return Err(format!("failed to fetch JWKS: {err}")), }; @@ -2038,7 +2213,7 @@ pub async fn validate_oidc_provider_config_with_extra_root_ca( Ok(OidcProviderValidationResult { issuer: state.metadata.issuer().to_string(), - authorization_endpoint: state.metadata.authorization_endpoint().to_string(), + authorization_endpoint: state.metadata.authorization_endpoint(), token_endpoint: state.metadata.token_endpoint().map(ToString::to_string), }) } @@ -2598,7 +2773,7 @@ mod tests { } } - fn read_mock_oidc_request_path(stream: &mut impl std::io::Read) -> String { + fn read_mock_oidc_request(stream: &mut impl std::io::Read) -> String { let mut request_bytes = Vec::new(); let mut buffer = [0u8; 4096]; loop { @@ -2615,8 +2790,11 @@ mod tests { break; } } - let request = String::from_utf8_lossy(&request_bytes); - request + String::from_utf8_lossy(&request_bytes).into_owned() + } + + fn read_mock_oidc_request_path(stream: &mut impl std::io::Read) -> String { + read_mock_oidc_request(stream) .lines() .next() .unwrap_or("") @@ -2627,7 +2805,7 @@ mod tests { } fn mock_oidc_response(path: &str, discovery_body: &str, expected_jwks_path: &str, jwks_body: &str) -> String { - let (status, body) = if path.contains("/.well-known/openid-configuration") { + let (status, body) = if path.ends_with("/.well-known/openid-configuration") { (200, discovery_body) } else if path == expected_jwks_path { (200, jwks_body) @@ -2646,6 +2824,7 @@ mod tests { build_discovery_issuer: F, max_requests: usize, signing_alg: &'static str, + workload: bool, jwks_response: J, ) -> Option<(String, std::thread::JoinHandle<()>)> where @@ -2670,7 +2849,7 @@ mod tests { }; let base = format!("http://{}", listener.local_addr().expect("listener local address should be available")); let (discovery_issuer, discovery_jwks_uri, expected_jwks_path) = build_discovery_issuer(&base); - let discovery_body = serde_json::json!({ + let mut discovery_document = serde_json::json!({ "issuer": discovery_issuer, "authorization_endpoint": format!("{base}/authorize"), "token_endpoint": format!("{base}/token"), @@ -2679,8 +2858,14 @@ mod tests { "response_modes_supported": ["query"], "subject_types_supported": ["public"], "id_token_signing_alg_values_supported": [signing_alg], - }) - .to_string(); + }); + if workload { + let fields = discovery_document.as_object_mut().expect("mock metadata is an object"); + fields.remove("authorization_endpoint"); + fields.remove("token_endpoint"); + fields.insert("response_types_supported".into(), serde_json::json!(["id_token"])); + } + let discovery_body = discovery_document.to_string(); let (ready_tx, ready_rx) = mpsc::channel(); let handle = std::thread::spawn(move || { @@ -2722,12 +2907,41 @@ mod tests { .set_read_timeout(Some(Duration::from_secs(1))) .expect("failed to set discovery mock read timeout"); - let path = read_mock_oidc_request_path(&mut stream); + let request = read_mock_oidc_request(&mut stream); + let path = request + .lines() + .next() + .and_then(|line| line.split_whitespace().nth(1)) + .unwrap_or(""); let jwks_body = jwks_response(jwks_fetches); if path == expected_jwks_path { jwks_fetches += 1; } - let response = mock_oidc_response(&path, &discovery_body, &expected_jwks_path, &jwks_body); + let mut response = mock_oidc_response(path, &discovery_body, &expected_jwks_path, &jwks_body); + if path.contains("/.well-known/openid-configuration") { + assert!( + request + .lines() + .filter_map(|line| line.split_once(':')) + .any(|(name, value)| { name.eq_ignore_ascii_case("accept") && value.trim() == "application/json" }), + "discovery Accept must remain unchanged" + ); + } + if path == expected_jwks_path { + let expected_type = if workload { + "application/jwk-set+json" + } else { + "application/json" + }; + let accepts_type = request.lines().filter_map(|line| line.split_once(':')).any(|(name, value)| { + name.eq_ignore_ascii_case("accept") && value.split(',').any(|item| item.trim() == expected_type) + }); + if !accepts_type { + response = "HTTP/1.1 406 Not Acceptable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".into(); + } else if workload { + response = response.replace("Content-Type: application/json", "Content-Type: application/jwk-set+json"); + } + } let _ = stream.write_all(response.as_bytes()); let _ = stream.flush(); let _ = stream.shutdown(Shutdown::Both); @@ -2752,7 +2966,7 @@ mod tests { where F: Fn(&str) -> (String, String, String) + Send + 'static, { - start_mock_oidc_discovery_server_with_jwks(build_discovery_issuer, max_requests, "RS256", |_| { + start_mock_oidc_discovery_server_with_jwks(build_discovery_issuer, max_requests, "RS256", false, |_| { r#"{"keys":[]}"#.to_string() }) } @@ -2779,6 +2993,7 @@ mod tests { |base| (base.to_string(), format!("{base}/jwks"), "/jwks".to_string()), 4, "ES256", + false, move |fetch| { if fetch == 0 { initial_jwks.clone() @@ -2833,6 +3048,391 @@ mod tests { handle.join().expect("rotating JWKS mock server should exit cleanly"); } + #[tokio::test] + async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + for hidden in [false, true] { + for explicit in [false, true] { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let issuer = if explicit { + "https://issuer.example.com".to_string() + } else { + base.clone() + }; + let mut config = + build_mocked_oidc_provider_config("console", &format!("{base}/.well-known/openid-configuration")); + config.hide_from_ui = hidden; + config.issuer = explicit.then(|| issuer.clone()); + config.client_secret = Some(Nonce::new_random().secret().clone()); + let server_config = config.clone(); + let server_base = base.clone(); + let redirect = "https://console.example.com/oauth_callback"; + let (key, jwk) = oidc_es256_key_and_jwk("console"); + let (auth_tx, auth_rx) = tokio::sync::oneshot::channel::>(); + let server = tokio::spawn(async move { + let mut auth_rx = Some(auth_rx); + for expected_path in ["/.well-known/openid-configuration", "/jwks", "/token"] { + let (mut stream, _) = tokio::time::timeout(StdDuration::from_secs(10), listener.accept()) + .await + .unwrap() + .unwrap(); + let mut bytes = Vec::new(); + let header_end = loop { + bytes.push(stream.read_u8().await.unwrap()); + assert!(bytes.len() < 8192); + if bytes.ends_with(b"\r\n\r\n") { + break bytes.len(); + } + }; + let headers = String::from_utf8(bytes).unwrap(); + let request_line = headers.lines().next().unwrap(); + assert_eq!(request_line.split_whitespace().nth(1), Some(expected_path)); + let headers_map: HashMap<_, _> = headers + .lines() + .skip(1) + .filter_map(|line| line.split_once(':')) + .map(|(name, value)| (name.to_ascii_lowercase(), value.trim().to_string())) + .collect(); + let body = match expected_path { + "/.well-known/openid-configuration" => { + assert!(request_line.starts_with("GET ")); + assert_eq!(headers_map["accept"], "application/json"); + serde_json::json!({ + "issuer": issuer, "authorization_endpoint": format!("{server_base}/authorize"), + "token_endpoint": format!("{server_base}/token"), "jwks_uri": format!("{server_base}/jwks"), + "response_types_supported": ["code"], "subject_types_supported": ["public"], + "id_token_signing_alg_values_supported": ["ES256"] + }) + } + "/jwks" => { + assert!(request_line.starts_with("GET ")); + assert!(headers_map["accept"].contains("application/json")); + serde_json::json!({"keys": [jwk]}) + } + "/token" => { + assert!(request_line.starts_with("POST ")); + assert_eq!(headers_map["accept"], "application/json"); + assert!(headers_map["content-type"].starts_with("application/x-www-form-urlencoded")); + let length: usize = headers_map["content-length"].parse().unwrap(); + assert!(header_end + length < 16384); + let mut body = vec![0; length]; + stream.read_exact(&mut body).await.unwrap(); + let form: HashMap = url::form_urlencoded::parse(&body).into_owned().collect(); + assert_eq!(form["grant_type"], "authorization_code"); + assert_eq!(form["code"], "test-authorization-code"); + assert_eq!(form["client_id"], server_config.client_id); + assert_eq!(Some(&form["client_secret"]), server_config.client_secret.as_ref()); + assert_eq!(form["redirect_uri"], redirect); + let auth = auth_rx.take().unwrap().await.unwrap(); + let challenge = PkceCodeChallenge::from_code_verifier_sha256(&PkceCodeVerifier::new(form["code_verifier"].clone())); + assert_eq!(challenge.as_str(), auth["code_challenge"]); + let now = time::OffsetDateTime::now_utc().unix_timestamp(); + let mut header = Header::new(Algorithm::ES256); + header.kid = Some("console".into()); + let token = jsonwebtoken::encode(&header, &serde_json::json!({ + "iss": issuer, "sub": "existing-user", "aud": server_config.client_id, + "iat": now, "exp": now + 300, "nonce": auth["nonce"], + "email": "user@example.com", "groups": ["readwrite"] + }), &key).unwrap(); + serde_json::json!({"access_token": "test-access-token", "token_type": "Bearer", "id_token": token}) + } + _ => unreachable!(), + }.to_string(); + stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap(); + } + }); + let http_client = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).unwrap()); + let discovered = OidcSys::discover_provider(&config, &http_client).await.unwrap(); + let sys = OidcSys { + configs: HashMap::from([(config.id.clone(), config)]), + provider_states: RwLock::new(HashMap::from([("console".into(), discovered)])), + state_store: OidcStateStore::new(), + http_client, + }; + let auth_url = sys.authorize_url("console", redirect, Some("/buckets".into())).await.unwrap(); + let auth_url = Url::parse(&auth_url).unwrap(); + assert_eq!(auth_url.as_str().split('?').next(), Some(format!("{base}/authorize").as_str())); + let auth: HashMap = auth_url.query_pairs().into_owned().collect(); + assert_eq!(auth["response_type"], "code"); + assert_eq!(auth["client_id"], "rustfs-oidc-test"); + assert!(!auth["nonce"].is_empty()); + assert!(!auth["state"].is_empty()); + assert_eq!(auth["redirect_uri"], redirect); + assert_eq!(auth["code_challenge_method"], "S256"); + assert!(auth["scope"].split_whitespace().any(|scope| scope == "openid")); + let state = auth["state"].clone(); + auth_tx.send(auth).unwrap(); + let (claims, provider, session, _) = sys + .exchange_code(&state, "test-authorization-code", redirect) + .await + .unwrap_or_else(|err| panic!("hidden={hidden}, explicit={explicit}: {err}")); + assert_eq!(provider, "console"); + assert_eq!(claims.sub, "existing-user"); + assert_eq!(claims.email, "user@example.com"); + assert_eq!(claims.groups, vec!["readwrite"]); + assert_eq!(session.redirect_after.as_deref(), Some("/buckets")); + assert!(matches!(sys.exchange_code(&state, "test-authorization-code", redirect).await, + Err(error) if error == "invalid or expired OIDC state")); + server.await.unwrap(); + } + } + } + + #[test] + fn workload_metadata_requires_hidden_provider_and_valid_verification_fields() { + let document = serde_json::json!({ + "issuer": "https://issuer.example.com", + "jwks_uri": "https://issuer.example.com/jwks", + "response_types_supported": ["id_token"], + "subject_types_supported": ["public"], + "id_token_signing_alg_values_supported": ["ES256"], + }); + let parse = + |value: &serde_json::Value, hidden| DiscoveredProviderMetadata::parse(&serde_json::to_vec(value).unwrap(), hidden); + let metadata = parse(&document, true).expect("hidden workload metadata should parse"); + assert!(metadata.authorization_endpoint().is_none()); + assert!(metadata.console().err().unwrap().contains("only web identity")); + assert!(parse(&document, false).err().unwrap().contains("authorization_endpoint")); + for field in ["issuer", "jwks_uri", "id_token_signing_alg_values_supported"] { + let mut invalid = document.clone(); + invalid.as_object_mut().unwrap().remove(field); + assert!(parse(&invalid, true).err().unwrap().contains(field), "missing {field}"); + } + for endpoint in [serde_json::Value::Null, serde_json::json!(""), serde_json::json!("not a URL")] { + let mut invalid = document.clone(); + invalid["authorization_endpoint"] = endpoint; + assert!(parse(&invalid, true).is_err(), "invalid endpoint must not select workload metadata"); + } + let mut complete = document; + complete["authorization_endpoint"] = serde_json::json!("https://issuer.example.com/authorize"); + for hidden in [true, false] { + let metadata = parse(&complete, hidden).expect("full providers keep the existing parser"); + assert!(metadata.console().is_ok()); + assert!(metadata.token_endpoint().is_none(), "token endpoint remains optional"); + } + } + + #[test] + fn workload_metadata_rejects_duplicate_fields() { + for hidden in [false, true] { + let document = format!( + r#"{{"issuer":"https://wrong.example.com","issuer":"https://issuer.example.com",{}"jwks_uri":"https://issuer.example.com/jwks","response_types_supported":["id_token"],"subject_types_supported":["public"],"id_token_signing_alg_values_supported":["ES256"]}}"#, + if hidden { + "" + } else { + r#""authorization_endpoint":"https://issuer.example.com/authorize","# + }, + ); + let error = DiscoveredProviderMetadata::parse(document.as_bytes(), hidden) + .err() + .expect("duplicate issuer must fail"); + assert!(error.contains("duplicate field"), "{error}"); + } + } + + #[test] + fn workload_verifier_preserves_algorithm_secret_and_audience_policy() { + let secret = Nonce::new_random().secret().to_string(); + let mut config = build_mocked_oidc_provider_config("workload", "https://issuer.example.com"); + config.client_secret = Some(secret.clone()); + config.other_audiences = vec!["additional-audience".into()]; + let now = time::OffsetDateTime::now_utc().unix_timestamp(); + let payload = serde_json::json!({ + "iss": config.config_url, "sub": "repo:example/project:ref:refs/heads/main", + "aud": [config.client_id, "additional-audience"], "iat": now, "exp": now + 300, + }); + let signed = + jsonwebtoken::encode(&Header::new(Algorithm::HS256), &payload, &EncodingKey::from_secret(secret.as_bytes())).unwrap(); + let token: CoreIdToken = signed.parse().unwrap(); + for workload in [false, true] { + for (algorithms, accepted) in [ + (serde_json::json!(["HS256", "unsupported-future-algorithm"]), true), + (serde_json::json!(["ES256"]), false), + (serde_json::json!(["unsupported-future-algorithm"]), false), + ] { + let mut document = serde_json::json!({ + "issuer": config.config_url, "jwks_uri": "https://issuer.example.com/jwks", + "response_types_supported": ["id_token"], "subject_types_supported": ["public"], + "id_token_signing_alg_values_supported": algorithms, + }); + if !workload { + document["authorization_endpoint"] = serde_json::json!("https://issuer.example.com/authorize"); + } + let metadata = DiscoveredProviderMetadata::parse(&serde_json::to_vec(&document).unwrap(), true).unwrap(); + let verifier = metadata + .verifier(&config) + .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud)); + assert_eq!( + token.claims(&verifier, |_: Option<&Nonce>| Ok(())).is_ok(), + accepted, + "workload={workload}, algorithms={algorithms}" + ); + if accepted { + assert!( + token.claims(&metadata.verifier(&config), |_: Option<&Nonce>| Ok(())).is_err(), + "additional audiences require explicit trust" + ); + let mut wrong_secret = config.clone(); + wrong_secret.client_secret = Some(Nonce::new_random().secret().to_string()); + let verifier = metadata + .verifier(&wrong_secret) + .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud)); + assert!( + token.claims(&verifier, |_: Option<&Nonce>| Ok(())).is_err(), + "incorrect client secret must fail" + ); + } + } + } + } + + #[tokio::test] + async fn workload_discovery_stops_after_forbidden_jwks() { + let requests = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = Arc::clone(&requests); + let (base, handle) = start_mock_oidc_discovery_server_with_jwks( + |base| (base.to_string(), "http://192.168.65.254:8080/jwks".into(), "/jwks".into()), + 2, + "ES256", + true, + move |_| { + seen.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + serde_json::json!({"keys": []}).to_string() + }, + ) + .expect("workload discovery mock must bind"); + let mut config = build_mocked_oidc_provider_config("workload", &base); + config.hide_from_ui = true; + let client = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).unwrap()); + let error = OidcSys::discover_provider(&config, &client) + .await + .err() + .expect("private JWKS must be blocked"); + assert!(error.starts_with(OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY), "{error}"); + handle.join().unwrap(); + assert_eq!( + requests.load(std::sync::atomic::Ordering::SeqCst), + 1, + "a policy denial must not retry discovery" + ); + } + + #[tokio::test] + async fn workload_config_validation_reports_absent_console_endpoints() { + let (base, handle) = start_mock_oidc_discovery_server_with_jwks( + |base| (base.to_string(), format!("{base}/jwks"), "/jwks".into()), + 2, + "ES256", + true, + |_| serde_json::json!({"keys": []}).to_string(), + ) + .expect("workload discovery mock must bind"); + let mut config = build_mocked_oidc_provider_config("workload", &base); + config.hide_from_ui = true; + // Inferred issuers keep the library's discovery URL construction. + config.config_url = format!("{base}/.well-known/openid-configuration?ignored=1#ignored"); + let result = validate_mocked_oidc_provider_config(&config) + .await + .expect("hidden workload configuration should validate"); + assert_eq!(result.issuer, base); + assert!(result.authorization_endpoint.is_none()); + assert!(result.token_endpoint.is_none()); + handle.join().unwrap(); + } + + #[tokio::test] + async fn workload_discovery_verification_and_rotation() { + for explicit_issuer in [false, true] { + let (_, initial_jwk) = oidc_es256_key_and_jwk("initial"); + let (key, rotated_jwk) = oidc_es256_key_and_jwk("rotated"); + let initial_jwks = serde_json::json!({"keys": [initial_jwk]}).to_string(); + let rotated_jwks = serde_json::json!({"keys": [rotated_jwk]}).to_string(); + let (base, handle) = start_mock_oidc_discovery_server_with_jwks( + |base| (base.to_string(), format!("{base}/jwks"), "/jwks".into()), + 4, + "ES256", + true, + move |fetch| { + if fetch == 0 { + initial_jwks.clone() + } else { + rotated_jwks.clone() + } + }, + ) + .expect("workload discovery mock must bind"); + let mut config = build_mocked_oidc_provider_config("workload", &base); + config.hide_from_ui = true; + if explicit_issuer { + config.issuer = Some(base.clone()); + } + let http_client = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).unwrap()); + let state = OidcSys::discover_provider(&config, &http_client) + .await + .expect("workload discovery must succeed"); + assert!(state.metadata.authorization_endpoint().is_none()); + let sys = OidcSys { + configs: HashMap::from([(config.id.clone(), config.clone())]), + provider_states: RwLock::new(HashMap::from([(config.id.clone(), state)])), + state_store: OidcStateStore::new(), + http_client, + }; + assert!(sys.has_providers()); + assert!(sys.list_visible_providers().is_empty()); + let error = sys + .authorize_url(&config.id, "https://console.example.com/callback", None) + .await + .unwrap_err(); + assert!(error.contains("only web identity")); + let now = time::OffsetDateTime::now_utc().unix_timestamp(); + let payload = serde_json::json!({ + "iss": base, "sub": "system:serviceaccount:default:reader", "aud": [config.client_id], + "iat": now, "exp": now + 300, "groups": ["readonly"], + "kubernetes.io": {"namespace": "default", "serviceaccount": {"name": "reader"}}, + }); + let mut header = Header::new(Algorithm::ES256); + header.kid = Some("rotated".into()); + let token = jsonwebtoken::encode(&header, &payload, &key).unwrap(); + let (claims, provider) = sys + .verify_web_identity_token(&token) + .await + .expect("rotation must retain workload discovery support"); + assert_eq!(provider, config.id); + assert_eq!(claims.sub, "system:serviceaccount:default:reader"); + assert_eq!(claims.groups, ["readonly"]); + // Repeat after the mock exits: the verified snapshot must be cached. + handle.join().unwrap(); + assert!(sys.verify_web_identity_token(&token).await.is_ok()); + for (field, value, expected) in [ + ("iss", serde_json::json!("https://wrong.example.com"), "issuer"), + ("aud", serde_json::json!("wrong-audience"), "audience"), + ("exp", serde_json::json!(now - 60), "expired"), + ] { + let mut invalid = payload.clone(); + invalid[field] = value; + let token = jsonwebtoken::encode(&header, &invalid, &key).unwrap(); + let error = sys + .verify_web_identity_token(&token) + .await + .expect_err("invalid workload token must fail"); + assert!(error.to_lowercase().contains(expected), "{field}: {error}"); + } + let (wrong_key, _) = oidc_es256_key_and_jwk("wrong"); + let invalid = jsonwebtoken::encode(&header, &payload, &wrong_key).unwrap(); + let error = sys + .verify_web_identity_token(&invalid) + .await + .expect_err("wrong signature must fail"); + assert!(error.to_lowercase().contains("signature"), "{error}"); + assert!( + sys.verify_web_identity_token(&token).await.is_ok(), + "failed refresh must preserve the cached keys" + ); + } + } + fn start_mock_oidc_tls_discovery_server( build_discovery_issuer: F, max_requests: usize, @@ -2958,7 +3558,7 @@ mod tests { Ok(OidcProviderValidationResult { issuer: state.metadata.issuer().to_string(), - authorization_endpoint: state.metadata.authorization_endpoint().to_string(), + authorization_endpoint: state.metadata.authorization_endpoint(), token_endpoint: state.metadata.token_endpoint().map(ToString::to_string), }) } @@ -3558,7 +4158,7 @@ mod tests { provider_states: RwLock::new(HashMap::from([( provider_id.to_string(), ProviderState { - metadata, + metadata: DiscoveredProviderMetadata::Console(Box::new(metadata)), discovered_at: Instant::now(), }, )])), diff --git a/docs/operations/oidc-console-integration.md b/docs/operations/oidc-console-integration.md index d4aec6274..3206b7800 100644 --- a/docs/operations/oidc-console-integration.md +++ b/docs/operations/oidc-console-integration.md @@ -39,7 +39,7 @@ Every provider key can be set as `RUSTFS_IDENTITY_OPENID_` in the process e | `email_claim`, `username_claim` | `RUSTFS_IDENTITY_OPENID_EMAIL_CLAIM`, `RUSTFS_IDENTITY_OPENID_USERNAME_CLAIM` | Identity claims shown in the Console. | | `role_policy` | `RUSTFS_IDENTITY_OPENID_ROLE_POLICY` | One fixed policy for every login from this provider. Connectivity testing only. | | `display_name` | `RUSTFS_IDENTITY_OPENID_DISPLAY_NAME` | Login button label. | -| `hide_from_ui` | `RUSTFS_IDENTITY_OPENID_HIDE_FROM_UI` | Hides the provider from `/oidc/providers`. | +| `hide_from_ui` | `RUSTFS_IDENTITY_OPENID_HIDE_FROM_UI` | Hides the provider from `/oidc/providers`. Required (`on`) for STS workload issuers whose discovery document omits `authorization_endpoint`; see [workload provider requirements](oidc-provider-requirements.md#sts-workload-providers). Complete hidden providers still support direct Console login. | Process-level settings (environment only, never suffixed per provider): diff --git a/docs/operations/oidc-provider-requirements.md b/docs/operations/oidc-provider-requirements.md index 11653fc04..1882df3da 100644 --- a/docs/operations/oidc-provider-requirements.md +++ b/docs/operations/oidc-provider-requirements.md @@ -7,6 +7,8 @@ RustFS is a standard OpenID Connect relying party using the authorization-code f ## Requirements +The table below describes Console login. For endpoint-free STS issuers, use the [workload contract](#sts-workload-providers) below. + | # | Requirement | Details | Code anchor | | --- | --- | --- | --- | | 1 | Discovery document | `RUSTFS_IDENTITY_OPENID_CONFIG_URL` names the provider (issuer base or full discovery URL); RustFS fetches `{issuer}/.well-known/openid-configuration` and needs `issuer`, `authorization_endpoint`, `token_endpoint`, `jwks_uri`, and the standard `*_supported` arrays. When `RUSTFS_IDENTITY_OPENID_ISSUER` is set, the document's `issuer` must equal it exactly; otherwise RustFS tries the issuer candidates derived from the config URL. | `crates/iam/src/oidc.rs` `discover_provider`, `discover_provider_from_config_url` | @@ -20,6 +22,16 @@ RustFS is a standard OpenID Connect relying party using the authorization-code f | 9 | Registered redirect URI | The provider must accept the callback `{public-origin}/rustfs/admin/v3/oidc/callback/{provider_id}`. RustFS picks the origin in this order: the provider's `redirect_uri` (`RUSTFS_IDENTITY_OPENID_REDIRECT_URI`), then `RUSTFS_BROWSER_REDIRECT_URL`, then the request's own scheme and host — the last only when `RUSTFS_IDENTITY_OPENID_REDIRECT_URI_DYNAMIC` is enabled. | `rustfs/src/admin/handlers/oidc.rs` `derive_callback_uri_with_provider_config`, `browser_redirect_url` | | 10 | Logout endpoint (optional) | When discovery advertises `end_session_endpoint`, RustFS builds an RP-initiated logout URL with `id_token_hint`, `client_id`, and `post_logout_redirect_uri`. Without it, logout falls back to the console login page. | `crates/iam/src/oidc.rs` `build_logout_url` (reads `end_session_endpoint` from `ProviderMetadataWithLogout`) | +## STS workload providers + +For workload identity tokens exchanged through `AssumeRoleWithWebIdentity` (for example Kubernetes service-account tokens), set `RUSTFS_IDENTITY_OPENID_HIDE_FROM_UI=on` (`hide_from_ui=on` in persisted configuration, or `hide_from_ui: true` in the admin JSON API) when discovery omits `authorization_endpoint`. The default is off; without this setting, RustFS applies the Console discovery contract and rejects the missing endpoint. + +Endpoint-free workload discovery requires `issuer`, `jwks_uri`, and `id_token_signing_alg_values_supported`. `token_endpoint` is optional; browser authorization endpoints, `response_types_supported`, and `subject_types_supported` are not required for this path. An explicitly null, empty, or malformed `authorization_endpoint` is rejected rather than treated as absent. Configuration validation returns `authorization_endpoint: null` for an accepted workload-only provider. + +Set `RUSTFS_IDENTITY_OPENID_CONFIG_URL` to the issuer/discovery URL and `RUSTFS_IDENTITY_OPENID_CLIENT_ID` to the intended token audience. Configure policy claims or `RUSTFS_IDENTITY_OPENID_ROLE_POLICY` to grant the required RustFS permissions. Signature, issuer, audience, and expiration verification remain enforced, and private endpoints still require the outbound allowlist. JWKS requests accept both `application/json` and `application/jwk-set+json`. + +Workload-only providers cannot perform Console authorization-code login. Hiding a provider with complete discovery metadata only hides its listing; its existing direct Console login remains available. + ## Deployment notes - Behind a load balancer, authorize and callback requests must reach the same RustFS node while the `state` is in flight, or set `RUSTFS_BROWSER_REDIRECT_URL` so the callback URL is stable; the callback error text names both remedies. diff --git a/rustfs/src/admin/handlers/oidc.rs b/rustfs/src/admin/handlers/oidc.rs index 19fc7f951..9ed5afdc4 100644 --- a/rustfs/src/admin/handlers/oidc.rs +++ b/rustfs/src/admin/handlers/oidc.rs @@ -465,7 +465,7 @@ impl Operation for ValidateOidcConfigHandler { valid: true, message: "OIDC configuration is valid".to_string(), issuer: Some(validation.issuer), - authorization_endpoint: Some(validation.authorization_endpoint), + authorization_endpoint: validation.authorization_endpoint, token_endpoint: validation.token_endpoint, }, ) @@ -1296,6 +1296,101 @@ mod tests { use http::{Extensions, HeaderMap, HeaderValue, Uri}; use temp_env::with_var; + #[tokio::test] + #[serial_test::serial] + async fn validate_handler_preserves_workload_null_and_console_endpoints() { + use crate::admin::runtime_sources::{AppContext, publish_test_app_context}; + use http_body_util::BodyExt as _; + use rustfs_iam::store::{Store as _, object::IAM_CONFIG_PREFIX}; + use std::sync::Arc; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + // The admin URL boundary rejects literal loopback hosts. A local proxy + // serves the public-shaped test origin without external DNS or traffic. + let proxy = format!("http://{}", listener.local_addr().unwrap()); + let base = "http://oidc-handler.example.invalid".to_string(); + temp_env::async_with_vars( + [("RUSTFS_OUTBOUND_ALLOW_ORIGINS", Some(base.as_str())), + ("HTTP_PROXY", Some(proxy.as_str())), ("http_proxy", Some(proxy.as_str())), + ("HTTPS_PROXY", None), ("https_proxy", None), ("ALL_PROXY", None), ("all_proxy", None), + ("NO_PROXY", Some("")), ("no_proxy", Some(""))], + async { + let _ = rustfs_credentials::init_global_action_credentials(Some("OIDCVALIDATEROOT".into()), Some("oidcValidateRootSecret123".into())); + let env = rustfs_test_utils::TestECStoreEnv::builder().prefix("oidc_validate_handler") + .disk_count(1).init_bucket_metadata(false).build().await; + rustfs_iam::store::object::ObjectStore::new(Arc::clone(&env.ecstore)) + .save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX)).await.unwrap(); + let iam = rustfs_iam::init_iam_sys(Arc::clone(&env.ecstore)).await.unwrap(); + publish_test_app_context(Arc::new(AppContext::with_default_interfaces( + Arc::clone(&env.ecstore), iam, Arc::new(rustfs_kms::KmsServiceManager::new()), + ))); + let server_base = base.clone(); + let server = tokio::spawn(async move { + for path in ["/.well-known/openid-configuration", "/jwks", "/.well-known/openid-configuration", "/complete/.well-known/openid-configuration", "/complete/jwks"] { + let (mut stream, _) = tokio::time::timeout(std::time::Duration::from_secs(15), listener.accept()).await.unwrap().unwrap(); + let mut request = Vec::new(); + while !request.ends_with(b"\r\n\r\n") { + request.push(stream.read_u8().await.unwrap()); + assert!(request.len() < 8192); + } + let request = String::from_utf8(request).unwrap(); + let target = Url::parse(request.lines().next().unwrap().split_whitespace().nth(1).unwrap()).unwrap(); + assert_eq!(target.origin().ascii_serialization(), server_base); + assert_eq!(target.path(), path); + let mut body = if path.ends_with("/jwks") { serde_json::json!({"keys": []}) } else { + serde_json::json!({"issuer": server_base, "jwks_uri": format!("{server_base}/jwks"), "id_token_signing_alg_values_supported": ["RS256"]}) + }; + if path == "/complete/.well-known/openid-configuration" { + body["authorization_endpoint"] = serde_json::json!(format!("{server_base}/authorize")); + body["token_endpoint"] = serde_json::json!(format!("{server_base}/token")); + body["response_types_supported"] = serde_json::json!(["code"]); + body["subject_types_supported"] = serde_json::json!(["public"]); + } + let body = body.to_string(); + stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap(); + } + }); + for (hidden, complete) in [(true, false), (false, false), (false, true)] { + let document = serde_json::json!({"provider_id": "workload", "client_id": "rustfs-test", "issuer": base, + "config_url": format!("{base}{}/.well-known/openid-configuration", if complete { "/complete" } else { "" }), "hide_from_ui": hidden}); + let request = || { + let mut req = build_oidc_request("/rustfs/admin/v3/oidc/validate", None, None); + req.method = Method::POST; + req.input = Body::from(document.to_string()); + req + }; + let denied = ValidateOidcConfigHandler {}.call(request(), Params::new()).await.unwrap_err(); + assert_eq!(denied.code(), &S3ErrorCode::InvalidRequest); + assert_eq!(denied.message(), Some("authentication required")); + let mut req = request(); + req.credentials = Some(s3s::auth::Credentials { access_key: "OIDCVALIDATEROOT".into(), secret_key: "oidcValidateRootSecret123".into() }); + let result = ValidateOidcConfigHandler {}.call(req, Params::new()).await; + if !hidden && !complete { + let err = result.unwrap_err(); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + assert!(err.message().unwrap().contains("authorization_endpoint")); + continue; + } + let (status, body) = result.unwrap().output; + assert_eq!(status, StatusCode::OK); + let body = body.collect().await.unwrap().to_bytes(); + let response: serde_json::Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(response["valid"], true); + assert_eq!(response["issuer"], base); + if complete { + assert_eq!(response["authorization_endpoint"], format!("{base}/authorize")); + assert_eq!(response["token_endpoint"], format!("{base}/token")); + } else { + assert_eq!(response.get("authorization_endpoint"), Some(&serde_json::Value::Null)); + assert_eq!(response.get("token_endpoint"), Some(&serde_json::Value::Null)); + } + } + server.await.unwrap(); + }, + ).await; + } + fn build_oidc_request( uri: &'static str, host: Option<&'static str>,