i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-09 04:30:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(workers): isolate legacy catalog credentials

Browse Source
This commit is contained in:
elky
2026-09-04 19:42:10 +08:00
parent 7c5cce4b3c
commit b72b6ab137
4 changed files with 346 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/aether-gateway/src/maintenance/runtime/oauth_token_refresh.rs
+83 -8
View File
@@ -51,9 +51,15 @@ pub(crate) async fn perform_oauth_token_refresh_once(
let endpoints = state
.list_provider_catalog_endpoints_by_provider_ids(&provider_ids)
.await?;
// Read the catalog rows without opening/decrypting credentials in bulk.
// A single legacy/plaintext row must not abort refresh for every healthy
// key, and this maintenance scan must not trigger the normal lazy v2
// credential rewrite path. Each candidate is opened in isolation below.
let keys = state
.data
.list_provider_catalog_keys_by_provider_ids(&provider_ids)
.await?;
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?;
let endpoints_by_provider = group_endpoints_by_provider(endpoints);
let keys_by_provider = group_keys_by_provider(keys);
let mut summary = OAuthTokenRefreshRunSummary::default();
@@ -85,12 +91,31 @@ pub(crate) async fn perform_oauth_token_refresh_once(
continue;
};
let Some(transport) = state
let transport = match state
.read_provider_transport_snapshot(&provider.id, &endpoint.id, &key.id)
.await?
else {
summary.skipped = summary.skipped.saturating_add(1);
continue;
.await
{
Ok(Some(transport)) => transport,
Ok(None) => {
summary.skipped = summary.skipped.saturating_add(1);
continue;
}
Err(err) if is_nonfatal_legacy_credential_error(&err) => {
// Keep malformed historical credentials untouched. They
// are intentionally skipped while other keys continue.
summary.skipped = summary.skipped.saturating_add(1);
warn!(
event_name = "oauth_token_refresh_skipped_invalid_credential",
log_type = "ops",
worker = "oauth_token_refresh",
provider_id = %provider.id,
key_id = %key.id,
reason = "invalid_stored_credential",
"gateway skipped oauth refresh for an invalid stored credential"
);
continue;
}
Err(err) => return Err(err),
};
let is_agent_identity =
crate::provider_transport::is_codex_agent_identity_transport(&transport);
@@ -236,8 +261,10 @@ async fn provider_key_credentials_changed(
before: &StoredProviderCatalogKey,
) -> Result<bool, GatewayError> {
let Some(after) = state
.data
.list_provider_catalog_keys_by_ids(std::slice::from_ref(&before.id))
.await?
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?
.into_iter()
.next()
else {
@@ -274,6 +301,29 @@ fn now_unix_secs() -> u64 {
.unwrap_or_default()
}
/// Credential decoding errors are expected for rows written by older
/// versions of the service. They are non-fatal for a best-effort maintenance
/// scan, but normal request/admin paths still fail closed on the same error.
fn is_nonfatal_legacy_credential_error(error: &GatewayError) -> bool {
let GatewayError::Internal(message) = error else {
return false;
};
let message = message.to_ascii_lowercase();
// Missing encryption configuration is an operational failure and must
// remain fail-closed. Only errors that identify a stored field or a
// malformed legacy ciphertext are safe to isolate to one key.
if message.contains("encryption key is not configured") {
return false;
}
message.contains("provider_api_keys.api_key")
|| message.contains("provider_api_keys.auth_config")
|| message.contains("legacy provider catalog credential")
|| message.contains("provider catalog credential is not an authenticated ciphertext")
|| message.contains("provider catalog credential contains reserved framing")
|| message.contains("provider catalog credential authentication failed")
|| message.contains("provider catalog credential envelope")
}
#[cfg(test)]
mod tests {
use aether_data_contracts::repository::provider_catalog::{
@@ -281,8 +331,10 @@ mod tests {
};
use super::{
agent_identity_needs_task_recovery, auth_config_has_refresh_token, oauth_refresh_candidate,
agent_identity_needs_task_recovery, auth_config_has_refresh_token,
is_nonfatal_legacy_credential_error, oauth_refresh_candidate,
};
use crate::GatewayError;
#[test]
fn legacy_antigravity_refresh_token_is_refreshable() {
@@ -335,4 +387,27 @@ mod tests {
Some("[REFRESH_FAILED] temporary"),
));
}
#[test]
fn only_stored_provider_credential_errors_are_non_fatal() {
assert!(is_nonfatal_legacy_credential_error(
&GatewayError::Internal(
"provider catalog credential is not an authenticated ciphertext".to_string(),
)
));
assert!(is_nonfatal_legacy_credential_error(
&GatewayError::Internal(
"provider_api_keys.auth_config has an invalid provider catalog credential envelope"
.to_string(),
)
));
assert!(!is_nonfatal_legacy_credential_error(
&GatewayError::Internal("postgres error: connection refused".to_string(),)
));
assert!(!is_nonfatal_legacy_credential_error(
&GatewayError::Internal(
"provider catalog credential encryption key is not configured".to_string(),
)
));
}
}
apps/aether-gateway/src/model_fetch/runtime.rs
+239 -10
View File
@@ -153,12 +153,10 @@ where
.push(endpoint);
}
let mut keys_by_provider = HashMap::<String, Vec<StoredProviderCatalogKey>>::new();
for key in <S as ModelFetchAssociationStore>::list_provider_catalog_keys_by_provider_ids(
state,
&provider_ids,
)
.await
.map_err(GatewayError::Internal)?
for key in state
.list_provider_catalog_keys_for_model_fetch(&provider_ids)
.await
.map_err(GatewayError::Internal)?
{
keys_by_provider
.entry(key.provider_id.clone())
@@ -180,6 +178,7 @@ where
continue;
}
let selected_endpoints = selected_models_fetch_endpoints(&endpoints, &key);
let key = sanitize_model_fetch_key(key);
targets.push(SelectedFetchTarget {
provider: provider.clone(),
key,
@@ -190,6 +189,56 @@ where
Ok(targets)
}
/// Keep only the key metadata needed after target collection. Raw catalog
/// rows contain encrypted credentials and transport secrets; model discovery
/// reopens a single snapshot by id when it actually needs to make a request.
fn sanitize_model_fetch_key(mut key: StoredProviderCatalogKey) -> StoredProviderCatalogKey {
// `SelectedFetchTarget` lives across endpoint selection and the complete
// fetch/persist operation. Keep only fields consumed by that operation;
// in particular, do not retain historical diagnostics, scheduling state,
// usage counters, or transport configuration copied from a raw database
// row. The actual credential/proxy snapshot is reopened by id for one
// endpoint at a time.
key.capabilities = None;
key.auth_type_by_format = None;
key.allow_auth_channel_mismatch_formats = None;
key.encrypted_api_key = None;
key.encrypted_auth_config = None;
key.note = None;
key.internal_priority = 0;
key.rate_multipliers = None;
key.global_priority_by_format = None;
key.expires_at_unix_secs = None;
key.cache_ttl_minutes = 0;
key.max_probe_interval_minutes = 0;
key.proxy = None;
key.fingerprint = None;
key.rpm_limit = None;
key.concurrent_limit = None;
key.learned_rpm_limit = None;
key.concurrent_429_count = None;
key.rpm_429_count = None;
key.last_429_at_unix_secs = None;
key.last_429_type = None;
key.adjustment_history = None;
key.utilization_samples = None;
key.last_probe_increase_at_unix_secs = None;
key.last_rpm_peak = None;
key.request_count = None;
key.total_tokens = 0;
key.total_cost_usd = 0.0;
key.success_count = None;
key.error_count = None;
key.total_response_time_ms = None;
key.last_used_at_unix_secs = None;
key.last_models_fetch_at_unix_secs = None;
key.last_models_fetch_error = None;
key.oauth_invalid_at_unix_secs = None;
key.oauth_invalid_reason = None;
key.status_snapshot = None;
key
}
async fn execute_fetch_targets<S>(
state: &S,
targets: Vec<SelectedFetchTarget>,
@@ -293,13 +342,14 @@ async fn fetch_and_persist_key_models(
}
let mut transports = Vec::new();
let mut skipped_invalid_credential = false;
for endpoint in &target.endpoints {
match state
.read_provider_transport_snapshot(&target.provider.id, &endpoint.id, &target.key.id)
.await?
.await
{
Some(transport) => transports.push(transport),
None => {
Ok(Some(transport)) => transports.push(transport),
Ok(None) => {
warn!(
provider_id = %target.provider.id,
endpoint_id = %endpoint.id,
@@ -307,9 +357,29 @@ async fn fetch_and_persist_key_models(
"gateway model fetch transport snapshot unavailable"
);
}
Err(error) if is_nonfatal_legacy_credential_error(&error) => {
skipped_invalid_credential = true;
warn!(
event_name = "model_fetch_skipped_invalid_credential",
log_type = "ops",
provider_id = %target.provider.id,
endpoint_id = %endpoint.id,
key_id = %target.key.id,
reason = "invalid_stored_credential",
"gateway skipped model fetch for an invalid stored credential"
);
}
Err(error) => return Err(error),
}
}
// A malformed legacy credential is isolated to its key. Do not turn it
// into a cycle-wide failure (or rewrite the row merely to record a fetch
// error), and let other eligible keys continue through the worker.
if transports.is_empty() && skipped_invalid_credential {
return Ok(KeyFetchDisposition::Skipped);
}
if transports.is_empty() {
persist_key_fetch_failure(
state,
@@ -482,6 +552,30 @@ pub(crate) fn safe_model_fetch_error(error: &str) -> String {
"Upstream models fetch failed".to_string()
}
/// Credential decoding failures from old catalog rows are isolated by the
/// background model-fetch worker. Normal request/admin paths remain fail
/// closed; this predicate only controls whether one maintenance item may be
/// skipped without aborting the whole cycle.
fn is_nonfatal_legacy_credential_error(error: &GatewayError) -> bool {
let GatewayError::Internal(message) = error else {
return false;
};
let message = message.to_ascii_lowercase();
// Missing encryption configuration is an operational failure and must
// remain fail-closed. Only errors that identify a stored field or a
// malformed legacy ciphertext are safe to isolate to one key.
if message.contains("encryption key is not configured") {
return false;
}
message.contains("provider_api_keys.api_key")
|| message.contains("provider_api_keys.auth_config")
|| message.contains("legacy provider catalog credential")
|| message.contains("provider catalog credential is not an authenticated ciphertext")
|| message.contains("provider catalog credential contains reserved framing")
|| message.contains("provider catalog credential authentication failed")
|| message.contains("provider catalog credential envelope")
}
fn model_fetch_error_http_status(error: &str) -> Option<u16> {
[
"http ",
@@ -553,7 +647,8 @@ fn now_unix_secs() -> u64 {
#[cfg(test)]
mod tests {
use super::{
perform_model_fetch_once_with_state, safe_model_fetch_error, state::ModelFetchRuntimeState,
perform_model_fetch_once_with_state, safe_model_fetch_error, sanitize_model_fetch_key,
state::ModelFetchRuntimeState,
};
use aether_contracts::{ExecutionPlan, ExecutionResult, ProxySnapshot};
use aether_data_contracts::repository::global_models::{
@@ -585,6 +680,7 @@ mod tests {
endpoints: Arc<Vec<StoredProviderCatalogEndpoint>>,
keys: Arc<Mutex<Vec<StoredProviderCatalogKey>>>,
transports: Arc<HashMap<(String, String, String), GatewayProviderTransportSnapshot>>,
transport_errors: Arc<HashMap<(String, String, String), String>>,
execution_results: Arc<Mutex<VecDeque<ExecutionResult>>>,
executed_plans: Arc<Mutex<Vec<ExecutionPlan>>>,
cached_models: Arc<Mutex<HashMap<(String, String), Vec<Value>>>>,
@@ -604,6 +700,7 @@ mod tests {
endpoints: Arc::new(endpoints),
keys: Arc::new(Mutex::new(keys)),
transports: Arc::new(transports),
transport_errors: Arc::new(HashMap::new()),
execution_results: Arc::new(Mutex::new(VecDeque::from(execution_results))),
executed_plans: Arc::new(Mutex::new(Vec::new())),
cached_models: Arc::new(Mutex::new(HashMap::new())),
@@ -611,6 +708,14 @@ mod tests {
}
}
fn with_transport_errors(
mut self,
transport_errors: HashMap<(String, String, String), String>,
) -> Self {
self.transport_errors = Arc::new(transport_errors);
self
}
fn key(&self, key_id: &str) -> StoredProviderCatalogKey {
self.keys
.lock()
@@ -758,6 +863,13 @@ mod tests {
endpoint_id: &str,
key_id: &str,
) -> Result<Option<GatewayProviderTransportSnapshot>, GatewayError> {
if let Some(error) = self.transport_errors.get(&(
provider_id.to_string(),
endpoint_id.to_string(),
key_id.to_string(),
)) {
return Err(GatewayError::Internal(error.clone()));
}
Ok(self
.transports
.get(&(
@@ -1044,6 +1156,15 @@ mod tests {
}
}
#[test]
fn invalid_credential_classifier_does_not_swallow_missing_key_configuration() {
assert!(!super::is_nonfatal_legacy_credential_error(
&GatewayError::Internal(
"provider catalog credential encryption key is not configured".to_string(),
)
));
}
#[tokio::test]
async fn gateway_runtime_state_supports_shared_models_fetch_plan_builder() {
let state = TestState::default();
@@ -1383,6 +1504,114 @@ mod tests {
);
}
#[tokio::test]
async fn model_fetch_isolates_malformed_legacy_key_from_healthy_key() {
let provider = sample_provider("provider-openai", "openai");
let endpoint = sample_endpoint(
"endpoint-openai-responses",
"provider-openai",
"openai:responses",
);
let mut malformed = sample_key(
"key-openai-malformed",
"provider-openai",
"api_key",
&["openai:responses"],
);
malformed.encrypted_api_key = Some("legacy-plaintext-or-corrupt".to_string());
malformed.allowed_models = Some(json!(["legacy-model"]));
malformed.last_models_fetch_error = Some("previous error".to_string());
let malformed_ciphertext = malformed.encrypted_api_key.clone();
let healthy = sample_key(
"key-openai-healthy",
"provider-openai",
"api_key",
&["openai:responses"],
);
let healthy_transport = sample_transport(
"openai",
"provider-openai",
"endpoint-openai-responses",
"key-openai-healthy",
"openai:responses",
"api_key",
None,
);
let state = TestState::new(
vec![provider],
vec![endpoint],
vec![malformed, healthy],
HashMap::from([(
(
"provider-openai".to_string(),
"endpoint-openai-responses".to_string(),
"key-openai-healthy".to_string(),
),
healthy_transport,
)]),
vec![execution_result(json!({
"data": [{"id": "gpt-healthy"}]
}))],
)
.with_transport_errors(HashMap::from([(
(
"provider-openai".to_string(),
"endpoint-openai-responses".to_string(),
"key-openai-malformed".to_string(),
),
"provider_api_keys.api_key is not an authenticated ciphertext".to_string(),
)]));
let summary = perform_model_fetch_once_with_state(&state)
.await
.expect("one malformed key must not abort the cycle");
assert_eq!(summary.attempted, 2);
assert_eq!(summary.skipped, 1);
assert_eq!(summary.succeeded, 1);
let malformed_after = state.key("key-openai-malformed");
assert_eq!(malformed_after.encrypted_api_key, malformed_ciphertext);
assert_eq!(
malformed_after.allowed_models,
Some(json!(["legacy-model"]))
);
assert_eq!(
malformed_after.last_models_fetch_error.as_deref(),
Some("previous error")
);
assert_eq!(
state.key("key-openai-healthy").allowed_models,
Some(json!(["gpt-healthy"]))
);
}
#[test]
fn sanitized_model_fetch_key_drops_raw_transport_and_diagnostic_state() {
let mut key = sample_key("key-sanitize", "provider", "api_key", &["openai:responses"]);
key.capabilities = Some(json!({"secret": "capability"}));
key.note = Some("operator note".to_string());
key.proxy = Some(json!({"url": "http://user:pass@example.test"}));
key.last_models_fetch_error = Some("upstream detail".to_string());
key.oauth_invalid_reason = Some("token detail".to_string());
key.allowed_models = Some(json!(["keep-model-filter"]));
key.upstream_metadata = Some(json!({"provider": {"quota": 1}}));
let sanitized = sanitize_model_fetch_key(key);
assert_eq!(sanitized.encrypted_api_key, None);
assert_eq!(sanitized.encrypted_auth_config, None);
assert_eq!(sanitized.proxy, None);
assert_eq!(sanitized.fingerprint, None);
assert_eq!(sanitized.note, None);
assert_eq!(sanitized.last_models_fetch_error, None);
assert_eq!(sanitized.oauth_invalid_reason, None);
assert_eq!(sanitized.allowed_models, Some(json!(["keep-model-filter"])));
assert_eq!(
sanitized.upstream_metadata,
Some(json!({"provider": {"quota": 1}}))
);
}
#[tokio::test]
async fn model_fetch_failure_does_not_persist_upstream_error_body_credentials() {
const UPSTREAM_SECRET: &str = "upstream-secret-token-value";
apps/aether-gateway/src/model_fetch/runtime/state.rs
+14
View File
@@ -31,6 +31,20 @@ pub(crate) trait ModelFetchRuntimeState:
provider_ids: &[String],
) -> Result<Vec<StoredProviderCatalogEndpoint>, GatewayError>;
/// Return raw catalog key rows for the background fetcher. Production
/// implementations should avoid the normal bulk credential-opening
/// wrapper here: one malformed legacy row must not prevent healthy keys
/// from being considered, and a maintenance scan must not lazily rewrite
/// historical ciphertext. Test implementations can use the association
/// store's existing method via this default.
async fn list_provider_catalog_keys_for_model_fetch(
&self,
provider_ids: &[String],
) -> Result<Vec<StoredProviderCatalogKey>, String> {
self.list_provider_catalog_keys_by_provider_ids(provider_ids)
.await
}
async fn read_provider_transport_snapshot(
&self,
provider_id: &str,
apps/aether-gateway/src/state/integrations.rs
+10
View File
@@ -404,6 +404,16 @@ impl ModelFetchRuntimeState for AppState {
AppState::list_provider_catalog_endpoints_by_provider_ids(self, provider_ids).await
}
async fn list_provider_catalog_keys_for_model_fetch(
&self,
provider_ids: &[String],
) -> Result<Vec<StoredProviderCatalogKey>, String> {
self.data
.list_provider_catalog_keys_by_provider_ids(provider_ids)
.await
.map_err(|err| err.to_string())
}
async fn read_provider_transport_snapshot(
&self,
provider_id: &str,