From ad5f3ba9974c26543d058fed4f0732215657fa58 Mon Sep 17 00:00:00 2001 From: zyycn Date: Mon, 27 Jul 2026 01:51:41 +0800 Subject: [PATCH] fix(xai): tolerate unknown JWKS members and corrupt catalog cache; drop dead sync Stop rejecting a JWKS document that carries members the struct does not model, so a token still verifies when the IdP adds a field. Treat an unparseable catalog cache entry as a cache miss that refetches instead of a hard failure. Remove the unused GrokCredentialCatalogService::synchronize cluster and its snapshot re-export. Co-authored-by: Claude --- .../providers/xai/src/credential/catalog.rs | 98 +++++-------------- .../providers/xai/src/credential/mod.rs | 4 +- .../xai/src/credential/oidc_verifier.rs | 3 +- backend/crates/providers/xai/src/lib.rs | 22 ++--- .../providers/xai/tests/credential/catalog.rs | 92 ++++++++++++----- .../xai/tests/credential/oidc_verifier.rs | 28 ++++++ 6 files changed, 137 insertions(+), 110 deletions(-) diff --git a/backend/crates/providers/xai/src/credential/catalog.rs b/backend/crates/providers/xai/src/credential/catalog.rs index c81ddc9f8..e0aff4323 100644 --- a/backend/crates/providers/xai/src/credential/catalog.rs +++ b/backend/crates/providers/xai/src/credential/catalog.rs @@ -15,7 +15,7 @@ use gateway_core::engine::provider::ProviderCatalogGeneration; use gateway_core::provider_ports::{ ProviderCatalogCacheKey, ProviderCatalogCachePort, ProviderCatalogScope, }; -use gateway_core::routing::{ConfigRevision, ProviderKind}; +use gateway_core::routing::ProviderKind; use tokio::sync::Mutex; use super::repository::{GrokCredentialRepository, LoadedGrokCredential}; @@ -796,12 +796,25 @@ impl GrokCredentialCatalogCache for GrokCatalogCache { &self, scope: &GrokCatalogScope, ) -> Result, GrokCatalogCacheError> { - self.port + let Some(document) = self + .port .read(&self.key(scope)) .await .map_err(|_| GrokCatalogCacheError::Unavailable)? - .map(|document| Self::decode(scope, document)) - .transpose() + else { + return Ok(None); + }; + // cache 只保存可重建 TTL 数据：损坏条目按 miss 丢弃，由调用方实时重建。 + match Self::decode(scope, document) { + Ok(catalog) => Ok(Some(catalog)), + Err(_) => { + tracing::warn!( + scope = scope.as_str(), + "discarding corrupt xAI model catalog cache entry" + ); + Ok(None) + } + } } async fn observed_model_support( @@ -824,37 +837,6 @@ pub enum GrokCatalogCacheError { InvalidData, } -/// 一次 Provider 同步得到的套餐目录和严格模型并集。 -#[derive(Clone, Debug)] -pub struct GrokCredentialCatalogSnapshot { - config_revision: ConfigRevision, - observed_at: DateTime, - plans: Vec, - models: Vec, -} - -impl GrokCredentialCatalogSnapshot { - #[must_use] - pub const fn config_revision(&self) -> ConfigRevision { - self.config_revision - } - - #[must_use] - pub const fn observed_at(&self) -> DateTime { - self.observed_at - } - - #[must_use] - pub fn plans(&self) -> &[GrokPlanCatalog] { - &self.plans - } - - #[must_use] - pub fn models(&self) -> &[GrokCatalogModel] { - &self.models - } -} - #[derive(Debug, thiserror::Error)] pub enum GrokCredentialCatalogError { #[error("Grok model catalog has no eligible OAuth account")] @@ -1011,26 +993,12 @@ impl GrokCredentialCatalogService { .map_err(|_| GrokCredentialCatalogError::Cache) } - /// 实时拉取全部 eligible OAuth account；结果只写可重建 TTL cache，不写 PostgreSQL。 - pub async fn synchronize( - &self, - config_revision: ConfigRevision, - ) -> Result { - let catalog = self.fetch_and_cache().await?; - Ok(GrokCredentialCatalogSnapshot { - config_revision, - observed_at: catalog.observed_at, - plans: catalog.plans, - models: catalog.models, - }) - } - /// Provider Registry 构建 RuntimeSnapshot 时使用的实时能力目录。 pub async fn query_models(&self) -> Result, GrokCredentialCatalogError> { - Ok(self.fetch_and_cache().await?.models) + self.fetch_and_cache().await } - async fn fetch_and_cache(&self) -> Result { + async fn fetch_and_cache(&self) -> Result, GrokCredentialCatalogError> { let candidates = self .repository .list_loaded_for_provider() @@ -1047,27 +1015,19 @@ impl GrokCredentialCatalogService { fetched.sort_by(|left, right| left.scope.cmp(&right.scope)); let models = strict_model_union(&fetched)?; let observed_at = Utc::now(); - let plans = fetched - .into_iter() - .map(|fetched| GrokPlanCatalog { - scope: fetched.scope, - observed_at, - seed: fetched.seed, - }) - .collect::>(); - for plan in &plans { + for fetched in fetched { self.cache - .replace(plan.clone()) + .replace(GrokPlanCatalog { + scope: fetched.scope, + observed_at, + seed: fetched.seed, + }) .await .map_err(|_| GrokCredentialCatalogError::Cache)?; } self.publish_models(&models); - Ok(FetchedProviderCatalog { - observed_at, - plans, - models, - }) + Ok(models) } async fn refresh_scope_catalog( @@ -1136,12 +1096,6 @@ struct FetchedCredentialCatalog { seed: GrokCredentialCatalogSeed, } -struct FetchedProviderCatalog { - observed_at: DateTime, - plans: Vec, - models: Vec, -} - fn catalog_candidates_by_scope( candidates: Vec, ) -> Result>, GrokCredentialCatalogError> { diff --git a/backend/crates/providers/xai/src/credential/mod.rs b/backend/crates/providers/xai/src/credential/mod.rs index 07bc9e33d..f2344712e 100644 --- a/backend/crates/providers/xai/src/credential/mod.rs +++ b/backend/crates/providers/xai/src/credential/mod.rs @@ -29,8 +29,8 @@ pub use catalog::{ GROK_FREE_ROLLING_WINDOW_SECONDS, GrokBillingPresentation, GrokCatalogCache, GrokCatalogCacheError, GrokCatalogScope, GrokCredentialCatalogCache, GrokCredentialCatalogError, GrokCredentialCatalogSeed, GrokCredentialCatalogService, - GrokCredentialCatalogSnapshot, GrokCredentialQuotaService, GrokPlanCatalog, GrokQuotaError, - GrokQuotaPeriodKind, GrokQuotaSnapshot, + GrokCredentialQuotaService, GrokPlanCatalog, GrokQuotaError, GrokQuotaPeriodKind, + GrokQuotaSnapshot, }; pub use discovery::DiscoveryDocument; pub use error::{ diff --git a/backend/crates/providers/xai/src/credential/oidc_verifier.rs b/backend/crates/providers/xai/src/credential/oidc_verifier.rs index 044402993..60638b4f7 100644 --- a/backend/crates/providers/xai/src/credential/oidc_verifier.rs +++ b/backend/crates/providers/xai/src/credential/oidc_verifier.rs @@ -348,14 +348,13 @@ impl CachedJwk { } } +// RFC 7517 要求忽略 JWKS 与 JWK 的未知成员；字段本身的强校验在 parse_jwks。 #[derive(Deserialize)] -#[serde(deny_unknown_fields)] struct JwksWire { keys: Vec, } #[derive(Deserialize)] -#[serde(deny_unknown_fields)] struct JwkWire { kty: String, #[serde(rename = "use")] diff --git a/backend/crates/providers/xai/src/lib.rs b/backend/crates/providers/xai/src/lib.rs index 7f2635828..d3e5cb753 100644 --- a/backend/crates/providers/xai/src/lib.rs +++ b/backend/crates/providers/xai/src/lib.rs @@ -30,17 +30,17 @@ pub use credential::{ GrokAccountExport, GrokAccountProfile, GrokAccountSessionSelector, GrokBillingPresentation, GrokCatalogCache, GrokCatalogCacheError, GrokCatalogScope, GrokCredentialAdmin, GrokCredentialAvailability, GrokCredentialCatalogCache, GrokCredentialCatalogError, - GrokCredentialCatalogSeed, GrokCredentialCatalogService, GrokCredentialCatalogSnapshot, - GrokCredentialLifecycle, GrokCredentialQuotaService, GrokCredentialRecord, - GrokCredentialRecovery, GrokCredentialRecoveryOutcome, GrokCredentialRefreshError, - GrokCredentialRefreshOutcome, GrokCredentialRefreshService, GrokCredentialRefresher, - GrokCredentialRepository, GrokCredentialRepositoryError, GrokOAuthClient, GrokOAuthConfig, - GrokOAuthImportCandidate, GrokOAuthImportDocument, GrokOAuthImportEntry, GrokOAuthImportError, - GrokOAuthImportMetadata, GrokOAuthImportTokens, GrokOAuthRefreshClient, GrokOAuthSecret, - GrokPlanCatalog, GrokQuotaError, GrokQuotaPeriodKind, GrokQuotaSnapshot, GrokRefreshFailure, - GrokRefreshTokens, HttpHeader, HttpMethod, OAuthError, OAuthErrorCode, OAuthHttpRequest, - OAuthHttpResponse, OAuthHttpTransport, OAuthOperation, OAuthPrincipal, OFFICIAL_CLIENT_ID, - OFFICIAL_ISSUER, OFFICIAL_REDIRECT_URI, OFFICIAL_SCOPES, PendingAuthorization, Pkce, + GrokCredentialCatalogSeed, GrokCredentialCatalogService, GrokCredentialLifecycle, + GrokCredentialQuotaService, GrokCredentialRecord, GrokCredentialRecovery, + GrokCredentialRecoveryOutcome, GrokCredentialRefreshError, GrokCredentialRefreshOutcome, + GrokCredentialRefreshService, GrokCredentialRefresher, GrokCredentialRepository, + GrokCredentialRepositoryError, GrokOAuthClient, GrokOAuthConfig, GrokOAuthImportCandidate, + GrokOAuthImportDocument, GrokOAuthImportEntry, GrokOAuthImportError, GrokOAuthImportMetadata, + GrokOAuthImportTokens, GrokOAuthRefreshClient, GrokOAuthSecret, GrokPlanCatalog, + GrokQuotaError, GrokQuotaPeriodKind, GrokQuotaSnapshot, GrokRefreshFailure, GrokRefreshTokens, + HttpHeader, HttpMethod, OAuthError, OAuthErrorCode, OAuthHttpRequest, OAuthHttpResponse, + OAuthHttpTransport, OAuthOperation, OAuthPrincipal, OFFICIAL_CLIENT_ID, OFFICIAL_ISSUER, + OFFICIAL_REDIRECT_URI, OFFICIAL_SCOPES, PendingAuthorization, Pkce, PreparedGrokCredentialRotation, PreparedGrokCredentialRotationGuard, ProtocolViolation, RedirectUriAllowlist, RefreshTokenGrant, RefreshedTokenSet, ReqwestOidcTokenVerifier, RotateGrokCredential, RotateManagedGrokCredential, SecretValue, TokenCandidate, diff --git a/backend/crates/providers/xai/tests/credential/catalog.rs b/backend/crates/providers/xai/tests/credential/catalog.rs index a6ecc4012..250e25086 100644 --- a/backend/crates/providers/xai/tests/credential/catalog.rs +++ b/backend/crates/providers/xai/tests/credential/catalog.rs @@ -3,17 +3,19 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime}; -use futures::future::join_all; +use futures::future::{BoxFuture, join_all}; use gateway_core::engine::credential::{ AccountAvailability, AccountStateChange, CredentialCasUpdate, CredentialRevision, OpaqueProviderData, ProviderAccountId, ProviderAccountStore, ProviderAccountUpdate, QuotaObservation, }; -use gateway_core::routing::ConfigRevision; +use gateway_core::provider_ports::{ + ProviderCatalogCacheKey, ProviderCatalogCachePort, ProviderStoreError, +}; use provider_xai::{ GrokBillingRequest, GrokBillingTransport, GrokBillingTransportError, GrokBillingTransportErrorKind, GrokBillingTransportFuture, GrokBillingTransportResponse, - GrokCatalogScope, GrokCredentialCatalogCache, GrokCredentialCatalogError, + GrokCatalogCache, GrokCatalogScope, GrokCredentialCatalogCache, GrokCredentialCatalogError, GrokCredentialCatalogSeed, GrokCredentialRepository, GrokModelCatalogRequest, GrokModelCatalogTransport, GrokModelCatalogTransportError, GrokModelCatalogTransportErrorKind, GrokModelCatalogTransportFuture, GrokModelCatalogTransportResponse, GrokPlanCatalog, @@ -122,6 +124,31 @@ impl GrokBillingTransport for QueueBillingTransport { } } +/// 每次读取都返回同一份无法解码的 catalog 文档的存储端口。 +struct CorruptCatalogCachePort; + +impl ProviderCatalogCachePort for CorruptCatalogCachePort { + fn replace<'a>( + &'a self, + _key: &'a ProviderCatalogCacheKey, + _catalog: &'a OpaqueProviderData, + _ttl: Duration, + ) -> BoxFuture<'a, Result<(), ProviderStoreError>> { + Box::pin(async { Ok(()) }) + } + + fn read<'a>( + &'a self, + _key: &'a ProviderCatalogCacheKey, + ) -> BoxFuture<'a, Result, ProviderStoreError>> { + Box::pin(async { + let mut document = serde_json::Map::new(); + document.insert("version".to_owned(), serde_json::json!("corrupt")); + Ok(Some(OpaqueProviderData::new(document))) + }) + } +} + enum BillingMutation { State(AccountStateChange), Credential(CredentialCasUpdate), @@ -204,7 +231,7 @@ async fn concurrent_cold_scheduling_hydration_reads_quota_once() { } #[tokio::test] -async fn synchronization_caches_each_plan_and_returns_strict_union() { +async fn catalog_query_caches_each_plan_and_returns_strict_union() { let (store, repository) = repository_with_accounts(&[("catalog-a", "subject-a"), ("catalog-b", "subject-b")]).await; let cache = MemoryGrokCatalogCache::shared(); @@ -217,17 +244,13 @@ async fn synchronization_caches_each_plan_and_returns_strict_union() { ]); let service = crate::support::grok_catalog_service(repository, transport, cache_port); assert_eq!(service.catalog_generation().get(), 0); - let snapshot = service - .synchronize(ConfigRevision::new(7).expect("config revision")) - .await - .expect("catalog sync"); + let models = service.query_models().await.expect("catalog sync"); assert_eq!(service.catalog_generation().get(), 1); service.query_models().await.expect("same catalog sync"); assert_eq!(service.catalog_generation().get(), 1); - assert_eq!(snapshot.plans().len(), 1); - assert_eq!(snapshot.models().len(), 1); - assert_eq!(snapshot.models()[0].request_model().as_str(), "grok-4.5"); + assert_eq!(models.len(), 1); + assert_eq!(models[0].request_model().as_str(), "grok-4.5"); let account = store .account(&account_id("catalog-a")) .expect("created account"); @@ -292,6 +315,35 @@ async fn single_account_catalog_read_miss_does_not_call_upstream() { ); } +#[tokio::test] +async fn corrupt_catalog_cache_entry_reads_as_miss_and_refetches() { + let (store, repository) = + repository_with_accounts(&[("catalog-corrupt", "subject-corrupt")]).await; + let account = store + .account(&account_id("catalog-corrupt")) + .expect("created account"); + let scope = GrokCatalogScope::for_account(&account).expect("catalog scope"); + let cache = GrokCatalogCache::new(Arc::new(CorruptCatalogCachePort)).expect("provider cache"); + assert!( + cache + .read(&scope) + .await + .expect("corrupt cache entry must degrade to a miss") + .is_none() + ); + + let transport = QueueCatalogTransport::from_bodies([OFFICIAL_FIXTURE.to_vec()]); + let service = + crate::support::grok_catalog_service(repository, transport.clone(), Arc::new(cache)); + let catalog = service + .cached_or_refresh_account_catalog(&account) + .await + .expect("corrupt cache entry falls back to a live refresh"); + + assert_eq!(catalog.seed().models(), ["grok-4.5"]); + assert_eq!(transport.calls(), 1); +} + #[tokio::test] async fn cached_account_catalog_uses_shared_plan_cache_without_upstream_call() { let (store, repository) = @@ -424,9 +476,7 @@ async fn disabled_accounts_are_not_sent_to_catalog_transport() { cache_port, ); assert!(matches!( - service - .synchronize(ConfigRevision::new(1).expect("revision")) - .await, + service.query_models().await, Err(GrokCredentialCatalogError::NoEligibleCredential) )); } @@ -453,12 +503,12 @@ async fn quota_exhausted_account_remains_eligible_for_catalog_discovery() { MemoryGrokCatalogCache::shared(), ); - let snapshot = service - .synchronize(ConfigRevision::new(1).expect("revision")) + let models = service + .query_models() .await .expect("discover catalog through quota exhausted account"); - assert_eq!(snapshot.models().len(), 1); + assert_eq!(models.len(), 1); } #[tokio::test] @@ -471,9 +521,7 @@ async fn one_upstream_failure_rejects_the_whole_catalog_cycle() { cache_port, ); assert!(matches!( - service - .synchronize(ConfigRevision::new(1).expect("revision")) - .await, + service.query_models().await, Err(GrokCredentialCatalogError::Upstream) )); } @@ -505,9 +553,7 @@ async fn conflicting_facts_for_same_slug_fail_closed() { MemoryGrokCatalogCache::shared(), ); assert!(matches!( - service - .synchronize(ConfigRevision::new(1).expect("revision")) - .await, + service.query_models().await, Err(GrokCredentialCatalogError::ConflictingModelFacts) )); } diff --git a/backend/crates/providers/xai/tests/credential/oidc_verifier.rs b/backend/crates/providers/xai/tests/credential/oidc_verifier.rs index ce5a51226..a462fd1dc 100644 --- a/backend/crates/providers/xai/tests/credential/oidc_verifier.rs +++ b/backend/crates/providers/xai/tests/credential/oidc_verifier.rs @@ -49,6 +49,34 @@ async fn authorization_code_should_validate_es256_and_cache_jwks() { } } +#[tokio::test] +async fn jwks_with_unknown_members_should_still_verify() { + let server = MockServer::start().await; + let endpoint = endpoints(&server); + let key = TestKey::new(23, "extended-key"); + let mut jwks = key.jwks(); + jwks["cache_max_age"] = json!(3600); + jwks["keys"][0]["x5t"] = json!("thumbprint"); + jwks["keys"][0]["key_ops"] = json!(["verify"]); + mount_jwks(&server, jwks, 1).await; + let verifier = verifier(&endpoint.origin, CACHE_TTL); + let nonce = SecretValue::new("unknown-member-nonce".to_owned()); + let algorithms = vec!["ES256".to_owned()]; + let token = SecretValue::new(key.sign(&valid_claims(&nonce), Algorithm::ES256)); + let access_token = SecretValue::new("access-token".to_owned()); + + let evidence = verifier + .verify( + auth_context(&endpoint, &algorithms, &nonce), + TokenCandidate::new(&access_token, Some(&token), None), + ) + .await + .expect("JWKS with unknown members must still verify"); + + assert_eq!(evidence.method(), VerificationMethod::IdToken); + assert_eq!(evidence.subject(), "user-123"); +} + #[tokio::test] async fn fresh_cache_miss_should_refresh_once_for_key_rotation() { let server = MockServer::start().await;