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

feat(oauth): 完善账号异常识别并在调度/展示层拦截失效 OAuth 密钥

Browse Source 
- 新增 aether-admin provider status 模块，统一解析账号状态（禁用/工作区停用等）
- 调度器 runtime 增加 oauth_invalid 判定，跳过刷新失败或已撤销的 OAuth 密钥（REQUEST_FAILED 保留可选）
- gateway state 在 local oauth 刷新返回 4xx 时持久化失败原因并同步状态快照
- admin pool 列表/详情回填 account 状态与 scheduling 阻塞原因（account_blocked）
- 共享 catalog 的 status_snapshot payload 附加 account 字段
This commit is contained in:
fawney19
2026-04-19 20:50:31 +08:00
parent d719a1329c
commit 77aac74590
12 changed files with 1409 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
apps/aether-gateway/src/handlers/admin/provider/pool_admin/payloads.rs
View File

@@ -674,6 +674,11 @@ fn admin_pool_scheduling_payload(
cooldown_ttl_seconds: Option<u64>,
health_score: f64,
circuit_breaker_open: bool,
account_blocked: bool,
account_status_code: Option<&str>,
account_status_label: Option<&str>,
account_status_reason: Option<&str>,
account_status_source: Option<&str>,
account_quota_exhausted: bool,
) -> (String, String, String, Vec<serde_json::Value>) {
if !key.is_active {
@@ -691,6 +696,21 @@ fn admin_pool_scheduling_payload(
})],
);
}
if account_blocked {
return (
"blocked".to_string(),
"account_blocked".to_string(),
account_status_label.unwrap_or("账号异常").to_string(),
vec![json!({
"code": account_status_code.unwrap_or("account_blocked"),
"label": account_status_label.unwrap_or("账号异常"),
"blocking": true,
"source": account_status_source,
"ttl_seconds": serde_json::Value::Null,
"detail": account_status_reason,
})],
);
}
if account_quota_exhausted {
return (
"blocked".to_string(),
@@ -776,15 +796,6 @@ pub(super) fn build_admin_pool_key_payload(
.as_ref()
.is_some_and(|config| config.skip_exhausted_accounts)
&& admin_provider_pool_pure::admin_pool_key_account_quota_exhausted(key, provider_type);
let (scheduling_status, scheduling_reason, scheduling_label, scheduling_reasons) =
admin_pool_scheduling_payload(
key,
cooldown_reason.as_deref(),
cooldown_ttl_seconds,
health_score,
circuit_breaker_open,
account_quota_exhausted,
);
let auth_config = state.parse_catalog_auth_config_json(key);
let oauth_expires_at =
admin_pool_derive_oauth_expires_at(provider_type, key, auth_config.as_ref());
@@ -841,6 +852,20 @@ pub(super) fn build_admin_pool_key_payload(
.unwrap_or(false);
let account_status_source =
admin_pool_trimmed_string(account_snapshot.and_then(|item| item.get("source")));
let (scheduling_status, scheduling_reason, scheduling_label, scheduling_reasons) =
admin_pool_scheduling_payload(
key,
cooldown_reason.as_deref(),
cooldown_ttl_seconds,
health_score,
circuit_breaker_open,
account_status_blocked,
account_status_code.as_deref(),
account_status_label.as_deref(),
account_status_reason.as_deref(),
account_status_source.as_deref(),
account_quota_exhausted,
);
let mut payload = serde_json::Map::new();
payload.insert("key_id".to_string(), json!(key.id));

96
apps/aether-gateway/src/handlers/shared/catalog.rs
View File

@@ -2,6 +2,7 @@ use crate::handlers::shared::{json_string_list, unix_secs_to_rfc3339};
use crate::provider_key_auth::provider_key_auth_semantics;
use crate::AppState;
use aether_admin::provider::quota as admin_provider_quota_pure;
use aether_admin::provider::status as admin_provider_status_pure;
#[cfg(test)]
use aether_crypto::DEVELOPMENT_ENCRYPTION_KEY;
use aether_crypto::{decrypt_python_fernet_ciphertext, encrypt_python_fernet_plaintext};
@@ -154,6 +155,25 @@ pub(crate) fn default_provider_key_status_snapshot() -> serde_json::Value {
})
}
fn build_provider_key_account_status_snapshot(
key: &StoredProviderCatalogKey,
provider_type: &str,
) -> Value {
let snapshot = admin_provider_status_pure::resolve_account_status_snapshot(
Some(provider_type),
key.upstream_metadata.as_ref(),
key.oauth_invalid_reason.as_deref(),
);
json!({
"code": snapshot.code,
"label": snapshot.label,
"reason": snapshot.reason,
"blocked": snapshot.blocked,
"source": snapshot.source,
"recoverable": snapshot.recoverable,
})
}
fn provider_key_status_snapshot_object(
status_snapshot: Option<&Value>,
) -> Option<Map<String, Value>> {
@@ -817,20 +837,29 @@ pub(crate) fn provider_key_status_snapshot_payload(
.and_then(|snapshot| snapshot.get("quota"))
.and_then(Value::as_object);
if quota_snapshot_has_materialized_data(quota_snapshot, provider_type) {
return status_snapshot
let payload = if quota_snapshot_has_materialized_data(quota_snapshot, provider_type) {
status_snapshot
.cloned()
.unwrap_or_else(default_provider_key_status_snapshot);
}
.unwrap_or_else(default_provider_key_status_snapshot)
} else {
sync_provider_key_quota_status_snapshot(
status_snapshot,
provider_type,
key.upstream_metadata.as_ref(),
"catalog_fallback",
)
.or_else(|| status_snapshot.cloned())
.unwrap_or_else(default_provider_key_status_snapshot)
};
sync_provider_key_quota_status_snapshot(
status_snapshot,
provider_type,
key.upstream_metadata.as_ref(),
"catalog_fallback",
)
.or_else(|| status_snapshot.cloned())
.unwrap_or_else(default_provider_key_status_snapshot)
let mut snapshot = provider_key_status_snapshot_object(Some(&payload))
.or_else(|| default_provider_key_status_snapshot().as_object().cloned())
.unwrap_or_default();
snapshot.insert(
"account".to_string(),
build_provider_key_account_status_snapshot(key, provider_type),
);
Value::Object(snapshot)
}
pub(crate) fn provider_key_health_summary(
@@ -1540,4 +1569,47 @@ mod tests {
Some(2usize)
);
}
#[test]
fn provider_key_status_snapshot_payload_backfills_account_block_from_oauth_invalid_reason() {
let mut key = sample_catalog_key();
key.oauth_invalid_reason = Some("[ACCOUNT_BLOCK] account has been deactivated".to_string());
let payload = provider_key_status_snapshot_payload(&key, "codex");
let account = payload
.get("account")
.and_then(Value::as_object)
.expect("account snapshot should be object");
assert_eq!(account.get("code"), Some(&json!("account_disabled")));
assert_eq!(account.get("label"), Some(&json!("账号停用")));
assert_eq!(
account.get("reason"),
Some(&json!("account has been deactivated"))
);
assert_eq!(account.get("blocked"), Some(&json!(true)));
assert_eq!(account.get("source"), Some(&json!("oauth_invalid")));
}
#[test]
fn provider_key_status_snapshot_payload_backfills_workspace_deactivated_from_metadata() {
let mut key = sample_catalog_key();
key.upstream_metadata = Some(json!({
"codex": {
"account_disabled": true,
"reason": "deactivated_workspace"
}
}));
let payload = provider_key_status_snapshot_payload(&key, "codex");
let account = payload
.get("account")
.and_then(Value::as_object)
.expect("account snapshot should be object");
assert_eq!(account.get("code"), Some(&json!("workspace_deactivated")));
assert_eq!(account.get("label"), Some(&json!("工作区停用")));
assert_eq!(account.get("blocked"), Some(&json!(true)));
assert_eq!(account.get("source"), Some(&json!("metadata")));
}
}

48
apps/aether-gateway/src/scheduler/candidate/runtime.rs
View File

@@ -22,6 +22,7 @@ pub(super) struct CandidateRuntimeSelectionSnapshot {
pub(super) provider_key_rpm_states: BTreeMap<String, StoredProviderCatalogKey>,
provider_quota_blocks_requests: BTreeMap<String, bool>,
key_account_quota_exhausted: BTreeMap<String, bool>,
key_oauth_invalid: BTreeMap<String, bool>,
provider_key_rpm_reset_ats: BTreeMap<String, Option<u64>>,
}
@@ -40,6 +41,8 @@ pub(super) async fn read_candidate_runtime_selection_snapshot(
&provider_key_rpm_states,
&provider_skip_exhausted_accounts,
);
let key_oauth_invalid =
read_key_oauth_invalid_map(candidates, &provider_key_rpm_states, now_unix_secs);
let provider_quota_blocks_requests =
read_provider_quota_block_map(state, candidates, now_unix_secs).await?;
let provider_key_rpm_reset_ats =
@@ -51,6 +54,7 @@ pub(super) async fn read_candidate_runtime_selection_snapshot(
provider_key_rpm_states,
provider_quota_blocks_requests,
key_account_quota_exhausted,
key_oauth_invalid,
provider_key_rpm_reset_ats,
})
}
@@ -104,6 +108,11 @@ pub(super) fn is_candidate_selectable(
.get(candidate.key_id.as_str())
.copied()
.unwrap_or(false),
oauth_invalid: snapshot
.key_oauth_invalid
.get(candidate.key_id.as_str())
.copied()
.unwrap_or(false),
rpm_reset_at: snapshot
.provider_key_rpm_reset_ats
.get(candidate.key_id.as_str())
@@ -142,6 +151,11 @@ pub(super) fn current_candidate_runtime_skip_reason(
.get(candidate.key_id.as_str())
.copied()
.unwrap_or(false),
oauth_invalid: snapshot
.key_oauth_invalid
.get(candidate.key_id.as_str())
.copied()
.unwrap_or(false),
rpm_reset_at,
})
}
@@ -270,6 +284,40 @@ fn read_key_account_quota_exhaustion_map(
.collect()
}
fn read_key_oauth_invalid_map(
candidates: &[SchedulerMinimalCandidateSelectionCandidate],
provider_key_rpm_states: &BTreeMap<String, StoredProviderCatalogKey>,
now_unix_secs: u64,
) -> BTreeMap<String, bool> {
candidates
.iter()
.map(|candidate| {
let oauth_invalid = provider_key_rpm_states
.get(candidate.key_id.as_str())
.is_some_and(|key| key_requires_oauth_reauth(key, now_unix_secs));
(candidate.key_id.clone(), oauth_invalid)
})
.collect()
}
fn key_requires_oauth_reauth(key: &StoredProviderCatalogKey, now_unix_secs: u64) -> bool {
if !key.auth_type.trim().eq_ignore_ascii_case("oauth") {
return false;
}
let invalid_reason = key
.oauth_invalid_reason
.as_deref()
.map(str::trim)
.unwrap_or_default();
if !invalid_reason.is_empty() {
return !invalid_reason.starts_with("[REQUEST_FAILED]");
}
key.expires_at_unix_secs
.is_some_and(|value| value > 0 && value <= now_unix_secs)
}
fn read_provider_key_rpm_reset_at_map(
state: &(impl SchedulerRuntimeState + ?Sized),
candidates: &[SchedulerMinimalCandidateSelectionCandidate],

137
apps/aether-gateway/src/scheduler/candidate/tests/selection.rs
View File

@@ -1241,6 +1241,143 @@ async fn skips_codex_candidate_when_account_quota_is_exhausted_and_pool_flag_ena
assert_eq!(skipped[0].skip_reason, "account_quota_exhausted");
}
#[tokio::test]
async fn skips_oauth_invalid_candidate_before_local_auth_resolution() {
let mut first = sample_row();
first.provider_id = "provider-codex".to_string();
first.provider_name = "codex".to_string();
first.provider_type = "codex".to_string();
first.endpoint_id = "endpoint-codex".to_string();
first.endpoint_api_format = "openai:cli".to_string();
first.key_id = "key-codex".to_string();
first.key_name = "codex-invalid".to_string();
first.key_auth_type = "oauth".to_string();
first.key_api_formats = Some(vec!["openai:cli".to_string()]);
first.key_global_priority_by_format = Some(serde_json::json!({"openai:cli": 1}));
let mut second = sample_row();
second.provider_id = "provider-openai".to_string();
second.provider_name = "openai".to_string();
second.endpoint_id = "endpoint-openai".to_string();
second.endpoint_api_format = "openai:cli".to_string();
second.key_id = "key-openai".to_string();
second.key_name = "fallback".to_string();
second.key_api_formats = Some(vec!["openai:cli".to_string()]);
second.key_global_priority_by_format = Some(serde_json::json!({"openai:cli": 2}));
let candidates = Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed(vec![
first, second,
]));
let mut codex_provider = sample_provider("provider-codex", None);
codex_provider.provider_type = "codex".to_string();
let provider_catalog = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![codex_provider, sample_provider("provider-openai", None)],
Vec::new(),
vec![
{
let mut key = sample_key("key-codex", "provider-codex", Some(10));
key.auth_type = "oauth".to_string();
key.oauth_invalid_at_unix_secs = Some(1_710_000_000);
key.oauth_invalid_reason = Some(
"[REFRESH_FAILED] Token 续期失败 (401): refresh_token 已被使用并轮换，请重新登录授权"
.to_string(),
);
key
},
sample_key("key-openai", "provider-openai", Some(10)),
],
));
let quotas = Arc::new(InMemoryProviderQuotaRepository::seed(vec![]));
let request_candidates = Arc::new(InMemoryRequestCandidateRepository::seed(vec![]));
let state = AppState::new()
.expect("state should build")
.with_data_state_for_tests(
GatewayDataState::with_candidate_selection_provider_catalog_quota_and_request_candidates_for_tests(
candidates,
provider_catalog,
quotas,
request_candidates,
),
);
let (selected, skipped) = collect_selectable_candidates_with_skip_reasons(
state.data.as_ref(),
&state,
"openai:cli",
"gpt-4.1",
false,
None,
1_710_000_100,
)
.await
.expect("selection should succeed");
assert_eq!(selected.len(), 1);
assert_eq!(selected[0].provider_id, "provider-openai");
assert_eq!(skipped.len(), 1);
assert_eq!(skipped[0].candidate.provider_id, "provider-codex");
assert_eq!(skipped[0].skip_reason, "oauth_invalid");
}
#[tokio::test]
async fn keeps_request_failed_oauth_candidate_selectable() {
let mut row = sample_row();
row.provider_id = "provider-codex".to_string();
row.provider_name = "codex".to_string();
row.provider_type = "codex".to_string();
row.endpoint_id = "endpoint-codex".to_string();
row.endpoint_api_format = "openai:cli".to_string();
row.key_id = "key-codex".to_string();
row.key_name = "codex-check-failed".to_string();
row.key_auth_type = "oauth".to_string();
row.key_api_formats = Some(vec!["openai:cli".to_string()]);
let candidates = Arc::new(InMemoryMinimalCandidateSelectionReadRepository::seed(vec![
row,
]));
let mut provider = sample_provider("provider-codex", None);
provider.provider_type = "codex".to_string();
let provider_catalog = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
Vec::new(),
vec![{
let mut key = sample_key("key-codex", "provider-codex", Some(10));
key.auth_type = "oauth".to_string();
key.oauth_invalid_at_unix_secs = Some(1_710_000_000);
key.oauth_invalid_reason = Some("[REQUEST_FAILED] 账号状态检查失败".to_string());
key
}],
));
let quotas = Arc::new(InMemoryProviderQuotaRepository::seed(vec![]));
let request_candidates = Arc::new(InMemoryRequestCandidateRepository::seed(vec![]));
let state = AppState::new()
.expect("state should build")
.with_data_state_for_tests(
GatewayDataState::with_candidate_selection_provider_catalog_quota_and_request_candidates_for_tests(
candidates,
provider_catalog,
quotas,
request_candidates,
),
);
let (selected, skipped) = collect_selectable_candidates_with_skip_reasons(
state.data.as_ref(),
&state,
"openai:cli",
"gpt-4.1",
false,
None,
1_710_000_100,
)
.await
.expect("selection should succeed");
assert_eq!(selected.len(), 1);
assert_eq!(selected[0].provider_id, "provider-codex");
assert!(skipped.is_empty());
}
#[tokio::test]
async fn keeps_codex_candidate_selectable_when_exhausted_account_flag_is_disabled() {
let mut first = sample_row();

252
apps/aether-gateway/src/state/oauth.rs
View File

@@ -49,6 +49,157 @@ fn tagged_reason(reason: Option<&str>, prefix: &str) -> Option<String> {
})
}
fn oauth_invalid_reason_is_account_block(reason: Option<&str>) -> bool {
reason
.map(str::trim)
.is_some_and(|value| value.starts_with(OAUTH_ACCOUNT_BLOCK_PREFIX))
}
fn normalize_local_oauth_refresh_error_message(
status_code: Option<u16>,
body_excerpt: Option<&str>,
) -> String {
let mut message = None::<String>;
let mut error_code = None::<String>;
let mut error_type = None::<String>;
if let Some(body_excerpt) = body_excerpt {
if let Ok(value) = serde_json::from_str::<serde_json::Value>(body_excerpt) {
if let Some(object) = value.as_object() {
if let Some(error_object) =
object.get("error").and_then(serde_json::Value::as_object)
{
message = error_object
.get("message")
.or_else(|| error_object.get("error_description"))
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned);
error_code = error_object
.get("code")
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| value.to_ascii_lowercase());
error_type = error_object
.get("type")
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| value.to_ascii_lowercase());
}
if message.is_none() {
message = object
.get("message")
.or_else(|| object.get("error_description"))
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned);
}
if error_code.is_none() {
error_code = object
.get("code")
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| value.to_ascii_lowercase());
}
if error_type.is_none() {
error_type = object
.get("type")
.and_then(serde_json::Value::as_str)
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| value.to_ascii_lowercase());
}
}
}
}
let message = message
.or_else(|| {
body_excerpt
.map(str::trim)
.filter(|value| !value.is_empty())
.map(|value| value.chars().take(300).collect::<String>())
})
.unwrap_or_default();
let lowered = message.to_ascii_lowercase();
let error_code = error_code.unwrap_or_default();
let error_type = error_type.unwrap_or_default();
if error_code == "refresh_token_reused"
|| lowered.contains("already been used to generate a new access token")
{
return "refresh_token 已被使用并轮换，请重新登录授权".to_string();
}
if error_code == "invalid_grant"
|| error_code == "invalid_refresh_token"
|| (lowered.contains("refresh token")
&& ["expired", "revoked", "invalid"]
.iter()
.any(|keyword| lowered.contains(keyword)))
{
return "refresh_token 无效、已过期或已撤销，请重新登录授权".to_string();
}
if error_type == "invalid_request_error" && !message.is_empty() {
return message;
}
if !message.is_empty() {
return message;
}
status_code
.map(|status_code| format!("HTTP {status_code}"))
.unwrap_or_else(|| "未知错误".to_string())
}
fn merge_local_oauth_refresh_failure_reason(
current_reason: Option<&str>,
refresh_reason: &str,
) -> Option<String> {
let current_reason = current_reason.map(str::trim).unwrap_or_default();
let refresh_reason = refresh_reason.trim();
if refresh_reason.is_empty() {
return (!current_reason.is_empty()).then(|| current_reason.to_string());
}
if current_reason.is_empty() {
return Some(refresh_reason.to_string());
}
if current_reason.starts_with(OAUTH_EXPIRED_PREFIX) {
return None;
}
if current_reason.starts_with(OAUTH_ACCOUNT_BLOCK_PREFIX) {
if let Some((head, _)) = current_reason.split_once("[REFRESH_FAILED]") {
return Some(
format!("{}\n{}", head.trim_end(), refresh_reason)
.trim()
.to_string(),
);
}
return Some(format!("{current_reason}\n{refresh_reason}"));
}
Some(refresh_reason.to_string())
}
fn local_oauth_refresh_success_invalid_state(
key: &StoredProviderCatalogKey,
) -> (Option<u64>, Option<String>) {
let current_reason = key
.oauth_invalid_reason
.as_deref()
.map(str::trim)
.unwrap_or_default();
if oauth_invalid_reason_is_account_block(Some(current_reason)) {
return (
key.oauth_invalid_at_unix_secs,
Some(current_reason.to_string()),
);
}
(None, None)
}
fn default_oauth_status_snapshot_value() -> Value {
default_provider_key_status_snapshot()
.get("oauth")
@@ -557,7 +708,7 @@ impl AppState {
let executor = GatewayLocalOAuthHttpExecutor { state: self };
for _ in 0..2 {
let resolution = self
let resolution = match self
.oauth_refresh
.resolve_with_result(
&executor,
@@ -566,7 +717,32 @@ impl AppState {
Some(lock_owner.as_str()),
)
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?;
{
Ok(resolution) => resolution,
Err(provider_transport::LocalOAuthRefreshError::HttpStatus {
status_code,
body_excerpt,
..
}) if matches!(status_code, 400 | 401 | 403) => {
if let Err(err) = self
.persist_local_oauth_refresh_failure_state(
&current_transport,
status_code,
body_excerpt.as_str(),
)
.await
{
tracing::warn!(
key_id = %current_transport.key.id,
provider_type = %current_transport.provider.provider_type,
error = ?err,
"gateway local oauth refresh failure persistence failed"
);
}
return Ok(None);
}
Err(err) => return Err(GatewayError::Internal(err.to_string())),
};
if resolution
.as_ref()
@@ -790,8 +966,10 @@ impl AppState {
latest_key.encrypted_auth_config = encrypted_auth_config;
latest_key.is_active = true;
latest_key.expires_at_unix_secs = entry.expires_at_unix_secs;
latest_key.oauth_invalid_at_unix_secs = None;
latest_key.oauth_invalid_reason = None;
let (oauth_invalid_at_unix_secs, oauth_invalid_reason) =
local_oauth_refresh_success_invalid_state(&latest_key);
latest_key.oauth_invalid_at_unix_secs = oauth_invalid_at_unix_secs;
latest_key.oauth_invalid_reason = oauth_invalid_reason;
latest_key.updated_at_unix_secs = Some(
SystemTime::now()
.duration_since(UNIX_EPOCH)
@@ -806,6 +984,72 @@ impl AppState {
Ok(())
}
async fn persist_local_oauth_refresh_failure_state(
&self,
transport: &provider_transport::GatewayProviderTransportSnapshot,
status_code: u16,
body_excerpt: &str,
) -> Result<bool, GatewayError> {
let key_id = transport.key.id.trim();
if key_id.is_empty() {
return Ok(false);
}
let Some(mut latest_key) = self
.data
.list_provider_catalog_keys_by_ids(&[key_id.to_string()])
.await
.map_err(|err| GatewayError::Internal(err.to_string()))?
.into_iter()
.next()
else {
return Ok(false);
};
if !provider_key_is_oauth_managed(&latest_key, transport.provider.provider_type.as_str()) {
return Ok(false);
}
let refresh_reason = format!(
"{OAUTH_REFRESH_FAILED_PREFIX}Token 续期失败 ({status_code}): {}",
normalize_local_oauth_refresh_error_message(Some(status_code), Some(body_excerpt))
);
let Some(merged_reason) = merge_local_oauth_refresh_failure_reason(
latest_key.oauth_invalid_reason.as_deref(),
&refresh_reason,
) else {
return Ok(false);
};
if latest_key.oauth_invalid_reason.as_deref() == Some(merged_reason.as_str())
&& latest_key.oauth_invalid_at_unix_secs.is_some()
{
return Ok(false);
}
let now_unix_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.ok()
.map(|duration| duration.as_secs())
.unwrap_or(0);
latest_key.oauth_invalid_at_unix_secs = latest_key
.oauth_invalid_at_unix_secs
.or(Some(now_unix_secs));
latest_key.oauth_invalid_reason = Some(merged_reason);
latest_key.updated_at_unix_secs = Some(now_unix_secs);
let current_status_snapshot = latest_key.status_snapshot.take();
latest_key.status_snapshot =
sync_provider_key_oauth_status_snapshot(current_status_snapshot, &latest_key);
let updated = self
.update_provider_catalog_key(&latest_key)
.await?
.is_some();
if updated {
let _ = self.invalidate_local_oauth_refresh_entry(key_id).await;
}
Ok(updated)
}
async fn execute_local_oauth_http_request(
&self,
provider_type: &'static str,

132
apps/aether-gateway/src/tests/control/admin/oauth.rs
View File

@@ -2737,6 +2737,138 @@ async fn gateway_starts_admin_provider_oauth_kiro_batch_import_task_locally_with
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_marks_lazy_codex_oauth_refresh_failures_as_invalid() {
let token_hits = Arc::new(Mutex::new(0usize));
let token_hits_clone = Arc::clone(&token_hits);
let token_server = Router::new().route(
"/oauth/token",
post(move |_request: Request| {
let token_hits_inner = Arc::clone(&token_hits_clone);
async move {
*token_hits_inner.lock().expect("mutex should lock") += 1;
(
StatusCode::UNAUTHORIZED,
Json(json!({
"error": {
"message": "Your refresh token has already been used to generate a new access token. Please try signing in again.",
"type": "invalid_request_error",
"param": serde_json::Value::Null,
"code": "refresh_token_reused"
}
})),
)
}
}),
);
let mut provider = sample_provider("provider-codex", "codex", 10);
provider.provider_type = "codex".to_string();
let endpoint = sample_endpoint(
"endpoint-codex-cli",
"provider-codex",
"openai:cli",
"https://chatgpt.com/backend-api/codex",
);
let mut key = sample_key(
"key-codex-oauth-lazy",
"provider-codex",
"openai:cli",
"stale-codex-access-token",
);
key.auth_type = "oauth".to_string();
key.expires_at_unix_secs = Some(1);
key.encrypted_auth_config = Some(
encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"codex","refresh_token":"used-refresh-token","email":"alice@example.com","account_id":"acct-codex-123","plan_type":"plus","expires_at":1}"#,
)
.expect("auth config ciphertext should build"),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
vec![endpoint],
vec![key],
));
let (token_url, token_handle) = start_server(token_server).await;
let oauth_refresh =
crate::provider_transport::LocalOAuthRefreshCoordinator::with_adapters_for_tests(vec![
Arc::new(
crate::provider_transport::oauth_refresh::GenericOAuthRefreshAdapter::default()
.with_token_url_for_tests("codex", format!("{token_url}/oauth/token")),
),
]);
let state = AppState::new()
.expect("state should build")
.with_data_state_for_tests(
GatewayDataState::with_provider_catalog_repository_for_tests(
provider_catalog_repository.clone(),
)
.with_encryption_key_for_tests(DEVELOPMENT_ENCRYPTION_KEY),
)
.with_oauth_refresh_coordinator_for_tests(oauth_refresh);
let transport = state
.read_provider_transport_snapshot(
"provider-codex",
"endpoint-codex-cli",
"key-codex-oauth-lazy",
)
.await
.expect("transport snapshot should load")
.expect("transport snapshot should exist");
let resolved = state
.resolve_local_oauth_request_auth(&transport)
.await
.expect("refresh-token reuse should degrade into oauth-unavailable");
assert_eq!(resolved, None);
assert_eq!(*token_hits.lock().expect("mutex should lock"), 1);
let stored_key = provider_catalog_repository
.list_keys_by_ids(&["key-codex-oauth-lazy".to_string()])
.await
.expect("keys should list")
.into_iter()
.next()
.expect("oauth key should exist");
assert!(stored_key.oauth_invalid_at_unix_secs.is_some());
assert_eq!(
stored_key.oauth_invalid_reason.as_deref(),
Some("[REFRESH_FAILED] Token 续期失败 (401): refresh_token 已被使用并轮换，请重新登录授权")
);
let oauth_snapshot = stored_key
.status_snapshot
.as_ref()
.and_then(serde_json::Value::as_object)
.and_then(|snapshot| snapshot.get("oauth"))
.and_then(serde_json::Value::as_object)
.expect("oauth status snapshot should exist");
assert_eq!(
oauth_snapshot
.get("code")
.and_then(serde_json::Value::as_str),
Some("invalid")
);
assert_eq!(
oauth_snapshot
.get("source")
.and_then(serde_json::Value::as_str),
Some("oauth_refresh")
);
assert_eq!(
oauth_snapshot
.get("requires_reauth")
.and_then(serde_json::Value::as_bool),
Some(true)
);
token_handle.abort();
}
#[tokio::test]
async fn gateway_refreshes_admin_provider_oauth_key_locally_with_trusted_admin_principal() {
let upstream_hits = Arc::new(Mutex::new(0usize));

70
apps/aether-gateway/src/tests/control/admin/pool.rs
View File

@@ -691,6 +691,76 @@ async fn gateway_handles_admin_pool_list_keys_locally_with_trusted_admin_princip
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_marks_account_blocked_pool_key_in_list_keys_response() {
let mut provider = sample_provider("provider-codex", "codex", 10).with_transport_fields(
true,
false,
true,
None,
None,
None,
None,
None,
Some(json!({
"pool_advanced": {
"enabled": true
}
})),
);
provider.provider_type = "codex".to_string();
let mut key = sample_key(
"key-codex-blocked",
"provider-codex",
"openai:cli",
"oauth-placeholder",
);
key.name = "blocked-codex".to_string();
key.auth_type = "oauth".to_string();
key.oauth_invalid_reason = Some("[ACCOUNT_BLOCK] account has been deactivated".to_string());
key.oauth_invalid_at_unix_secs = Some(1_700_000_000);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
Vec::new(),
vec![key],
));
let state = AppState::new()
.expect("gateway should build")
.with_data_state_for_tests(GatewayDataState::with_provider_catalog_reader_for_tests(
provider_catalog_repository,
));
let response = local_admin_pool_response(
&state,
http::Method::GET,
"/api/admin/pool/provider-codex/keys?page=1&page_size=50&status=all",
None,
)
.await;
assert_eq!(response.status(), StatusCode::OK);
let payload: serde_json::Value = serde_json::from_slice(
&to_bytes(response.into_body(), usize::MAX)
.await
.expect("body should read"),
)
.expect("json body should parse");
let keys = payload["keys"].as_array().expect("keys should be array");
assert_eq!(keys.len(), 1);
assert_eq!(keys[0]["account_status_code"], json!("account_disabled"));
assert_eq!(keys[0]["account_status_blocked"], json!(true));
assert_eq!(keys[0]["scheduling_status"], json!("blocked"));
assert_eq!(keys[0]["scheduling_reason"], json!("account_blocked"));
assert_eq!(keys[0]["scheduling_label"], json!("账号停用"));
assert_eq!(
keys[0]["status_snapshot"]["account"]["source"],
json!("oauth_invalid")
);
}
#[tokio::test]
async fn gateway_handles_admin_pool_list_keys_with_quota_compatibility_fields() {
let upstream_hits = Arc::new(Mutex::new(0usize));