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/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) { 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" + ); +} 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/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index f2337c3ea..d96f90353 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", @@ -2216,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/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.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() { 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() { 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(), 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>,