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 <noreply@anthropic.com>
This commit is contained in:
zyycn
2026-07-27 01:51:41 +08:00
co-authored by Claude
parent 15455ca619
commit ad5f3ba997
6 changed files with 137 additions and 110 deletions
@@ -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<Option<GrokPlanCatalog>, 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<Utc>,
plans: Vec<GrokPlanCatalog>,
models: Vec<GrokCatalogModel>,
}
impl GrokCredentialCatalogSnapshot {
#[must_use]
pub const fn config_revision(&self) -> ConfigRevision {
self.config_revision
}
#[must_use]
pub const fn observed_at(&self) -> DateTime<Utc> {
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<GrokCredentialCatalogSnapshot, GrokCredentialCatalogError> {
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<Vec<GrokCatalogModel>, GrokCredentialCatalogError> {
Ok(self.fetch_and_cache().await?.models)
self.fetch_and_cache().await
}
async fn fetch_and_cache(&self) -> Result<FetchedProviderCatalog, GrokCredentialCatalogError> {
async fn fetch_and_cache(&self) -> Result<Vec<GrokCatalogModel>, 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::<Vec<_>>();
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<Utc>,
plans: Vec<GrokPlanCatalog>,
models: Vec<GrokCatalogModel>,
}
fn catalog_candidates_by_scope(
candidates: Vec<LoadedGrokCredential>,
) -> Result<BTreeMap<GrokCatalogScope, Vec<LoadedGrokCredential>>, GrokCredentialCatalogError> {
@@ -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::{
@@ -348,14 +348,13 @@ impl CachedJwk {
}
}
// RFC 7517 要求忽略 JWKS 与 JWK 的未知成员;字段本身的强校验在 parse_jwks。
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct JwksWire {
keys: Vec<JwkWire>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct JwkWire {
kty: String,
#[serde(rename = "use")]
+11 -11
View File
@@ -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,
@@ -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<Option<OpaqueProviderData>, 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)
));
}
@@ -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;