mirror of
https://github.com/fawney19/Aether.git
synced 2026-09-02 01:10:23 +08:00
fix provider oauth endpoint reconciliation
This commit is contained in:
@@ -18,7 +18,7 @@ use crate::handlers::admin::provider::oauth::provisioning::{
|
|||||||
provider_oauth_key_proxy_value, update_existing_provider_oauth_catalog_key,
|
provider_oauth_key_proxy_value, update_existing_provider_oauth_catalog_key,
|
||||||
};
|
};
|
||||||
use crate::handlers::admin::provider::oauth::runtime::{
|
use crate::handlers::admin::provider::oauth::runtime::{
|
||||||
provider_oauth_runtime_endpoint_for_provider,
|
resolve_provider_oauth_runtime_endpoints,
|
||||||
spawn_provider_oauth_account_state_refresh_after_update,
|
spawn_provider_oauth_account_state_refresh_after_update,
|
||||||
};
|
};
|
||||||
use crate::handlers::admin::provider::oauth::state::{
|
use crate::handlers::admin::provider::oauth::state::{
|
||||||
@@ -224,11 +224,11 @@ pub(super) async fn execute_admin_provider_oauth_batch_import(
|
|||||||
});
|
});
|
||||||
};
|
};
|
||||||
|
|
||||||
let endpoints = state
|
let endpoint_resolution =
|
||||||
.list_provider_catalog_endpoints_by_provider_ids(&[provider_id.to_string()])
|
resolve_provider_oauth_runtime_endpoints(state, &provider, provider_type).await?;
|
||||||
.await?;
|
let endpoints = endpoint_resolution.endpoints;
|
||||||
let api_formats = provider_oauth_active_api_formats(&endpoints);
|
let api_formats = provider_oauth_active_api_formats(&endpoints);
|
||||||
let runtime_endpoint = provider_oauth_runtime_endpoint_for_provider(provider_type, &endpoints);
|
let runtime_endpoint = endpoint_resolution.runtime_endpoint;
|
||||||
let request_proxy = state
|
let request_proxy = state
|
||||||
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
||||||
proxy_node_id,
|
proxy_node_id,
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ use crate::handlers::admin::provider::oauth::provisioning::{
|
|||||||
provider_oauth_key_proxy_value, update_existing_provider_oauth_catalog_key,
|
provider_oauth_key_proxy_value, update_existing_provider_oauth_catalog_key,
|
||||||
};
|
};
|
||||||
use crate::handlers::admin::provider::oauth::runtime::{
|
use crate::handlers::admin::provider::oauth::runtime::{
|
||||||
provider_oauth_runtime_endpoint_for_provider,
|
resolve_provider_oauth_runtime_endpoints,
|
||||||
spawn_provider_oauth_account_state_refresh_after_update,
|
spawn_provider_oauth_account_state_refresh_after_update,
|
||||||
};
|
};
|
||||||
use crate::handlers::admin::provider::oauth::state::decode_jwt_claims;
|
use crate::handlers::admin::provider::oauth::state::decode_jwt_claims;
|
||||||
@@ -59,11 +59,11 @@ pub(super) async fn execute_admin_provider_oauth_kiro_batch_import(
|
|||||||
});
|
});
|
||||||
};
|
};
|
||||||
|
|
||||||
let endpoints = state
|
let endpoint_resolution =
|
||||||
.list_provider_catalog_endpoints_by_provider_ids(&[provider_id.to_string()])
|
resolve_provider_oauth_runtime_endpoints(state, &provider, "kiro").await?;
|
||||||
.await?;
|
let endpoints = endpoint_resolution.endpoints;
|
||||||
let api_formats = provider_oauth_active_api_formats(&endpoints);
|
let api_formats = provider_oauth_active_api_formats(&endpoints);
|
||||||
let runtime_endpoint = provider_oauth_runtime_endpoint_for_provider("kiro", &endpoints);
|
let runtime_endpoint = endpoint_resolution.runtime_endpoint;
|
||||||
let request_proxy = state
|
let request_proxy = state
|
||||||
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
||||||
proxy_node_id,
|
proxy_node_id,
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
use super::super::super::errors::build_internal_control_error_response;
|
use super::super::super::errors::build_internal_control_error_response;
|
||||||
use super::super::super::provisioning::provider_oauth_token_payload_expires_at_unix_secs;
|
use super::super::super::provisioning::provider_oauth_token_payload_expires_at_unix_secs;
|
||||||
use super::super::super::quota::codex::refresh_codex_provider_quota_locally;
|
use super::super::super::quota::codex::refresh_codex_provider_quota_locally;
|
||||||
use super::super::super::runtime::provider_oauth_runtime_endpoint_for_provider;
|
use super::super::super::runtime::resolve_provider_oauth_runtime_endpoints;
|
||||||
use super::super::super::state::{
|
use super::super::super::state::{
|
||||||
admin_provider_oauth_template, enrich_admin_provider_oauth_auth_config,
|
admin_provider_oauth_template, enrich_admin_provider_oauth_auth_config,
|
||||||
is_fixed_provider_type_for_provider_oauth, json_non_empty_string,
|
is_fixed_provider_type_for_provider_oauth, json_non_empty_string,
|
||||||
@@ -126,10 +126,10 @@ pub(super) async fn handle_admin_provider_oauth_complete_key(
|
|||||||
"该 Provider 不支持 OAuth 授权",
|
"该 Provider 不支持 OAuth 授权",
|
||||||
));
|
));
|
||||||
};
|
};
|
||||||
let endpoints = state
|
let endpoint_resolution =
|
||||||
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider_id))
|
resolve_provider_oauth_runtime_endpoints(state, &provider, &provider_type).await?;
|
||||||
.await?;
|
let endpoints = endpoint_resolution.endpoints;
|
||||||
let runtime_endpoint = provider_oauth_runtime_endpoint_for_provider(&provider_type, &endpoints);
|
let runtime_endpoint = endpoint_resolution.runtime_endpoint;
|
||||||
let request_proxy = state
|
let request_proxy = state
|
||||||
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
||||||
payload.proxy_node_id.as_deref(),
|
payload.proxy_node_id.as_deref(),
|
||||||
@@ -223,9 +223,6 @@ pub(super) async fn handle_admin_provider_oauth_complete_key(
|
|||||||
let mut account_state_recheck_attempted = false;
|
let mut account_state_recheck_attempted = false;
|
||||||
let mut account_state_recheck_error = None::<String>;
|
let mut account_state_recheck_error = None::<String>;
|
||||||
if provider_type == "codex" {
|
if provider_type == "codex" {
|
||||||
let endpoints = state
|
|
||||||
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider_id))
|
|
||||||
.await?;
|
|
||||||
if let Some(endpoint) = endpoints.into_iter().find(|endpoint| {
|
if let Some(endpoint) = endpoints.into_iter().find(|endpoint| {
|
||||||
endpoint.is_active
|
endpoint.is_active
|
||||||
&& crate::ai_serving::is_openai_responses_format(&endpoint.api_format)
|
&& crate::ai_serving::is_openai_responses_format(&endpoint.api_format)
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ use super::super::super::provisioning::{
|
|||||||
update_existing_provider_oauth_catalog_key,
|
update_existing_provider_oauth_catalog_key,
|
||||||
};
|
};
|
||||||
use super::super::super::runtime::{
|
use super::super::super::runtime::{
|
||||||
provider_oauth_runtime_endpoint_for_provider,
|
resolve_provider_oauth_runtime_endpoints,
|
||||||
spawn_provider_oauth_account_state_refresh_after_update,
|
spawn_provider_oauth_account_state_refresh_after_update,
|
||||||
};
|
};
|
||||||
use super::super::super::state::{
|
use super::super::super::state::{
|
||||||
@@ -117,10 +117,10 @@ pub(super) async fn handle_admin_provider_oauth_complete_provider(
|
|||||||
"该 Provider 不支持 OAuth 授权",
|
"该 Provider 不支持 OAuth 授权",
|
||||||
));
|
));
|
||||||
};
|
};
|
||||||
let endpoints = state
|
let endpoint_resolution =
|
||||||
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider_id))
|
resolve_provider_oauth_runtime_endpoints(state, &provider, &provider_type).await?;
|
||||||
.await?;
|
let endpoints = endpoint_resolution.endpoints;
|
||||||
let runtime_endpoint = provider_oauth_runtime_endpoint_for_provider(&provider_type, &endpoints);
|
let runtime_endpoint = endpoint_resolution.runtime_endpoint;
|
||||||
let request_proxy = state
|
let request_proxy = state
|
||||||
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
||||||
payload.proxy_node_id.as_deref(),
|
payload.proxy_node_id.as_deref(),
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
use super::session::AdminProviderOAuthDeviceAuthorizePayload;
|
use super::session::AdminProviderOAuthDeviceAuthorizePayload;
|
||||||
use crate::handlers::admin::provider::oauth::errors::build_internal_control_error_response;
|
use crate::handlers::admin::provider::oauth::errors::build_internal_control_error_response;
|
||||||
use crate::handlers::admin::provider::oauth::runtime::provider_oauth_runtime_endpoint_for_provider;
|
use crate::handlers::admin::provider::oauth::runtime::resolve_provider_oauth_runtime_endpoints;
|
||||||
use crate::handlers::admin::provider::oauth::state::{
|
use crate::handlers::admin::provider::oauth::state::{
|
||||||
build_admin_provider_oauth_backend_unavailable_response, current_unix_secs,
|
build_admin_provider_oauth_backend_unavailable_response, current_unix_secs,
|
||||||
default_kiro_device_start_url, generate_provider_oauth_nonce, json_non_empty_string,
|
default_kiro_device_start_url, generate_provider_oauth_nonce, json_non_empty_string,
|
||||||
@@ -169,10 +169,9 @@ pub(super) async fn handle_admin_provider_oauth_device_authorize(
|
|||||||
"设备授权仅支持 Kiro provider",
|
"设备授权仅支持 Kiro provider",
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
let endpoints = state
|
let endpoint_resolution =
|
||||||
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider_id))
|
resolve_provider_oauth_runtime_endpoints(state, &provider, "kiro").await?;
|
||||||
.await?;
|
let runtime_endpoint = endpoint_resolution.runtime_endpoint;
|
||||||
let runtime_endpoint = provider_oauth_runtime_endpoint_for_provider("kiro", &endpoints);
|
|
||||||
let request_proxy = state
|
let request_proxy = state
|
||||||
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
||||||
payload.proxy_node_id.as_deref(),
|
payload.proxy_node_id.as_deref(),
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ use crate::handlers::admin::provider::oauth::provisioning::{
|
|||||||
provider_oauth_active_api_formats, provider_oauth_key_proxy_value,
|
provider_oauth_active_api_formats, provider_oauth_key_proxy_value,
|
||||||
};
|
};
|
||||||
use crate::handlers::admin::provider::oauth::runtime::{
|
use crate::handlers::admin::provider::oauth::runtime::{
|
||||||
provider_oauth_runtime_endpoint_for_provider,
|
resolve_provider_oauth_runtime_endpoints,
|
||||||
spawn_provider_oauth_account_state_refresh_after_update,
|
spawn_provider_oauth_account_state_refresh_after_update,
|
||||||
};
|
};
|
||||||
use crate::handlers::admin::provider::oauth::state::{
|
use crate::handlers::admin::provider::oauth::state::{
|
||||||
@@ -322,10 +322,10 @@ pub(super) async fn handle_admin_provider_oauth_device_poll(
|
|||||||
"Provider 不存在",
|
"Provider 不存在",
|
||||||
));
|
));
|
||||||
};
|
};
|
||||||
let endpoints = state
|
let endpoint_resolution =
|
||||||
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider_id))
|
resolve_provider_oauth_runtime_endpoints(state, &provider, "kiro").await?;
|
||||||
.await?;
|
let endpoints = endpoint_resolution.endpoints;
|
||||||
let runtime_endpoint = provider_oauth_runtime_endpoint_for_provider("kiro", &endpoints);
|
let runtime_endpoint = endpoint_resolution.runtime_endpoint;
|
||||||
let request_proxy = state
|
let request_proxy = state
|
||||||
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
||||||
session.proxy_node_id.as_deref(),
|
session.proxy_node_id.as_deref(),
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ use super::super::provisioning::{
|
|||||||
update_existing_provider_oauth_catalog_key,
|
update_existing_provider_oauth_catalog_key,
|
||||||
};
|
};
|
||||||
use super::super::runtime::{
|
use super::super::runtime::{
|
||||||
provider_oauth_runtime_endpoint_for_provider,
|
resolve_provider_oauth_runtime_endpoints,
|
||||||
spawn_provider_oauth_account_state_refresh_after_update,
|
spawn_provider_oauth_account_state_refresh_after_update,
|
||||||
};
|
};
|
||||||
use super::super::state::{
|
use super::super::state::{
|
||||||
@@ -300,10 +300,10 @@ pub(super) async fn handle_admin_provider_oauth_import_refresh_token(
|
|||||||
let Some(template) = admin_provider_oauth_template(&provider_type) else {
|
let Some(template) = admin_provider_oauth_template(&provider_type) else {
|
||||||
return Ok(build_admin_provider_oauth_backend_unavailable_response());
|
return Ok(build_admin_provider_oauth_backend_unavailable_response());
|
||||||
};
|
};
|
||||||
let endpoints = state
|
let endpoint_resolution =
|
||||||
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider_id))
|
resolve_provider_oauth_runtime_endpoints(state, &provider, &provider_type).await?;
|
||||||
.await?;
|
let endpoints = endpoint_resolution.endpoints;
|
||||||
let runtime_endpoint = provider_oauth_runtime_endpoint_for_provider(&provider_type, &endpoints);
|
let runtime_endpoint = endpoint_resolution.runtime_endpoint;
|
||||||
let request_proxy = state
|
let request_proxy = state
|
||||||
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
.resolve_admin_provider_oauth_operation_proxy_snapshot(
|
||||||
proxy_node_id.as_deref(),
|
proxy_node_id.as_deref(),
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
use super::super::super::runtime::provider_oauth_runtime_endpoint_for_provider;
|
use super::super::super::runtime::resolve_provider_oauth_runtime_endpoints;
|
||||||
use super::super::super::state::is_fixed_provider_type_for_provider_oauth;
|
use super::super::super::state::is_fixed_provider_type_for_provider_oauth;
|
||||||
use super::helpers::{self, RefreshDispatch, RefreshRequestContext};
|
use super::helpers::{self, RefreshDispatch, RefreshRequestContext};
|
||||||
use super::response;
|
use super::response;
|
||||||
@@ -76,11 +76,19 @@ pub(super) async fn parse_admin_provider_oauth_refresh_request(
|
|||||||
)));
|
)));
|
||||||
}
|
}
|
||||||
|
|
||||||
let endpoints = state
|
let endpoint_resolution =
|
||||||
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider_id))
|
resolve_provider_oauth_runtime_endpoints(state, &provider, &provider_type).await?;
|
||||||
.await?;
|
let Some(endpoint) = endpoint_resolution.runtime_endpoint else {
|
||||||
let Some(endpoint) = provider_oauth_runtime_endpoint_for_provider(&provider_type, &endpoints)
|
if state
|
||||||
else {
|
.fixed_provider_template(&provider.provider_type)
|
||||||
|
.is_some()
|
||||||
|
&& !state.has_provider_catalog_data_writer()
|
||||||
|
{
|
||||||
|
return Ok(RefreshDispatch::Respond(response::control_error_response(
|
||||||
|
http::StatusCode::BAD_REQUEST,
|
||||||
|
"固定 Provider 端点缺失,且 provider catalog writer 不可用,无法自动补全端点",
|
||||||
|
)));
|
||||||
|
}
|
||||||
return Ok(RefreshDispatch::Respond(response::control_error_response(
|
return Ok(RefreshDispatch::Respond(response::control_error_response(
|
||||||
http::StatusCode::BAD_REQUEST,
|
http::StatusCode::BAD_REQUEST,
|
||||||
"找不到有效端点,无法 refresh",
|
"找不到有效端点,无法 refresh",
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ use super::quota::antigravity::refresh_antigravity_provider_quota_locally;
|
|||||||
use super::quota::chatgpt_web::refresh_chatgpt_web_provider_quota_locally;
|
use super::quota::chatgpt_web::refresh_chatgpt_web_provider_quota_locally;
|
||||||
use super::quota::codex::refresh_codex_provider_quota_locally;
|
use super::quota::codex::refresh_codex_provider_quota_locally;
|
||||||
use super::quota::kiro::refresh_kiro_provider_quota_locally;
|
use super::quota::kiro::refresh_kiro_provider_quota_locally;
|
||||||
|
use crate::handlers::admin::provider::write::provider::reconcile_admin_fixed_provider_template_endpoints;
|
||||||
use crate::handlers::admin::request::AdminAppState;
|
use crate::handlers::admin::request::AdminAppState;
|
||||||
use crate::provider_key_auth::provider_key_is_oauth_managed;
|
use crate::provider_key_auth::provider_key_is_oauth_managed;
|
||||||
use crate::task_runtime::{spawn_fire_and_forget, TASK_KEY_PROVIDER_OAUTH_ACCOUNT_REFRESH};
|
use crate::task_runtime::{spawn_fire_and_forget, TASK_KEY_PROVIDER_OAUTH_ACCOUNT_REFRESH};
|
||||||
@@ -109,6 +110,41 @@ pub(crate) fn provider_oauth_runtime_endpoint_for_provider(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub(crate) struct ProviderOAuthRuntimeEndpoints {
|
||||||
|
pub(crate) endpoints: Vec<StoredProviderCatalogEndpoint>,
|
||||||
|
pub(crate) runtime_endpoint: Option<StoredProviderCatalogEndpoint>,
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) async fn resolve_provider_oauth_runtime_endpoints(
|
||||||
|
state: &AdminAppState<'_>,
|
||||||
|
provider: &StoredProviderCatalogProvider,
|
||||||
|
provider_type: &str,
|
||||||
|
) -> Result<ProviderOAuthRuntimeEndpoints, GatewayError> {
|
||||||
|
let mut endpoints = state
|
||||||
|
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider.id))
|
||||||
|
.await?;
|
||||||
|
let mut runtime_endpoint =
|
||||||
|
provider_oauth_runtime_endpoint_for_provider(provider_type, &endpoints);
|
||||||
|
if runtime_endpoint.is_none()
|
||||||
|
&& state
|
||||||
|
.fixed_provider_template(&provider.provider_type)
|
||||||
|
.is_some()
|
||||||
|
&& state.has_provider_catalog_data_writer()
|
||||||
|
{
|
||||||
|
reconcile_admin_fixed_provider_template_endpoints(state, provider).await?;
|
||||||
|
endpoints = state
|
||||||
|
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider.id))
|
||||||
|
.await?;
|
||||||
|
runtime_endpoint = provider_oauth_runtime_endpoint_for_provider(provider_type, &endpoints);
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(ProviderOAuthRuntimeEndpoints {
|
||||||
|
endpoints,
|
||||||
|
runtime_endpoint,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
pub(crate) async fn refresh_provider_oauth_account_state_after_update(
|
pub(crate) async fn refresh_provider_oauth_account_state_after_update(
|
||||||
state: &AdminAppState<'_>,
|
state: &AdminAppState<'_>,
|
||||||
provider: &StoredProviderCatalogProvider,
|
provider: &StoredProviderCatalogProvider,
|
||||||
@@ -123,11 +159,10 @@ pub(crate) async fn refresh_provider_oauth_account_state_after_update(
|
|||||||
return Ok((false, None));
|
return Ok((false, None));
|
||||||
}
|
}
|
||||||
|
|
||||||
let endpoints = state
|
let ProviderOAuthRuntimeEndpoints {
|
||||||
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider.id))
|
runtime_endpoint, ..
|
||||||
.await?;
|
} = resolve_provider_oauth_runtime_endpoints(state, provider, &provider_type).await?;
|
||||||
let Some(endpoint) = provider_oauth_runtime_endpoint_for_provider(&provider_type, &endpoints)
|
let Some(endpoint) = runtime_endpoint else {
|
||||||
else {
|
|
||||||
return Ok((false, None));
|
return Ok((false, None));
|
||||||
};
|
};
|
||||||
let Some(key) = state
|
let Some(key) = state
|
||||||
|
|||||||
@@ -326,14 +326,14 @@ async fn refresh_imported_oauth_key_after_persist(
|
|||||||
provider: &aether_data_contracts::repository::provider_catalog::StoredProviderCatalogProvider,
|
provider: &aether_data_contracts::repository::provider_catalog::StoredProviderCatalogProvider,
|
||||||
key_id: &str,
|
key_id: &str,
|
||||||
) -> Result<(), GatewayError> {
|
) -> Result<(), GatewayError> {
|
||||||
let endpoints = state
|
|
||||||
.list_provider_catalog_endpoints_by_provider_ids(std::slice::from_ref(&provider.id))
|
|
||||||
.await?;
|
|
||||||
let Some(endpoint) =
|
let Some(endpoint) =
|
||||||
crate::handlers::admin::provider::oauth::runtime::provider_oauth_runtime_endpoint_for_provider(
|
crate::handlers::admin::provider::oauth::runtime::resolve_provider_oauth_runtime_endpoints(
|
||||||
|
state,
|
||||||
|
provider,
|
||||||
provider.provider_type.as_str(),
|
provider.provider_type.as_str(),
|
||||||
&endpoints,
|
|
||||||
)
|
)
|
||||||
|
.await?
|
||||||
|
.runtime_endpoint
|
||||||
else {
|
else {
|
||||||
return Ok(());
|
return Ok(());
|
||||||
};
|
};
|
||||||
|
|||||||
@@ -2349,7 +2349,15 @@ async fn gateway_completes_admin_provider_oauth_key_locally_with_trusted_admin_p
|
|||||||
assert_eq!(payload["has_refresh_token"], true);
|
assert_eq!(payload["has_refresh_token"], true);
|
||||||
assert_eq!(payload["expires_at"], 4_102_444_800u64);
|
assert_eq!(payload["expires_at"], 4_102_444_800u64);
|
||||||
assert_eq!(payload["email"], "alice@example.com");
|
assert_eq!(payload["email"], "alice@example.com");
|
||||||
assert_eq!(payload["account_state_recheck_attempted"], false);
|
assert_eq!(payload["account_state_recheck_attempted"], true);
|
||||||
|
let account_state_recheck_error = payload["account_state_recheck_error"]
|
||||||
|
.as_str()
|
||||||
|
.expect("account_state_recheck_error should be string when recheck is attempted");
|
||||||
|
assert!(
|
||||||
|
account_state_recheck_error == "wham/usage API 返回状态码 401"
|
||||||
|
|| account_state_recheck_error.starts_with("wham/usage 请求执行失败:"),
|
||||||
|
"unexpected account_state_recheck_error: {account_state_recheck_error}"
|
||||||
|
);
|
||||||
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
|
assert_eq!(*upstream_hits.lock().expect("mutex should lock"), 0);
|
||||||
assert_eq!(*token_hits.lock().expect("mutex should lock"), 1);
|
assert_eq!(*token_hits.lock().expect("mutex should lock"), 1);
|
||||||
|
|
||||||
@@ -4987,6 +4995,386 @@ async fn gateway_refreshes_admin_provider_oauth_key_locally_with_trusted_admin_p
|
|||||||
upstream_handle.abort();
|
upstream_handle.abort();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn gateway_manual_codex_oauth_refresh_reconciles_missing_fixed_endpoint() {
|
||||||
|
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;
|
||||||
|
Json(json!({
|
||||||
|
"access_token": "refreshed-codex-access-token",
|
||||||
|
"refresh_token": "refreshed-codex-refresh-token",
|
||||||
|
"expires_in": 3600,
|
||||||
|
"token_type": "Bearer"
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
|
||||||
|
let seen_endpoint_id = Arc::new(Mutex::new(None::<String>));
|
||||||
|
let seen_endpoint_id_clone = Arc::clone(&seen_endpoint_id);
|
||||||
|
let execution_runtime = Router::new().route(
|
||||||
|
"/v1/execute/sync",
|
||||||
|
any(move |request: Request| {
|
||||||
|
let seen_endpoint_id_inner = Arc::clone(&seen_endpoint_id_clone);
|
||||||
|
async move {
|
||||||
|
let plan: ExecutionPlan = serde_json::from_slice(
|
||||||
|
&to_bytes(request.into_body(), usize::MAX)
|
||||||
|
.await
|
||||||
|
.expect("body should read"),
|
||||||
|
)
|
||||||
|
.expect("plan should parse");
|
||||||
|
*seen_endpoint_id_inner.lock().expect("mutex should lock") =
|
||||||
|
Some(plan.endpoint_id.clone());
|
||||||
|
let result = aether_contracts::ExecutionResult {
|
||||||
|
request_id: plan.request_id,
|
||||||
|
candidate_id: None,
|
||||||
|
status_code: 200,
|
||||||
|
headers: std::collections::BTreeMap::new(),
|
||||||
|
body: Some(aether_contracts::ResponseBody {
|
||||||
|
json_body: Some(json!({
|
||||||
|
"plan_type": "plus",
|
||||||
|
"rate_limit": {
|
||||||
|
"primary_window": {
|
||||||
|
"used_percent": 12.5,
|
||||||
|
"window_minutes": 300
|
||||||
|
},
|
||||||
|
"secondary_window": {
|
||||||
|
"used_percent": 55.0,
|
||||||
|
"window_minutes": 10080
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})),
|
||||||
|
body_bytes_b64: None,
|
||||||
|
}),
|
||||||
|
telemetry: None,
|
||||||
|
error: None,
|
||||||
|
};
|
||||||
|
(StatusCode::OK, Json(result))
|
||||||
|
}
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
|
||||||
|
let mut provider = sample_provider("provider-codex-missing-endpoint", "codex", 10);
|
||||||
|
provider.provider_type = "codex".to_string();
|
||||||
|
let mut key = sample_key(
|
||||||
|
"key-codex-missing-endpoint-refresh",
|
||||||
|
"provider-codex-missing-endpoint",
|
||||||
|
"openai:responses",
|
||||||
|
"stale-codex-access-token",
|
||||||
|
);
|
||||||
|
key.auth_type = "oauth".to_string();
|
||||||
|
key.encrypted_auth_config = Some(
|
||||||
|
encrypt_python_fernet_plaintext(
|
||||||
|
DEVELOPMENT_ENCRYPTION_KEY,
|
||||||
|
r#"{"provider_type":"codex","refresh_token":"old-codex-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::new(),
|
||||||
|
vec![key],
|
||||||
|
));
|
||||||
|
|
||||||
|
let (token_url, token_handle) = start_server(token_server).await;
|
||||||
|
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).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 gateway = build_router_with_state(
|
||||||
|
build_state_with_execution_runtime_override(execution_runtime_url)
|
||||||
|
.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 (gateway_url, gateway_handle) = start_server(gateway).await;
|
||||||
|
|
||||||
|
let response = reqwest::Client::new()
|
||||||
|
.post(format!(
|
||||||
|
"{gateway_url}/api/admin/provider-oauth/keys/key-codex-missing-endpoint-refresh/refresh"
|
||||||
|
))
|
||||||
|
.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")
|
||||||
|
.send()
|
||||||
|
.await
|
||||||
|
.expect("request should succeed");
|
||||||
|
|
||||||
|
assert_eq!(response.status(), StatusCode::OK);
|
||||||
|
let payload: serde_json::Value = response.json().await.expect("json body should parse");
|
||||||
|
assert_eq!(payload["provider_type"], "codex");
|
||||||
|
assert_eq!(payload["has_refresh_token"], true);
|
||||||
|
assert_eq!(payload["account_state_recheck_attempted"], true);
|
||||||
|
assert_eq!(
|
||||||
|
payload["account_state_recheck_error"],
|
||||||
|
serde_json::Value::Null
|
||||||
|
);
|
||||||
|
assert_eq!(*token_hits.lock().expect("mutex should lock"), 1);
|
||||||
|
|
||||||
|
let endpoints = provider_catalog_repository
|
||||||
|
.list_endpoints_by_provider_ids(&["provider-codex-missing-endpoint".to_string()])
|
||||||
|
.await
|
||||||
|
.expect("endpoints should read");
|
||||||
|
let responses_endpoint = endpoints
|
||||||
|
.iter()
|
||||||
|
.find(|endpoint| endpoint.api_format == "openai:responses")
|
||||||
|
.expect("openai responses endpoint should be reconciled");
|
||||||
|
assert_eq!(
|
||||||
|
responses_endpoint.base_url,
|
||||||
|
"https://chatgpt.com/backend-api/codex"
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
*seen_endpoint_id.lock().expect("mutex should lock"),
|
||||||
|
Some(responses_endpoint.id.clone())
|
||||||
|
);
|
||||||
|
|
||||||
|
let stored_key = provider_catalog_repository
|
||||||
|
.list_keys_by_ids(&["key-codex-missing-endpoint-refresh".to_string()])
|
||||||
|
.await
|
||||||
|
.expect("keys should load")
|
||||||
|
.into_iter()
|
||||||
|
.next()
|
||||||
|
.expect("refreshed key should exist");
|
||||||
|
let decrypted_api_key = decrypt_python_fernet_ciphertext(
|
||||||
|
DEVELOPMENT_ENCRYPTION_KEY,
|
||||||
|
stored_key
|
||||||
|
.encrypted_api_key
|
||||||
|
.as_deref()
|
||||||
|
.expect("api key should exist"),
|
||||||
|
)
|
||||||
|
.expect("api key should decrypt");
|
||||||
|
assert_eq!(decrypted_api_key, "refreshed-codex-access-token");
|
||||||
|
let decrypted_auth_config = decrypt_python_fernet_ciphertext(
|
||||||
|
DEVELOPMENT_ENCRYPTION_KEY,
|
||||||
|
stored_key
|
||||||
|
.encrypted_auth_config
|
||||||
|
.as_deref()
|
||||||
|
.expect("auth config should exist"),
|
||||||
|
)
|
||||||
|
.expect("auth config should decrypt");
|
||||||
|
let auth_config: serde_json::Value =
|
||||||
|
serde_json::from_str(&decrypted_auth_config).expect("auth config should parse");
|
||||||
|
assert_eq!(auth_config["provider_type"], "codex");
|
||||||
|
assert_eq!(
|
||||||
|
auth_config["refresh_token"],
|
||||||
|
"refreshed-codex-refresh-token"
|
||||||
|
);
|
||||||
|
|
||||||
|
gateway_handle.abort();
|
||||||
|
execution_runtime_handle.abort();
|
||||||
|
token_handle.abort();
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn gateway_manual_kiro_oauth_refresh_reconciles_missing_fixed_endpoint() {
|
||||||
|
let refreshed_access_token = sample_kiro_device_access_token("kiro-refresh@example.com");
|
||||||
|
let expected_access_token = refreshed_access_token.clone();
|
||||||
|
let refreshed_refresh_token = "s".repeat(120);
|
||||||
|
let expected_refresh_token = refreshed_refresh_token.clone();
|
||||||
|
let token_hits = Arc::new(Mutex::new(0usize));
|
||||||
|
let token_hits_clone = Arc::clone(&token_hits);
|
||||||
|
let token_server = Router::new().route(
|
||||||
|
"/refreshToken",
|
||||||
|
post(move |_request: Request| {
|
||||||
|
let token_hits_inner = Arc::clone(&token_hits_clone);
|
||||||
|
let access_token_inner = refreshed_access_token.clone();
|
||||||
|
let refresh_token_inner = refreshed_refresh_token.clone();
|
||||||
|
async move {
|
||||||
|
*token_hits_inner.lock().expect("mutex should lock") += 1;
|
||||||
|
Json(json!({
|
||||||
|
"accessToken": access_token_inner,
|
||||||
|
"refreshToken": refresh_token_inner,
|
||||||
|
"expiresIn": 3600,
|
||||||
|
"profileArn": "arn:aws:kiro:profile/manual-refresh"
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
|
||||||
|
let seen_endpoint_id = Arc::new(Mutex::new(None::<String>));
|
||||||
|
let seen_endpoint_id_clone = Arc::clone(&seen_endpoint_id);
|
||||||
|
let execution_runtime = Router::new().route(
|
||||||
|
"/v1/execute/sync",
|
||||||
|
any(move |request: Request| {
|
||||||
|
let seen_endpoint_id_inner = Arc::clone(&seen_endpoint_id_clone);
|
||||||
|
async move {
|
||||||
|
let plan: ExecutionPlan = serde_json::from_slice(
|
||||||
|
&to_bytes(request.into_body(), usize::MAX)
|
||||||
|
.await
|
||||||
|
.expect("body should read"),
|
||||||
|
)
|
||||||
|
.expect("plan should parse");
|
||||||
|
*seen_endpoint_id_inner.lock().expect("mutex should lock") =
|
||||||
|
Some(plan.endpoint_id.clone());
|
||||||
|
let result = aether_contracts::ExecutionResult {
|
||||||
|
request_id: plan.request_id,
|
||||||
|
candidate_id: None,
|
||||||
|
status_code: 200,
|
||||||
|
headers: std::collections::BTreeMap::new(),
|
||||||
|
body: Some(aether_contracts::ResponseBody {
|
||||||
|
json_body: Some(json!({
|
||||||
|
"subscriptionInfo": {
|
||||||
|
"subscriptionTitle": "KIRO PRO"
|
||||||
|
},
|
||||||
|
"usageBreakdownList": [{
|
||||||
|
"currentUsageWithPrecision": 2.0,
|
||||||
|
"usageLimitWithPrecision": 10.0,
|
||||||
|
"nextDateReset": 1_900_000_000u64
|
||||||
|
}],
|
||||||
|
"desktopUserInfo": {
|
||||||
|
"email": "kiro-refresh@example.com"
|
||||||
|
}
|
||||||
|
})),
|
||||||
|
body_bytes_b64: None,
|
||||||
|
}),
|
||||||
|
telemetry: None,
|
||||||
|
error: None,
|
||||||
|
};
|
||||||
|
(StatusCode::OK, Json(result))
|
||||||
|
}
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
|
||||||
|
let mut provider = sample_provider("provider-kiro-oauth-refresh", "kiro", 10);
|
||||||
|
provider.provider_type = "kiro".to_string();
|
||||||
|
let mut key = sample_key(
|
||||||
|
"key-kiro-oauth-refresh",
|
||||||
|
"provider-kiro-oauth-refresh",
|
||||||
|
"claude:messages",
|
||||||
|
"stale-kiro-access-token",
|
||||||
|
);
|
||||||
|
key.auth_type = "oauth".to_string();
|
||||||
|
key.encrypted_auth_config = Some(
|
||||||
|
encrypt_python_fernet_plaintext(
|
||||||
|
DEVELOPMENT_ENCRYPTION_KEY,
|
||||||
|
&json!({
|
||||||
|
"provider_type": "kiro",
|
||||||
|
"auth_method": "social",
|
||||||
|
"refresh_token": "r".repeat(120),
|
||||||
|
"machine_id": "123e4567-e89b-12d3-a456-426614174000",
|
||||||
|
"kiro_version": "1.2.3",
|
||||||
|
"expires_at": 1u64
|
||||||
|
})
|
||||||
|
.to_string(),
|
||||||
|
)
|
||||||
|
.expect("auth config ciphertext should build"),
|
||||||
|
);
|
||||||
|
|
||||||
|
let provider_catalog_repository = Arc::new(InMemoryProviderCatalogReadRepository::seed(
|
||||||
|
vec![provider],
|
||||||
|
Vec::new(),
|
||||||
|
vec![key],
|
||||||
|
));
|
||||||
|
|
||||||
|
let (token_url, token_handle) = start_server(token_server).await;
|
||||||
|
let (execution_runtime_url, execution_runtime_handle) = start_server(execution_runtime).await;
|
||||||
|
let oauth_refresh =
|
||||||
|
crate::provider_transport::LocalOAuthRefreshCoordinator::with_adapters_for_tests(vec![
|
||||||
|
Arc::new(
|
||||||
|
crate::provider_transport::kiro::KiroOAuthRefreshAdapter::default()
|
||||||
|
.with_refresh_base_urls(Some(token_url), None),
|
||||||
|
)
|
||||||
|
as Arc<dyn crate::provider_transport::oauth_refresh::LocalOAuthRefreshAdapter>,
|
||||||
|
]);
|
||||||
|
let gateway = build_router_with_state(
|
||||||
|
build_state_with_execution_runtime_override(execution_runtime_url)
|
||||||
|
.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 (gateway_url, gateway_handle) = start_server(gateway).await;
|
||||||
|
|
||||||
|
let response = reqwest::Client::new()
|
||||||
|
.post(format!(
|
||||||
|
"{gateway_url}/api/admin/provider-oauth/keys/key-kiro-oauth-refresh/refresh"
|
||||||
|
))
|
||||||
|
.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")
|
||||||
|
.send()
|
||||||
|
.await
|
||||||
|
.expect("request should succeed");
|
||||||
|
|
||||||
|
assert_eq!(response.status(), StatusCode::OK);
|
||||||
|
let payload: serde_json::Value = response.json().await.expect("json body should parse");
|
||||||
|
assert_eq!(payload["provider_type"], "kiro");
|
||||||
|
assert_eq!(payload["has_refresh_token"], true);
|
||||||
|
assert_eq!(payload["account_state_recheck_attempted"], true);
|
||||||
|
assert_eq!(
|
||||||
|
payload["account_state_recheck_error"],
|
||||||
|
serde_json::Value::Null
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
*token_hits.lock().expect("mutex should lock") >= 1,
|
||||||
|
"Kiro refresh endpoint should be called"
|
||||||
|
);
|
||||||
|
|
||||||
|
let endpoints = provider_catalog_repository
|
||||||
|
.list_endpoints_by_provider_ids(&["provider-kiro-oauth-refresh".to_string()])
|
||||||
|
.await
|
||||||
|
.expect("endpoints should read");
|
||||||
|
assert_eq!(endpoints.len(), 1);
|
||||||
|
assert_eq!(endpoints[0].api_format, "claude:messages");
|
||||||
|
assert_eq!(endpoints[0].base_url, "https://q.{region}.amazonaws.com");
|
||||||
|
assert_eq!(
|
||||||
|
*seen_endpoint_id.lock().expect("mutex should lock"),
|
||||||
|
Some(endpoints[0].id.clone())
|
||||||
|
);
|
||||||
|
|
||||||
|
let stored_key = provider_catalog_repository
|
||||||
|
.list_keys_by_ids(&["key-kiro-oauth-refresh".to_string()])
|
||||||
|
.await
|
||||||
|
.expect("keys should load")
|
||||||
|
.into_iter()
|
||||||
|
.next()
|
||||||
|
.expect("refreshed key should exist");
|
||||||
|
let decrypted_api_key = decrypt_python_fernet_ciphertext(
|
||||||
|
DEVELOPMENT_ENCRYPTION_KEY,
|
||||||
|
stored_key
|
||||||
|
.encrypted_api_key
|
||||||
|
.as_deref()
|
||||||
|
.expect("api key should exist"),
|
||||||
|
)
|
||||||
|
.expect("api key should decrypt");
|
||||||
|
assert_eq!(decrypted_api_key, expected_access_token);
|
||||||
|
let decrypted_auth_config = decrypt_python_fernet_ciphertext(
|
||||||
|
DEVELOPMENT_ENCRYPTION_KEY,
|
||||||
|
stored_key
|
||||||
|
.encrypted_auth_config
|
||||||
|
.as_deref()
|
||||||
|
.expect("auth config should exist"),
|
||||||
|
)
|
||||||
|
.expect("auth config should decrypt");
|
||||||
|
let auth_config: serde_json::Value =
|
||||||
|
serde_json::from_str(&decrypted_auth_config).expect("auth config should parse");
|
||||||
|
assert_eq!(auth_config["provider_type"], "kiro");
|
||||||
|
assert_eq!(auth_config["refresh_token"], expected_refresh_token);
|
||||||
|
|
||||||
|
gateway_handle.abort();
|
||||||
|
execution_runtime_handle.abort();
|
||||||
|
token_handle.abort();
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn gateway_marks_manual_oauth_refresh_failures_as_invalid_in_pool_payload() {
|
async fn gateway_marks_manual_oauth_refresh_failures_as_invalid_in_pool_payload() {
|
||||||
let token_hits = Arc::new(Mutex::new(0usize));
|
let token_hits = Arc::new(Mutex::new(0usize));
|
||||||
|
|||||||
Reference in New Issue
Block a user