From 5228463a1c294c2efe41d5af8f6db9e1d77bcd4a Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Mon, 12 Oct 2026 18:40:07 +0800 Subject: [PATCH] fix(oidc): recover providers after startup discovery failures (#8475) * fix(oidc): recover providers after startup discovery failures * fix(oidc): report recovery failures with throttled diagnostics --- crates/iam/src/oidc/provider.rs | 139 ++++++-- crates/iam/src/oidc/provider_tests.rs | 1 + crates/iam/src/oidc/recovery_tests.rs | 489 ++++++++++++++++++++++++++ crates/iam/src/oidc/runtime.rs | 105 +++++- crates/iam/src/oidc/runtime_tests.rs | 68 +++- crates/iam/src/oidc/test_support.rs | 1 + crates/iam/src/oidc/transport.rs | 55 +-- rustfs/src/startup_auth.rs | 72 +++- rustfs/src/startup_lifecycle.rs | 13 + rustfs/src/startup_services.rs | 6 +- 10 files changed, 874 insertions(+), 75 deletions(-) create mode 100644 crates/iam/src/oidc/recovery_tests.rs diff --git a/crates/iam/src/oidc/provider.rs b/crates/iam/src/oidc/provider.rs index 6f0c72a32..6b4f083f8 100644 --- a/crates/iam/src/oidc/provider.rs +++ b/crates/iam/src/oidc/provider.rs @@ -13,7 +13,7 @@ // limitations under the License. use super::{ config::{OidcProviderConfig, OidcProviderValidationResult, SourcedOidcProviderConfig}, - transport::{OidcHttpError, ReqwestHttpClient, oidc_extra_root_certs}, + transport::{DiscoveryRequest, OidcHttpError, ReqwestHttpClient, oidc_extra_root_certs, oidc_http_error_diagnostics}, }; use crate::{EVENT_OIDC_DIAGNOSTICS, LOG_COMPONENT_IAM, LOG_SUBSYSTEM_OIDC}; use openidconnect::core::{CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm}; @@ -25,7 +25,7 @@ use std::collections::HashMap; use std::sync::RwLock; use std::time::{Duration as StdDuration, Instant}; use tokio::time::sleep; -use tracing::warn; +use tracing::{debug, warn}; use url::Url; pub(super) const OIDC_JWKS_REFRESH_INTERVAL: StdDuration = StdDuration::from_secs(24 * 60 * 60); @@ -34,6 +34,55 @@ pub(super) const OIDC_DISCOVERY_TRANSPORT_RETRY_DELAY: StdDuration = StdDuration pub(super) const OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY: &str = "JWKS request blocked by outbound policy"; pub(super) const OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY: &str = "OIDC provider discovery blocked by outbound policy"; +/// Preserve recovery eligibility before discovery errors become public diagnostic strings. +#[derive(Debug, thiserror::Error)] +#[error("{message}")] +pub(super) struct DiscoveryFailure { + message: String, + pub(super) retryable: bool, + pub(super) diagnostics: (&'static str, String), +} + +impl From for DiscoveryFailure { + fn from(message: String) -> Self { + Self { + message, + retryable: false, + diagnostics: ("validation", String::new()), + } + } +} + +fn retryable_discovery_status(status: http::StatusCode) -> bool { + status.is_server_error() || matches!(status, http::StatusCode::REQUEST_TIMEOUT | http::StatusCode::TOO_MANY_REQUESTS) +} + +fn retryable_discovery_transport(error: &OidcHttpError) -> bool { + matches!(error, OidcHttpError::Reqwest(error) if !error.is_builder()) +} + +fn discovery_failure(error: DiscoveryError, context: &str, blocked_context: &str) -> DiscoveryFailure { + let retryable = match &error { + DiscoveryError::Request(error) => retryable_discovery_transport(error), + DiscoveryError::Response(status, _, _) => retryable_discovery_status(*status), + _ => false, + }; + let diagnostics = match &error { + DiscoveryError::Request(error) => oidc_http_error_diagnostics(error), + DiscoveryError::Response(status, _, _) => ("http_status", status.as_u16().to_string()), + _ => ("validation", String::new()), + }; + let message = match error { + DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason)) => format!("{blocked_context}: {reason}"), + error => format!("{context}: {error}"), + }; + DiscoveryFailure { + message, + retryable, + diagnostics, + } +} + // ---- Internal provider state ---- /// Discovered OIDC provider metadata. @@ -180,6 +229,7 @@ impl<'c> AsyncHttpClient<'c> for JwksAcceptClient<'_> { http::HeaderValue::from_static("application/json, application/jwk-set+json"), ); } + request.extensions_mut().insert(DiscoveryRequest); self.inner.call(request) } } @@ -246,7 +296,17 @@ impl ProviderRuntime { provider_id: &str, config: &OidcProviderConfig, ) -> Result { - let state = discover_provider(config, &self.http_client).await?; + self.recover_provider_state(provider_id, config) + .await + .map_err(|err| err.to_string()) + } + + pub(super) async fn recover_provider_state( + &self, + provider_id: &str, + config: &OidcProviderConfig, + ) -> Result { + let state = discover_provider_for_recovery(config, &self.http_client).await?; let mut map = self.states.write().map_err(|e| { let msg = e.to_string(); format!("provider state lock poisoned: {msg}") @@ -301,6 +361,15 @@ pub(super) async fn discover_provider( config: &OidcProviderConfig, http_client: &ReqwestHttpClient, ) -> Result { + discover_provider_for_recovery(config, http_client) + .await + .map_err(|err| err.to_string()) +} + +pub(super) async fn discover_provider_for_recovery( + config: &OidcProviderConfig, + http_client: &ReqwestHttpClient, +) -> Result { if let Some(issuer) = config.issuer.as_deref().filter(|issuer| !issuer.trim().is_empty()) { return discover_provider_from_config_url(config, issuer, http_client).await; } @@ -310,6 +379,8 @@ pub(super) async fn discover_provider( let base_issuer = normalize_config_url(&config.config_url)?; let candidates = issuer_candidates(&base_issuer); let mut last_errors = Vec::new(); + let mut retryable = false; + let mut diagnostics = ("validation", String::new()); for candidate_issuer in candidates.iter() { let issuer_url = IssuerUrl::new(candidate_issuer.clone()).map_err(|e| format!("invalid issuer URL: {e}"))?; @@ -332,26 +403,21 @@ pub(super) async fn discover_provider( metadata: DiscoveredProviderMetadata::Console(Box::new(metadata)), discovered_at: Instant::now(), }) - .map_err(|err| match err { - DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason)) => { - format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}") - } - err => format!("discovery failed: {err}"), - }) + .map_err(|err| discovery_failure(err, "discovery failed", OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY)) }; 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) => + if error.message.starts_with(OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY) + || error.message.starts_with(OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY) => { return Err(error); } Err(error) => { - let is_transient_transport = error.contains("Request failed"); + let is_transient_transport = error.message.contains("Request failed"); let should_retry = is_transient_transport && attempt + 1 < OIDC_DISCOVERY_TRANSPORT_RETRIES; if should_retry { - warn!( + debug!( event = EVENT_OIDC_DIAGNOSTICS, component = LOG_COMPONENT_IAM, subsystem = LOG_SUBSYSTEM_OIDC, @@ -368,8 +434,12 @@ pub(super) async fn discover_provider( continue; } + if error.retryable || !retryable { + diagnostics = error.diagnostics.clone(); + } + retryable |= error.retryable; last_errors.push(format!("issuer '{candidate_issuer}': {error}")); - warn!( + debug!( event = EVENT_OIDC_DIAGNOSTICS, component = LOG_COMPONENT_IAM, subsystem = LOG_SUBSYSTEM_OIDC, @@ -388,11 +458,11 @@ pub(super) async fn discover_provider( } } - Err(format!( - "discovery failed for all issuer variants {:?}: {}", - candidates, - last_errors.join("; ") - )) + Err(DiscoveryFailure { + message: format!("discovery failed for all issuer variants {:?}: {}", candidates, last_errors.join("; ")), + retryable, + diagnostics, + }) } pub(super) async fn validate_oidc_provider_config(config: &OidcProviderConfig) -> Result { @@ -429,7 +499,7 @@ async fn discover_provider_from_config_url( config: &OidcProviderConfig, issuer: &str, http_client: &ReqwestHttpClient, -) -> Result { +) -> Result { let issuer_url = IssuerUrl::new(issuer.trim().to_string()).map_err(|e| format!("invalid issuer URL: {e}"))?; let explicit_issuer = config.issuer.as_deref().is_some_and(|issuer| !issuer.trim().is_empty()); let discovery_url = if explicit_issuer { @@ -440,6 +510,7 @@ async fn discover_provider_from_config_url( .map_err(|err| err.to_string())? }; let request = http::Request::builder() + .extension(DiscoveryRequest) .uri(discovery_url.to_string()) .method(http::Method::GET) .header(http::header::ACCEPT, "application/json") @@ -449,12 +520,22 @@ async fn discover_provider_from_config_url( let response = match http_client.call(request).await { Ok(response) => response, Err(OidcHttpError::ForbiddenOutbound(reason)) => { - return Err(format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}")); + return Err(format!("{OIDC_DISCOVERY_BLOCKED_BY_OUTBOUND_POLICY}: {reason}").into()); + } + Err(err) => { + return Err(DiscoveryFailure { + retryable: retryable_discovery_transport(&err), + diagnostics: oidc_http_error_diagnostics(&err), + message: format!("discovery request failed: 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)); + return Err(DiscoveryFailure { + retryable: retryable_discovery_status(response.status()), + diagnostics: ("http_status", response.status().as_u16().to_string()), + message: format!("discovery failed: HTTP status code {} at {}", response.status(), discovery_url), + }); } if !explicit_issuer @@ -466,7 +547,9 @@ async fn discover_provider_from_config_url( .is_some_and(|essence| essence.eq_ignore_ascii_case("application/json")) }) { - return Err("Unexpected response Content-Type: expected application/json".into()); + return Err("Unexpected response Content-Type: expected application/json" + .to_string() + .into()); } let provider_metadata = DiscoveredProviderMetadata::parse(response.body(), config.hide_from_ui) @@ -476,7 +559,8 @@ async fn discover_provider_from_config_url( "unexpected issuer URI `{}` (expected `{}`)", provider_metadata.issuer().as_str(), issuer_url.as_str() - )); + ) + .into()); } let jwks_url = if explicit_issuer { @@ -494,10 +578,7 @@ async fn discover_provider_from_config_url( .await { Ok(jwks) => jwks, - Err(DiscoveryError::Request(OidcHttpError::ForbiddenOutbound(reason))) => { - return Err(format!("{OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY}: {reason}")); - } - Err(err) => return Err(format!("failed to fetch JWKS: {err}")), + Err(err) => return Err(discovery_failure(err, "failed to fetch JWKS", OIDC_JWKS_BLOCKED_BY_OUTBOUND_POLICY)), }; Ok(ProviderState { diff --git a/crates/iam/src/oidc/provider_tests.rs b/crates/iam/src/oidc/provider_tests.rs index efa6c543b..849b0d4d5 100644 --- a/crates/iam/src/oidc/provider_tests.rs +++ b/crates/iam/src/oidc/provider_tests.rs @@ -299,6 +299,7 @@ async fn workload_discovery_verification_and_rotation() { configs: HashMap::from([(config.id.clone(), test_sourced_config(config.clone()))]), provider_runtime: ProviderRuntime::new(http_client, HashMap::from([(config.id.clone(), state)])), state_store: OidcStateStore::new(), + pending_discovery: Default::default(), }; assert!(sys.has_providers()); assert!(sys.provider_configs().all(|provider| provider.hide_from_ui)); diff --git a/crates/iam/src/oidc/recovery_tests.rs b/crates/iam/src/oidc/recovery_tests.rs new file mode 100644 index 000000000..f1b500b85 --- /dev/null +++ b/crates/iam/src/oidc/recovery_tests.rs @@ -0,0 +1,489 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use super::*; +use crate::oidc::test_support::{build_mocked_oidc_provider_config, oidc_es256_key_and_jwk, test_sourced_config}; +use jsonwebtoken::{Algorithm, EncodingKey, Header}; +use rustfs_utils::egress::OutboundPolicy; +use std::sync::atomic::{AtomicU16, AtomicUsize, Ordering}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio_util::task::AbortOnDropHandle; + +struct RecoveryIdp { + base: String, + config: OidcProviderConfig, + key: EncodingKey, + discovery_status: Arc, + jwks_status: Arc, + requests: Arc, + cancelled_requests: Arc, + _server: AbortOnDropHandle<()>, +} + +impl RecoveryIdp { + async fn start(id: &str, workload: bool, explicit_issuer: bool) -> Self { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind recovery IdP"); + let base = format!("http://{}", listener.local_addr().expect("recovery IdP address")); + let mut config = build_mocked_oidc_provider_config(id, &base); + config.hide_from_ui = workload; + config.claim_name = "groups".into(); + config.claim_prefix = "team/".into(); + config.issuer = explicit_issuer.then(|| format!("https://issuer.example.com/{id}")); + let (key, jwk) = oidc_es256_key_and_jwk(id); + let mut document = serde_json::json!({ + "issuer": config.issuer.as_deref().unwrap_or(&base), + "authorization_endpoint": format!("{base}/authorize"), + "token_endpoint": format!("{base}/token"), + "jwks_uri": format!("{base}/jwks"), + "response_types_supported": ["code"], + "subject_types_supported": ["public"], + "id_token_signing_alg_values_supported": ["ES256"], + }); + if workload { + let fields = document.as_object_mut().expect("discovery document"); + fields.remove("authorization_endpoint"); + fields.remove("token_endpoint"); + } + let document = document.to_string(); + let jwks = serde_json::json!({"keys": [jwk]}).to_string(); + let discovery_status = Arc::new(AtomicU16::new(503)); + let jwks_status = Arc::new(AtomicU16::new(200)); + let requests = Arc::new(AtomicUsize::new(0)); + let cancelled_requests = Arc::new(AtomicUsize::new(0)); + let server_discovery_status = discovery_status.clone(); + let server_jwks_status = jwks_status.clone(); + let server_requests = requests.clone(); + let server_cancelled = cancelled_requests.clone(); + let server = tokio::spawn(async move { + loop { + let (mut stream, _) = listener.accept().await.expect("accept discovery request"); + let mut request = Vec::new(); + while !request.ends_with(b"\r\n\r\n") { + request.push(stream.read_u8().await.expect("read discovery request")); + assert!(request.len() < 8192); + } + let headers = String::from_utf8(request).expect("HTTP request headers"); + let path = headers + .lines() + .next() + .expect("request line") + .split_whitespace() + .nth(1) + .expect("path"); + let (status, body) = if path.ends_with("/.well-known/openid-configuration") { + (server_discovery_status.load(Ordering::SeqCst), document.as_str()) + } else { + assert_eq!(path, "/jwks"); + (server_jwks_status.load(Ordering::SeqCst), jwks.as_str()) + }; + server_requests.fetch_add(1, Ordering::SeqCst); + if status == 0 { + continue; // Simulate a transport failure before a response is available. + } + if status == 1 { + let mut byte = [0]; + assert_eq!(stream.read(&mut byte).await.expect("client closes cancelled discovery"), 0); + server_cancelled.fetch_add(1, Ordering::SeqCst); + continue; + } + let body = if status == 200 { body } else { "" }; + let response = format!( + "HTTP/1.1 {status} Test\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ); + stream.write_all(response.as_bytes()).await.expect("write discovery response"); + } + }); + Self { + base, + config, + key, + discovery_status, + jwks_status, + requests, + cancelled_requests, + _server: AbortOnDropHandle::new(server), + } + } + + fn token(&self, key: &EncodingKey, audience: &str) -> String { + let mut header = Header::new(Algorithm::ES256); + header.kid = Some(self.config.id.clone()); + let now = time::OffsetDateTime::now_utc().unix_timestamp(); + jsonwebtoken::encode( + &header, + &serde_json::json!({ + "iss": self.config.issuer.as_deref().unwrap_or(&self.base), + "sub": "recovered-user", "aud": audience, "iat": now, "exp": now + 300, + "groups": ["readonly"], + }), + key, + ) + .expect("sign recovery test token") + } +} + +async fn initialize(idps: &[&RecoveryIdp]) -> Arc { + let origins = idps.iter().map(|idp| idp.base.as_str()).collect::>().join(","); + let client = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&origins).expect("allow test IdPs")); + let snapshot = OidcConfigSnapshot::new(idps.iter().map(|idp| test_sourced_config(idp.config.clone())).collect()); + Arc::new( + OidcSys::new_with_config_snapshot(client, snapshot) + .await + .expect("initialize OIDC with unavailable IdPs"), + ) +} + +async fn wait_until(mut condition: impl FnMut() -> bool) { + tokio::time::timeout(Duration::from_secs(5), async { + while !condition() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("recovery condition should be reached"); +} + +#[tokio::test] +async fn web_identity_recovers_transport_failure_and_preserves_verification() { + for explicit in [false, true] { + let idp = RecoveryIdp::start("workload", true, explicit).await; + idp.discovery_status.store(0, Ordering::SeqCst); + let sys = initialize(&[&idp]).await; + let adapter = StandardOidcAdapter::new(sys.clone()); + let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules()); + assert!(sys.has_providers()); + assert!(sys.list_visible_providers().is_empty()); + let token = idp.token(&idp.key, &idp.config.client_id); + assert!(adapter.verify_identity(&token).await.is_err()); + + idp.discovery_status.store(200, Ordering::SeqCst); + tokio::time::timeout(Duration::from_secs(5), sys.recover_failed_providers()) + .await + .expect("recover workload discovery"); + let identity = adapter + .verify_identity(&token) + .await + .expect("verify recovered workload token"); + let authorization = mapper.map(identity); + assert_eq!(authorization.provider_id, "workload"); + assert_eq!(authorization.claims.sub, "recovered-user"); + assert_eq!(authorization.policies, vec!["team/readonly"]); + assert!( + sys.authorize_url("workload", "https://console.example.com/callback", None) + .await + .is_err() + ); + + let (wrong_key, _) = oidc_es256_key_and_jwk("wrong"); + assert!( + adapter + .verify_identity(&idp.token(&wrong_key, &idp.config.client_id)) + .await + .is_err() + ); + assert!(adapter.verify_identity(&idp.token(&idp.key, "wrong-audience")).await.is_err()); + } +} + +#[tokio::test] +async fn recovery_waits_for_jwks_and_claims_pending_providers_once() { + let idp = RecoveryIdp::start("console", false, true).await; + idp.discovery_status.store(200, Ordering::SeqCst); + idp.jwks_status.store(503, Ordering::SeqCst); + let sys = initialize(&[&idp]).await; + assert!(sys.provider_runtime.get_provider_state("console").is_err()); + let requests = idp.requests.load(Ordering::SeqCst); + idp.jwks_status.store(200, Ordering::SeqCst); + tokio::time::timeout( + Duration::from_secs(5), + futures::future::join_all((0..16).map(|_| sys.recover_failed_providers())), + ) + .await + .expect("concurrent recovery calls complete"); + assert!(sys.provider_runtime.get_provider_state("console").is_ok()); + assert_eq!( + idp.requests.load(Ordering::SeqCst), + requests + 2, + "only one discovery/JWKS round should run" + ); +} + +#[tokio::test] +async fn recovery_backoff_is_independent_of_login_traffic() { + let idp = RecoveryIdp::start("console", false, true).await; + let sys = initialize(&[&idp]).await; + let requests = idp.requests.load(Ordering::SeqCst); + let recovery_sys = sys.clone(); + let recovery = AbortOnDropHandle::new(tokio::spawn(async move { recovery_sys.recover_failed_providers().await })); + wait_until(|| idp.requests.load(Ordering::SeqCst) > requests).await; + assert_eq!(idp.requests.load(Ordering::SeqCst), requests + 1); + for _ in 0..32 { + assert!( + sys.authorize_url("console", "https://console.example.com/callback", None) + .await + .is_err() + ); + } + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!( + idp.requests.load(Ordering::SeqCst), + requests + 1, + "login attempts must not bypass discovery backoff" + ); + idp.discovery_status.store(200, Ordering::SeqCst); + tokio::time::timeout(Duration::from_secs(5), recovery) + .await + .expect("recovery deadline") + .expect("recovery task"); + assert!( + sys.authorize_url("console", "https://console.example.com/callback", None) + .await + .is_ok() + ); +} + +#[tokio::test] +async fn unavailable_provider_does_not_block_another_provider() { + let unavailable = RecoveryIdp::start("unavailable", false, true).await; + let available = RecoveryIdp::start("available", false, true).await; + let sys = initialize(&[&unavailable, &available]).await; + available.discovery_status.store(200, Ordering::SeqCst); + let recovery_sys = sys.clone(); + let recovery = AbortOnDropHandle::new(tokio::spawn(async move { recovery_sys.recover_failed_providers().await })); + wait_until(|| sys.provider_runtime.get_provider_state("available").is_ok()).await; + assert!(sys.provider_runtime.get_provider_state("unavailable").is_err()); + assert!( + sys.authorize_url("available", "https://console.example.com/callback", None) + .await + .is_ok() + ); + drop(recovery); +} + +#[tokio::test] +async fn recovery_preserves_ready_provider_and_inflight_sessions() { + let ready = RecoveryIdp::start("ready", false, true).await; + let recovering = RecoveryIdp::start("recovering", false, true).await; + ready.discovery_status.store(200, Ordering::SeqCst); + let sys = initialize(&[&ready, &recovering]).await; + let auth_url = sys + .authorize_url("ready", "https://console.example.com/callback", Some("/buckets".into())) + .await + .expect("begin login with ready provider"); + let auth: HashMap = url::Url::parse(&auth_url) + .expect("authorization URL") + .query_pairs() + .into_owned() + .collect(); + let logout = sys + .create_logout_token("ready", "verified-id-token") + .await + .expect("retain logout continuation"); + let discovered_at = sys + .provider_runtime + .get_provider_state("ready") + .expect("ready state") + .discovered_at; + let ready_requests = ready.requests.load(Ordering::SeqCst); + + recovering.discovery_status.store(200, Ordering::SeqCst); + tokio::time::timeout(Duration::from_secs(5), sys.recover_failed_providers()) + .await + .expect("recover second provider"); + assert_eq!(ready.requests.load(Ordering::SeqCst), ready_requests); + assert_eq!( + sys.provider_runtime + .get_provider_state("ready") + .expect("ready state remains") + .discovered_at, + discovered_at + ); + let session = sys + .state_store + .take(&auth["state"]) + .await + .expect("in-flight login remains available"); + assert_eq!(session.provider_id, "ready"); + assert_eq!(session.nonce, auth["nonce"]); + assert_eq!(session.redirect_after.as_deref(), Some("/buckets")); + assert_eq!( + PkceCodeChallenge::from_code_verifier_sha256(&PkceCodeVerifier::new(session.pkce_verifier)).as_str(), + auth["code_challenge"] + ); + assert_eq!( + sys.state_store + .take_logout(&logout) + .await + .expect("logout continuation remains available") + .provider_id, + "ready" + ); +} + +#[tokio::test] +async fn startup_schedules_only_recoverable_http_errors() { + for explicit in [false, true] { + let idp = RecoveryIdp::start("status", false, explicit).await; + for (status, retryable) in [ + (408, true), + (429, true), + (500, true), + (503, true), + (400, false), + (401, false), + (403, false), + (404, false), + ] { + idp.discovery_status.store(status, Ordering::SeqCst); + let sys = initialize(&[&idp]).await; + assert_eq!( + sys.pending_discovery.lock().await.len(), + usize::from(retryable), + "explicit={explicit}, status={status}" + ); + assert!(sys.provider_runtime.get_provider_state("status").is_err()); + } + } +} + +#[tokio::test] +async fn recovery_bounds_concurrency_and_cancels_inflight_requests() { + let mut idps = Vec::new(); + for index in 0..6 { + idps.push(RecoveryIdp::start(&format!("provider-{index}"), false, true).await); + } + let refs: Vec<_> = idps.iter().collect(); + let sys = initialize(&refs).await; + let initial_requests: usize = idps.iter().map(|idp| idp.requests.load(Ordering::SeqCst)).sum(); + for idp in &idps { + idp.discovery_status.store(1, Ordering::SeqCst); + } + let recovery_sys = sys.clone(); + let recovery = AbortOnDropHandle::new(tokio::spawn(async move { recovery_sys.recover_failed_providers().await })); + let requests = || idps.iter().map(|idp| idp.requests.load(Ordering::SeqCst)).sum::(); + wait_until(|| requests() >= initial_requests + 4).await; + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!(requests(), initial_requests + 4, "at most four discovery rounds may run concurrently"); + drop(recovery); + wait_until(|| { + idps.iter() + .map(|idp| idp.cancelled_requests.load(Ordering::SeqCst)) + .sum::() + == 4 + }) + .await; + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!(requests(), initial_requests + 4, "cancelled recovery must not start queued work"); + assert!( + idps.iter() + .all(|idp| sys.provider_runtime.get_provider_state(&idp.config.id).is_err()) + ); +} + +#[tokio::test] +async fn disabled_and_rejected_providers_do_not_schedule_recovery() { + let mut disabled = RecoveryIdp::start("disabled", false, true).await; + disabled.config.enabled = false; + let sys = initialize(&[&disabled]).await; + sys.recover_failed_providers().await; + assert!(!sys.has_providers()); + assert_eq!(disabled.requests.load(Ordering::SeqCst), 0); + + let blocked = RecoveryIdp::start("blocked", false, true).await; + let sys = OidcSys::new_with_config_snapshot( + ReqwestHttpClient::with_policy(OutboundPolicy::default()), + OidcConfigSnapshot::new(vec![test_sourced_config(blocked.config.clone())]), + ) + .await + .expect("policy rejection remains non-fatal for startup"); + assert!(sys.pending_discovery.lock().await.is_empty()); + sys.recover_failed_providers().await; + assert_eq!(blocked.requests.load(Ordering::SeqCst), 0); + assert!(sys.provider_runtime.get_provider_state("blocked").is_err()); +} + +#[tokio::test] +async fn recovery_stops_on_issuer_validation_failure() { + let mut idp = RecoveryIdp::start("console", false, true).await; + idp.config.issuer = Some("https://unexpected.example.com".into()); + let sys = initialize(&[&idp]).await; + idp.discovery_status.store(200, Ordering::SeqCst); + tokio::time::timeout(Duration::from_secs(5), sys.recover_failed_providers()) + .await + .expect("validation failure stops recovery"); + assert_eq!( + idp.requests.load(Ordering::SeqCst), + 2, + "issuer rejection must not fetch JWKS or schedule another attempt" + ); + assert!(sys.provider_runtime.get_provider_state("console").is_err()); + assert!( + sys.authorize_url("console", "https://console.example.com/callback", None) + .await + .is_err() + ); +} + +#[tokio::test] +async fn recovery_logs_failure_reasons_without_repeating_warnings() { + use std::io::{Read, Seek}; + + for (explicit, status, expected_kind) in [ + (false, 503, "http_status"), + (true, 503, "http_status"), + (true, 0, "request"), + (true, 1, "timeout"), + ] { + let idp = RecoveryIdp::start("logging", false, explicit).await; + let sys = initialize(&[&idp]).await; + let requests = idp.requests.load(Ordering::SeqCst); + idp.discovery_status.store(status, Ordering::SeqCst); + let mut output = tempfile::tempfile().expect("create captured log file"); + let subscriber = tracing_subscriber::fmt() + .without_time() + .with_ansi(false) + .with_max_level(tracing::Level::WARN) + .with_writer(output.try_clone().expect("clone captured log file")) + .finish(); + let _subscriber = tracing::subscriber::set_default(subscriber); + tracing::callsite::rebuild_interest_cache(); + tokio::time::timeout(Duration::from_secs(25), async { + tokio::join!(sys.recover_failed_providers(), async { + // Two failing rounds exercise suppression; a timeout is already bounded by the HTTP client. + while if status == 1 { + idp.cancelled_requests.load(Ordering::SeqCst) == 0 + } else { + idp.requests.load(Ordering::SeqCst) < requests + if explicit { 2 } else { 4 } + } { + tokio::time::sleep(Duration::from_millis(10)).await; + } + idp.discovery_status.store(200, Ordering::SeqCst); + }); + }) + .await + .expect("recover after recording failure diagnostics"); + output.rewind().expect("rewind captured log file"); + let mut logs = String::new(); + output.read_to_string(&mut logs).expect("read captured log file"); + assert_eq!(logs.lines().count(), 1, "explicit={explicit}, status={status}: {logs}"); + assert!(logs.contains("provider_discovery_recovery_pending"), "{logs}"); + assert!(logs.contains(&format!("error_kind=\"{expected_kind}\"")), "{logs}"); + if status == 503 { + assert!(logs.contains("error_detail=503"), "{logs}"); + } + assert!(logs.contains("provider_id=logging"), "{logs}"); + assert!(logs.contains("suppressed=0"), "{logs}"); + } +} diff --git a/crates/iam/src/oidc/runtime.rs b/crates/iam/src/oidc/runtime.rs index 3a5797fc1..7088d31d9 100644 --- a/crates/iam/src/oidc/runtime.rs +++ b/crates/iam/src/oidc/runtime.rs @@ -16,7 +16,7 @@ use super::{ OidcConfigQuery, OidcConfigSnapshot, OidcProviderConfig, OidcProviderSummary, OidcSiteReplicationSnapshot, SourcedOidcProviderConfig, load_oidc_config_snapshot, }, - provider::{ProviderRuntime, discover_provider}, + provider::{ProviderRuntime, discover_provider_for_recovery}, state::{OidcAuthSession, OidcLogoutSession, OidcStateStore}, transport::{OidcExtraRootCaProvider, ReqwestHttpClient, inspect_token_response_body, oidc_http_error_diagnostics}, }; @@ -37,11 +37,19 @@ use openidconnect::{ PkceCodeVerifier, PostLogoutRedirectUrl, RedirectUrl, RequestTokenError, Scope, }; use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive}; +use rustfs_utils::logging::LogThrottle; use serde::{Deserialize, Serialize}; use std::borrow::Cow; use std::collections::HashMap; use std::sync::Arc; -use tracing::{debug, error, warn}; +use std::time::Duration; +use tokio::sync::{Mutex, Semaphore}; +use tracing::{debug, error, info, warn}; + +const OIDC_DISCOVERY_RECOVERY_CONCURRENCY: usize = 4; +const OIDC_DISCOVERY_RECOVERY_MAX_DELAY_MS: u64 = 30_000; +// A round can try two issuer variants three times, each fetching discovery and JWKS. +const OIDC_DISCOVERY_RECOVERY_TIMEOUT: Duration = Duration::from_secs(125); /// Claims extracted from an OIDC ID token. #[derive(Debug, Clone, Serialize, Deserialize, Default)] @@ -60,6 +68,7 @@ pub struct OidcSys { pub(super) configs: HashMap, pub(super) provider_runtime: ProviderRuntime, pub(super) state_store: OidcStateStore, + pub(super) pending_discovery: Mutex>, } pub(super) fn trusted_aud(other_audiences: &[String], audience: &Audience) -> bool { @@ -81,8 +90,16 @@ impl OidcSys { async fn new_with_http_client(http_client: ReqwestHttpClient) -> Result { let server_config = crate::server_config::current_server_config(); let parsed_configs = load_oidc_config_snapshot(server_config.as_ref()); + Self::new_with_config_snapshot(http_client, parsed_configs).await + } + + async fn new_with_config_snapshot( + http_client: ReqwestHttpClient, + parsed_configs: OidcConfigSnapshot, + ) -> Result { let mut configs = HashMap::new(); let mut provider_states = HashMap::new(); + let mut pending_discovery = Vec::new(); for sourced_config in parsed_configs.into_providers() { let config = &sourced_config.config; @@ -91,13 +108,15 @@ impl OidcSys { continue; } - match discover_provider(config, &http_client).await { + match discover_provider_for_recovery(config, &http_client).await { Ok(state) => { debug!(provider = %config.id, "OIDC provider discovered"); provider_states.insert(config.id.clone(), state); - configs.insert(config.id.clone(), sourced_config); } Err(e) => { + if e.retryable { + pending_discovery.push(config.id.clone()); + } error!( event = EVENT_OIDC_DIAGNOSTICS, component = LOG_COMPONENT_IAM, @@ -114,21 +133,95 @@ impl OidcSys { ); } } + // Configuration also supplies redirect and authorization rules before recovery. + configs.insert(config.id.clone(), sourced_config); } Ok(Self { configs, provider_runtime: ProviderRuntime::new(http_client, provider_states), state_store: OidcStateStore::new(), + pending_discovery: Mutex::new(pending_discovery), }) } + /// Recover providers that failed startup discovery due to a transient error. + /// The server owns and cancels this future; dropping it cancels all outstanding requests. + /// Pending providers are claimed once so concurrent callers cannot duplicate recovery. + pub async fn recover_failed_providers(&self) { + let pending = std::mem::take(&mut *self.pending_discovery.lock().await); + let permits = Semaphore::new(OIDC_DISCOVERY_RECOVERY_CONCURRENCY); + futures::future::join_all(pending.iter().filter_map(|id| self.get_provider_config(id)).map(|config| { + let permits = &permits; + async move { + let mut delay_ms = 1_000_u64; + let failure_log = LogThrottle::new(60_000); + loop { + let sleep_ms = rand::random_range(delay_ms * 3 / 4..=delay_ms); + tokio::time::sleep(Duration::from_millis(sleep_ms)).await; + let Ok(permit) = permits.acquire().await else { + return; + }; + let result = tokio::time::timeout( + OIDC_DISCOVERY_RECOVERY_TIMEOUT, + self.provider_runtime.recover_provider_state(&config.id, config), + ) + .await; + drop(permit); + let (error_kind, error_detail) = match result { + Ok(Ok(_)) => { + info!( + event = EVENT_OIDC_DIAGNOSTICS, + component = LOG_COMPONENT_IAM, + subsystem = LOG_SUBSYSTEM_OIDC, + result = "provider_discovery_recovered", + provider_id = %config.id, + "oidc provider discovery recovered" + ); + return; + } + Ok(Err(error)) if !error.retryable => { + error!( + event = EVENT_OIDC_DIAGNOSTICS, + component = LOG_COMPONENT_IAM, + subsystem = LOG_SUBSYSTEM_OIDC, + result = "provider_discovery_recovery_stopped", + provider_id = %config.id, + error = %error, + "oidc provider discovery requires configuration correction" + ); + return; + } + Ok(Err(error)) => error.diagnostics, + Err(_) => ("attempt_timeout", OIDC_DISCOVERY_RECOVERY_TIMEOUT.as_secs().to_string()), + }; + if let Some(suppressed) = failure_log.claim() { + warn!( + event = EVENT_OIDC_DIAGNOSTICS, + component = LOG_COMPONENT_IAM, + subsystem = LOG_SUBSYSTEM_OIDC, + result = "provider_discovery_recovery_pending", + provider_id = %config.id, + error_kind, + error_detail = %error_detail, + suppressed, + "oidc provider discovery remains unavailable" + ); + } + delay_ms = delay_ms.saturating_mul(2).min(OIDC_DISCOVERY_RECOVERY_MAX_DELAY_MS); + } + } + })) + .await; + } + /// Create an OidcSys with no providers (useful for when OIDC is not configured). pub fn empty() -> Result { Ok(Self { configs: HashMap::new(), provider_runtime: ProviderRuntime::new(ReqwestHttpClient::new()?, HashMap::new()), state_store: OidcStateStore::new(), + pending_discovery: Mutex::default(), }) } @@ -895,3 +988,7 @@ impl FederatedProviderQuery for StandardOidcAdapter { #[cfg(test)] #[path = "runtime_tests.rs"] mod tests; + +#[cfg(test)] +#[path = "recovery_tests.rs"] +mod recovery_tests; diff --git a/crates/iam/src/oidc/runtime_tests.rs b/crates/iam/src/oidc/runtime_tests.rs index afca6f4c3..84dd95943 100644 --- a/crates/iam/src/oidc/runtime_tests.rs +++ b/crates/iam/src/oidc/runtime_tests.rs @@ -175,6 +175,7 @@ async fn web_identity_verification_refreshes_rotated_jwks() { configs: HashMap::from([(config.id.clone(), test_sourced_config(config.clone()))]), provider_runtime: ProviderRuntime::new(http_client, HashMap::from([(config.id.clone(), state)])), state_store: OidcStateStore::new(), + pending_discovery: Default::default(), }; let now = std::time::SystemTime::now() @@ -208,6 +209,15 @@ async fn web_identity_verification_refreshes_rotated_jwks() { } #[tokio::test] async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() { + assert_complete_console_login(false).await; +} + +#[tokio::test] +async fn console_login_recovers_startup_discovery_with_authorization_rules() { + assert_complete_console_login(true).await; +} + +async fn assert_complete_console_login(initially_unavailable: bool) { use tokio::io::{AsyncReadExt, AsyncWriteExt}; for hidden in [false, true] { @@ -221,6 +231,8 @@ async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() { }; let mut config = build_mocked_oidc_provider_config("console", &format!("{base}/.well-known/openid-configuration")); config.hide_from_ui = hidden; + config.claim_name = "groups".into(); + config.claim_prefix = "team/".into(); config.issuer = explicit.then(|| issuer.clone()); config.client_secret = Some(Nonce::new_random().secret().clone()); let server_config = config.clone(); @@ -230,6 +242,23 @@ async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() { let (auth_tx, auth_rx) = tokio::sync::oneshot::channel::>(); let server = tokio::spawn(async move { let mut auth_rx = Some(auth_rx); + let failures = if initially_unavailable { + if explicit { 1 } else { 2 } + } else { + 0 + }; + for _ in 0..failures { + let (mut stream, _) = listener.accept().await.expect("startup discovery request"); + let mut request = Vec::new(); + while !request.ends_with(b"\r\n\r\n") { + request.push(stream.read_u8().await.expect("discovery request headers")); + assert!(request.len() < 8192); + } + stream + .write_all(b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") + .await + .expect("startup discovery failure response"); + } for expected_path in ["/.well-known/openid-configuration", "/jwks", "/token"] { let (mut stream, _) = tokio::time::timeout(StdDuration::from_secs(10), listener.accept()) .await @@ -309,12 +338,21 @@ async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() { } }); let http_client = ReqwestHttpClient::with_policy(OutboundPolicy::from_allowed_origins(&base).unwrap()); - let discovered = discover_provider(&config, &http_client).await.unwrap(); - let sys = OidcSys { - configs: HashMap::from([(config.id.clone(), test_sourced_config(config))]), - provider_runtime: ProviderRuntime::new(http_client, HashMap::from([("console".into(), discovered)])), - state_store: OidcStateStore::new(), - }; + let sys = Arc::new( + OidcSys::new_with_config_snapshot(http_client, OidcConfigSnapshot::new(vec![test_sourced_config(config)])) + .await + .expect("OIDC initialization remains available during an IdP outage"), + ); + let adapter = StandardOidcAdapter::new(sys.clone()); + let mapper = CoreFederatedAuthorizationMapper::new(adapter.authorization_rules()); + assert_eq!(sys.list_visible_providers().len(), usize::from(!hidden)); + if initially_unavailable { + assert!(sys.provider_runtime.get_provider_state("console").is_err()); + assert!(sys.authorize_url("console", redirect, None).await.is_err()); + tokio::time::timeout(StdDuration::from_secs(10), sys.recover_failed_providers()) + .await + .expect("discovery should recover after the IdP becomes available"); + } 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())); @@ -328,15 +366,18 @@ async fn complete_provider_console_login_preserves_hidden_and_issuer_modes() { 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) + let exchange = adapter + .exchange_identity(&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")); + let (identity, redirect_after, _) = exchange.into_parts(); + let authorization = mapper.map(identity); + assert_eq!(authorization.provider_id, "console"); + assert_eq!(authorization.claims.sub, "existing-user"); + assert_eq!(authorization.claims.email, "user@example.com"); + assert_eq!(authorization.claims.groups, vec!["readwrite"]); + assert_eq!(authorization.policies, vec!["team/readwrite"]); + assert_eq!(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(); @@ -470,6 +511,7 @@ async fn oidc_token_exchange_reports_forbidden_token_endpoint() { )]), ), state_store, + pending_discovery: Default::default(), }; let error = match sys diff --git a/crates/iam/src/oidc/test_support.rs b/crates/iam/src/oidc/test_support.rs index e81855537..51262f675 100644 --- a/crates/iam/src/oidc/test_support.rs +++ b/crates/iam/src/oidc/test_support.rs @@ -35,6 +35,7 @@ pub fn make_test_sys(configs: Vec) -> OidcSys { HashMap::new(), ), state_store: OidcStateStore::new(), + pending_discovery: Default::default(), } } diff --git a/crates/iam/src/oidc/transport.rs b/crates/iam/src/oidc/transport.rs index 1fd8706f0..18a708a4d 100644 --- a/crates/iam/src/oidc/transport.rs +++ b/crates/iam/src/oidc/transport.rs @@ -312,6 +312,10 @@ struct CachedOidcExtraRootCerts { certs: Vec, } +/// Discovery callers report failures with provider context and recovery throttling. +#[derive(Clone)] +pub(super) struct DiscoveryRequest; + /// HTTP client adapter bridging reqwest 0.13 to the `openidconnect` `AsyncHttpClient` trait. /// /// A fresh client is built for every request so the destination is re-validated and the @@ -555,6 +559,7 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient { Box::pin(async move { let started_at = Instant::now(); let (parts, body) = request.into_parts(); + let discovery_request = parts.extensions.get::().is_some(); let method = parts.method.clone(); let uri = parts.uri.to_string(); if tracing::enabled!(tracing::Level::DEBUG) { @@ -593,17 +598,19 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient { let response = response.map_err(|err| { let error = oidc_http_error_from_reqwest(&url, err); - error!( - event = EVENT_OIDC_HTTP, - component = LOG_COMPONENT_IAM, - subsystem = LOG_SUBSYSTEM_OIDC, - result = "request_failed", - method = %method, - uri = %uri, - elapsed_ms, - error = %error, - "oidc outbound http" - ); + if !discovery_request { + error!( + event = EVENT_OIDC_HTTP, + component = LOG_COMPONENT_IAM, + subsystem = LOG_SUBSYSTEM_OIDC, + result = "request_failed", + method = %method, + uri = %uri, + elapsed_ms, + error = %error, + "oidc outbound http" + ); + } error })?; @@ -612,18 +619,20 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient { let body_bytes = read_bounded_response_body(response, MAX_OIDC_RESPONSE_SIZE) .await .map_err(|err| { - error!( - event = EVENT_OIDC_HTTP, - component = LOG_COMPONENT_IAM, - subsystem = LOG_SUBSYSTEM_OIDC, - result = "response_body_failed", - method = %method, - uri = %uri, - status = status.as_u16(), - elapsed_ms, - error = %err, - "oidc outbound http" - ); + if !discovery_request { + error!( + event = EVENT_OIDC_HTTP, + component = LOG_COMPONENT_IAM, + subsystem = LOG_SUBSYSTEM_OIDC, + result = "response_body_failed", + method = %method, + uri = %uri, + status = status.as_u16(), + elapsed_ms, + error = %err, + "oidc outbound http" + ); + } err })?; if tracing::enabled!(tracing::Level::DEBUG) { diff --git a/rustfs/src/startup_auth.rs b/rustfs/src/startup_auth.rs index ec2cbe6f4..c704a9fbf 100644 --- a/rustfs/src/startup_auth.rs +++ b/rustfs/src/startup_auth.rs @@ -19,10 +19,12 @@ use rustfs_iam::{ }; use std::{ collections::hash_map::DefaultHasher, + future::Future, hash::{Hash, Hasher}, io::{Error, Result}, sync::Arc, }; +use tokio_util::{sync::CancellationToken, task::AbortOnDropHandle}; use tracing::{error, info, warn}; const LOG_COMPONENT_MAIN: &str = "main"; @@ -31,7 +33,7 @@ const EVENT_KEYSTONE_AUTH_INITIALIZED: &str = "keystone_auth_initialized"; const EVENT_KEYSTONE_AUTH_INITIALIZATION_FAILED: &str = "keystone_auth_initialization_failed"; const EVENT_OIDC_INITIALIZATION_FAILED: &str = "oidc_initialization_failed"; -pub(crate) async fn init_auth_integrations() -> Result<()> { +pub(crate) async fn init_auth_integrations(shutdown: CancellationToken) -> Result>> { let keystone_config = rustfs_keystone::KeystoneConfig::from_env().map_err(Error::other)?; if keystone_config.enable { match crate::auth_keystone::init_keystone_auth(keystone_config).await { @@ -55,8 +57,13 @@ pub(crate) async fn init_auth_integrations() -> Result<()> { match build_oidc_sys_with_extra_root_ca_provider(oidc_extra_root_ca_provider()).await { Ok(oidc) => { - let (service, oidc_config_query) = standard_oidc_runtime(oidc); - crate::runtime_sources::publish_federated_identity_runtime(service, oidc_config_query); + let (service, oidc_config_query) = standard_oidc_runtime(oidc.clone()); + if crate::runtime_sources::publish_federated_identity_runtime(service, oidc_config_query) { + return Ok(Some(spawn_oidc_discovery_recovery( + async move { oidc.recover_failed_providers().await }, + shutdown, + ))); + } } Err(e) => { warn!( @@ -69,7 +76,20 @@ pub(crate) async fn init_auth_integrations() -> Result<()> { } } - Ok(()) + Ok(None) +} + +fn spawn_oidc_discovery_recovery( + recovery: impl Future + Send + 'static, + shutdown: CancellationToken, +) -> AbortOnDropHandle<()> { + AbortOnDropHandle::new(tokio::spawn(async move { + tokio::select! { + biased; + () = shutdown.cancelled() => {} + () = recovery => {} + } + })) } fn standard_oidc_runtime(oidc: Arc) -> (Arc, Arc) { @@ -133,4 +153,48 @@ mod tests { drop(oidc_config_query); assert_eq!(Arc::strong_count(&retained), 1, "dropping the runtime should release its OIDC Arc"); } + + #[tokio::test] + async fn oidc_recovery_shutdown_releases_pending_work() { + let shutdown = CancellationToken::new(); + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (released_tx, released_rx) = tokio::sync::oneshot::channel::<()>(); + let recovery = spawn_oidc_discovery_recovery( + async move { + let _release = released_tx; + let _ = started_tx.send(()); + std::future::pending::<()>().await; + }, + shutdown.child_token(), + ); + started_rx.await.expect("recovery started"); + shutdown.cancel(); + tokio::time::timeout(std::time::Duration::from_secs(5), recovery) + .await + .expect("recovery observes shutdown") + .expect("recovery joins cleanly"); + assert!(released_rx.await.is_err(), "shutdown drops in-flight recovery work"); + } + + #[tokio::test] + async fn oidc_recovery_owner_drop_releases_pending_work() { + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (released_tx, released_rx) = tokio::sync::oneshot::channel::<()>(); + let recovery = spawn_oidc_discovery_recovery( + async move { + let _release = released_tx; + let _ = started_tx.send(()); + std::future::pending::<()>().await; + }, + CancellationToken::new(), + ); + started_rx.await.expect("recovery started"); + drop(recovery); + assert!( + tokio::time::timeout(std::time::Duration::from_secs(5), released_rx) + .await + .expect("owner drop aborts recovery") + .is_err() + ); + } } diff --git a/rustfs/src/startup_lifecycle.rs b/rustfs/src/startup_lifecycle.rs index ed17225ed..d3360935e 100644 --- a/rustfs/src/startup_lifecycle.rs +++ b/rustfs/src/startup_lifecycle.rs @@ -130,6 +130,7 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec readiness, } = lifecycle; let StartupServiceRuntime { + oidc_recovery, optional_runtimes, heartbeat, inventory, @@ -166,6 +167,18 @@ pub(crate) async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifec shutdown_token, ) .await; + if let Some(recovery) = oidc_recovery + && let Err(error) = recovery.await + { + tracing::warn!( + event = EVENT_SERVER_SHUTDOWN_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + result = "oidc_discovery_recovery_join_failed", + error = %error, + "OIDC discovery recovery task failed to join" + ); + } shutdown_connect_runtimes(heartbeat, inventory).await; if let Some(runtime) = local_trace_capture { runtime.shutdown().await; diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index 5dd0fe9a7..4a3b66a5e 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -36,13 +36,14 @@ use crate::{ }; use rustfs_common::GlobalReadiness; use std::{collections::BTreeSet, io::Result, sync::Arc}; -use tokio_util::sync::CancellationToken; +use tokio_util::{sync::CancellationToken, task::AbortOnDropHandle}; const EVENT_DRIVE_UNAVAILABLE: &str = "drive_unavailable"; const LOG_COMPONENT_CONNECT: &str = "connect"; const LOG_SUBSYSTEM_INVENTORY: &str = "inventory"; pub(crate) struct StartupServiceRuntime { + pub(crate) oidc_recovery: Option>, pub(crate) optional_runtimes: OptionalRuntimeServices, pub(crate) heartbeat: Option, pub(crate) inventory: Option, @@ -111,7 +112,7 @@ pub(crate) async fn init_startup_runtime_services( // self-pointing replication rules until the next restart. The task waits for IAM and // bucket metadata itself, and its first pass runs off this path. spawn_site_replication_reconcile_task(ctx.clone()); - init_auth_integrations().await?; + let oidc_recovery = init_auth_integrations(ctx.child_token()).await?; init_notification_runtime(endpoint_pools, buckets).await?; let enable_scanner = init_background_service_runtime(store.clone()).await?; init_observability_runtime(store.clone(), ctx.clone()).await; @@ -119,6 +120,7 @@ pub(crate) async fn init_startup_runtime_services( let inventory = start_inventory_runtime(heartbeat_config, heartbeat_nodes, inventory_drives, store, &ctx)?; Ok(StartupServiceRuntime { + oidc_recovery, optional_runtimes, heartbeat, inventory,