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

fix: align provider oauth refresh redirects

Browse Source
This commit is contained in:
fawney19
2026-04-27 22:08:02 +08:00
parent a7a3c9f023
commit ed17147281
6 changed files with 242 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/refresh/execution.rs
View File

@@ -21,12 +21,20 @@ pub(super) async fn execute_admin_provider_oauth_refresh(
key,
provider,
provider_type,
trace_id,
transport,
} = request;
let refreshed_entry = match state.force_local_oauth_refresh_entry(&transport).await {
Ok(Some(entry)) => Some(entry),
Ok(None) => {
tracing::warn!(
trace_id = %trace_id,
key_id = %key_id,
provider_id = %provider.id,
provider_type = %provider_type,
"gateway manual provider oauth refresh did not run"
);
return Ok(RefreshDispatch::Respond(response::control_error_response(
http::StatusCode::BAD_REQUEST,
"Token 刷新未执行，请检查授权配置",
@@ -41,6 +49,15 @@ pub(super) async fn execute_admin_provider_oauth_refresh(
Some(status_code),
Some(body_excerpt.as_str()),
);
tracing::warn!(
trace_id = %trace_id,
key_id = %key_id,
provider_id = %provider.id,
provider_type = %provider_type,
status_code,
reason = %error_reason,
"gateway manual provider oauth refresh failed"
);
if matches!(status_code, 400 | 401 | 403) {
let failure_reason = format!(
"{OAUTH_REFRESH_FAILED_PREFIX}Token 续期失败 ({status_code}): {error_reason}"
@@ -66,11 +83,27 @@ pub(super) async fn execute_admin_provider_oauth_refresh(
));
}
Err(AdminLocalOAuthRefreshError::Transport { source, .. }) => {
tracing::warn!(
trace_id = %trace_id,
key_id = %key_id,
provider_id = %provider.id,
provider_type = %provider_type,
error = %source,
"gateway manual provider oauth refresh transport failed"
);
return Ok(RefreshDispatch::Respond(
response::oauth_refresh_failed_service_unavailable_response(source.to_string()),
));
}
Err(AdminLocalOAuthRefreshError::InvalidResponse { message, .. }) => {
tracing::warn!(
trace_id = %trace_id,
key_id = %key_id,
provider_id = %provider.id,
provider_type = %provider_type,
reason = %message,
"gateway manual provider oauth refresh returned invalid response"
);
return Ok(RefreshDispatch::Respond(
response::oauth_refresh_failed_bad_request_response(&message),
));

1
apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/refresh/helpers.rs
View File

@@ -17,6 +17,7 @@ pub(super) struct RefreshRequestContext {
pub(super) key: StoredProviderCatalogKey,
pub(super) provider: StoredProviderCatalogProvider,
pub(super) provider_type: String,
pub(super) trace_id: String,
pub(super) transport: AdminGatewayProviderTransportSnapshot,
}

1
apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/refresh/request.rs
View File

@@ -101,6 +101,7 @@ pub(super) async fn parse_admin_provider_oauth_refresh_request(
key,
provider,
provider_type,
trace_id: request_context.trace_id().to_string(),
transport,
}))
}

18
apps/aether-gateway/src/handlers/admin/provider/oauth/provisioning.rs
View File

@@ -123,7 +123,14 @@ pub(crate) async fn create_provider_oauth_catalog_key(
record.circuit_breaker_by_format = Some(json!({}));
record.created_at_unix_ms = Some(now_unix_secs);
record.updated_at_unix_secs = Some(now_unix_secs);
state.create_provider_catalog_key(&record).await
let created = state.create_provider_catalog_key(&record).await?;
if let Some(key) = created.as_ref() {
let _ = state
.app()
.invalidate_local_oauth_refresh_entry(&key.id)
.await;
}
Ok(created)
}
pub(crate) async fn update_existing_provider_oauth_catalog_key(
@@ -166,7 +173,14 @@ pub(crate) async fn update_existing_provider_oauth_catalog_key(
updated.proxy = Some(proxy);
}
updated.updated_at_unix_secs = Some(now_unix_secs);
state.update_provider_catalog_key(&updated).await
let persisted = state.update_provider_catalog_key(&updated).await?;
if let Some(key) = persisted.as_ref() {
let _ = state
.app()
.invalidate_local_oauth_refresh_entry(&key.id)
.await;
}
Ok(persisted)
}
fn provider_oauth_catalog_key_api_formats(

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

@@ -9,7 +9,9 @@ use crate::provider_transport::LocalOAuthHttpExecutor;
use super::super::provider_transport;
use crate::provider_key_auth::provider_key_is_oauth_managed;
use aether_admin::provider::quota as admin_provider_quota_pure;
use aether_contracts::{ExecutionPlan, ExecutionTimeouts, RequestBody};
use aether_contracts::{
ExecutionPlan, ExecutionTimeouts, RequestBody, EXECUTION_REQUEST_FOLLOW_REDIRECTS_HEADER,
};
use aether_data_contracts::repository::provider_catalog::StoredProviderCatalogKey;
use base64::{engine::general_purpose::STANDARD, Engine as _};
use flate2::read::{DeflateDecoder, GzDecoder};
@@ -1102,6 +1104,11 @@ impl AppState {
body_ref: None,
}
};
let mut headers = request.headers.clone();
headers.insert(
EXECUTION_REQUEST_FOLLOW_REDIRECTS_HEADER.to_string(),
"true".to_string(),
);
let plan = ExecutionPlan {
request_id: request.request_id.to_string(),
candidate_id: None,
@@ -1111,7 +1118,7 @@ impl AppState {
key_id: transport.key.id.clone(),
method: request.method.as_str().to_string(),
url: request.url.clone(),
headers: request.headers.clone(),
headers,
content_type: request
.headers
.get("content-type")

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

@@ -12,7 +12,9 @@ use aether_data::repository::oauth_providers::{
};
use aether_data::repository::provider_catalog::InMemoryProviderCatalogReadRepository;
use aether_data::repository::proxy_nodes::InMemoryProxyNodeRepository;
use aether_data_contracts::repository::provider_catalog::ProviderCatalogReadRepository;
use aether_data_contracts::repository::provider_catalog::{
ProviderCatalogReadRepository, ProviderCatalogWriteRepository,
};
use axum::body::{to_bytes, Body, Bytes};
use axum::response::{IntoResponse, Response};
use axum::routing::{any, delete, get, patch, post, put};
@@ -2476,6 +2478,178 @@ async fn gateway_imports_admin_provider_oauth_refresh_token_over_active_expired_
upstream_handle.abort();
}
#[tokio::test]
async fn gateway_import_invalidate_cached_oauth_entry_before_followup_resolution() {
let token_server = Router::new().route(
"/oauth/token",
post(move |body: Bytes| async move {
let body_text = String::from_utf8(body.to_vec()).unwrap_or_default();
if body_text.contains("refresh_token=old-refresh-token") {
Json(json!({
"access_token": "cached-old-codex-access-token",
"refresh_token": "cached-old-refresh-token",
"token_type": "Bearer",
"expires_in": 1800,
"scope": "openid email profile offline_access",
"email": "alice@example.com",
"account_id": "acct-codex-123",
"plan_type": "plus",
}))
} else {
assert!(
body_text.contains("refresh_token=provider-import-refresh-token"),
"unexpected token request body: {body_text}"
);
Json(json!({
"access_token": "imported-fresh-codex-access-token",
"refresh_token": "imported-fresh-refresh-token",
"token_type": "Bearer",
"expires_in": 1800,
"scope": "openid email profile offline_access",
"email": "alice@example.com",
"account_id": "acct-codex-123",
"plan_type": "plus",
}))
}
}),
);
let mut provider = sample_provider("provider-codex", "codex", 10);
provider.provider_type = "codex".to_string();
let endpoint = sample_endpoint(
"endpoint-codex-chat",
"provider-codex",
"openai:chat",
"https://chatgpt.com/backend-api/codex",
);
let mut existing_key = sample_key(
"key-codex-import-cache-duplicate",
"provider-codex",
"openai:chat",
"stale-imported-access-token",
);
existing_key.auth_type = "oauth".to_string();
existing_key.expires_at_unix_secs = Some(1);
existing_key.oauth_invalid_at_unix_secs = Some(1_700_000_000);
existing_key.oauth_invalid_reason = Some("[OAUTH_EXPIRED] token invalidated".to_string());
existing_key.encrypted_auth_config = Some(
encrypt_python_fernet_plaintext(
DEVELOPMENT_ENCRYPTION_KEY,
r#"{"provider_type":"codex","email":"alice@example.com","account_id":"acct-codex-123","plan_type":"plus","refresh_token":"old-refresh-token","expires_at":1}"#,
)
.expect("auth config ciphertext should build"),
);
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
vec![provider],
vec![endpoint],
vec![existing_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 app_state = AppState::new()
.expect("gateway 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_provider_oauth_token_url_for_tests("codex", format!("{token_url}/oauth/token"))
.with_oauth_refresh_coordinator_for_tests(oauth_refresh);
let stale_transport = app_state
.read_provider_transport_snapshot(
"provider-codex",
"endpoint-codex-chat",
"key-codex-import-cache-duplicate",
)
.await
.expect("transport should load")
.expect("transport should exist");
let cached_entry = app_state
.force_local_oauth_refresh_entry(&stale_transport)
.await
.expect("initial refresh should succeed")
.expect("initial refresh should return cached entry");
assert_eq!(
cached_entry.auth_header_value,
"Bearer cached-old-codex-access-token"
);
let mut replaceable_key = provider_catalog_repository
.list_keys_by_ids(&["key-codex-import-cache-duplicate".to_string()])
.await
.expect("keys should load")
.into_iter()
.next()
.expect("key should exist");
replaceable_key.oauth_invalid_at_unix_secs = Some(1_700_000_000);
replaceable_key.oauth_invalid_reason = Some("[OAUTH_EXPIRED] token invalidated".to_string());
provider_catalog_repository
.update_key(&replaceable_key)
.await
.expect("key should update");
let gateway = build_router_with_state(app_state.clone());
let (gateway_url, gateway_handle) = start_server(gateway).await;
let response = reqwest::Client::new()
.post(format!(
"{gateway_url}/api/admin/provider-oauth/providers/provider-codex/import-refresh-token"
))
.header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
.header(TRUSTED_ADMIN_USER_ID_HEADER, "admin-user-123")
.header(TRUSTED_ADMIN_USER_ROLE_HEADER, "admin")
.header(TRUSTED_ADMIN_SESSION_ID_HEADER, "session-123")
.json(&json!({
"refresh_token": "provider-import-refresh-token",
"name": "should-not-override-cache-duplicate-name"
}))
.send()
.await
.expect("request should succeed");
let status = response.status();
let payload: serde_json::Value = response.json().await.expect("json body should parse");
assert_eq!(status, StatusCode::OK, "payload={payload}");
assert_eq!(payload["key_id"], "key-codex-import-cache-duplicate");
assert_eq!(payload["replaced"], true);
let fresh_transport = app_state
.read_provider_transport_snapshot(
"provider-codex",
"endpoint-codex-chat",
"key-codex-import-cache-duplicate",
)
.await
.expect("transport should load")
.expect("transport should exist");
let resolved = app_state
.resolve_local_oauth_request_auth(&fresh_transport)
.await
.expect("oauth auth should resolve")
.expect("oauth auth should exist");
match resolved {
crate::provider_transport::LocalResolvedOAuthRequestAuth::Header { value, .. } => {
assert_eq!(value, "Bearer imported-fresh-codex-access-token");
}
crate::provider_transport::LocalResolvedOAuthRequestAuth::Kiro(_) => {
panic!("codex should resolve to header auth")
}
}
gateway_handle.abort();
token_handle.abort();
}
#[tokio::test]
async fn gateway_rejects_kiro_single_refresh_token_import_with_clear_error() {
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
@@ -4382,6 +4556,13 @@ async fn gateway_refreshes_admin_provider_oauth_key_locally_via_execution_runtim
.and_then(|proxy| proxy.node_id.as_deref()),
Some("proxy-node-provider")
);
assert_eq!(
refresh_plan
.headers
.get(EXECUTION_REQUEST_FOLLOW_REDIRECTS_HEADER)
.map(String::as_str),
Some("true")
);
gateway_handle.abort();
execution_runtime_handle.abort();