From 704b652f8614783804ff1aa4eb4f0af6f11b914b Mon Sep 17 00:00:00 2001 From: Yanshan Chen <16632048+chenyanshan@users.noreply.github.com> Date: Fri, 11 Sep 2026 11:32:55 +0800 Subject: [PATCH 1/4] feat: add managed account proxies and fix non-streaming WebSocket requests --- README.md | 2 +- backend/apps/gateway/src/bootstrap.rs | 2 +- backend/apps/gateway/tests/architecture.rs | 1 + backend/crates/gateway-admin/src/lib.rs | 18 +- .../gateway-admin/src/model/accounts.rs | 4 +- backend/crates/gateway-admin/src/model/mod.rs | 1 + .../src/model/provider_credentials.rs | 25 +- .../crates/gateway-admin/src/model/proxies.rs | 81 ++++ backend/crates/gateway-admin/src/ports/mod.rs | 1 + .../crates/gateway-admin/src/ports/proxy.rs | 45 ++ .../crates/gateway-admin/src/ports/store.rs | 8 + .../crates/gateway-admin/src/use_case/mod.rs | 50 +- .../gateway-admin/src/use_case/openai.rs | 13 + .../gateway-admin/src/use_case/proxies.rs | 151 ++++++ .../crates/gateway-admin/src/use_case/xai.rs | 13 + .../tests/model/provider_credentials.rs | 16 +- .../gateway-admin/tests/use_case/mod.rs | 4 +- .../gateway-admin/tests/use_case/openai.rs | 5 + .../gateway-admin/tests/use_case/proxies.rs | 51 ++ .../gateway-admin/tests/use_case/xai.rs | 3 + .../src/admin/accounts/credentials.rs | 17 +- .../gateway-api/src/admin/accounts/wire.rs | 30 +- backend/crates/gateway-api/src/admin/mod.rs | 2 + .../crates/gateway-api/src/admin/proxies.rs | 301 ++++++++++++ backend/crates/gateway-api/tests/admin/mod.rs | 10 +- .../crates/gateway-api/tests/admin/proxies.rs | 280 +++++++++++ .../crates/gateway-api/tests/architecture.rs | 2 + backend/crates/gateway-host/Cargo.toml | 2 +- backend/crates/gateway-host/src/lib.rs | 6 + .../crates/gateway-host/src/proxy_probe.rs | 88 ++++ backend/crates/gateway-host/tests/main.rs | 1 + .../crates/gateway-host/tests/proxy_probe.rs | 62 +++ backend/crates/gateway-store/src/bundle.rs | 1 + .../crates/gateway-store/src/postgres/mod.rs | 2 + .../provider_accounts/admin_adapter.rs | 13 + .../postgres/provider_accounts/repository.rs | 72 ++- .../src/postgres/provider_accounts/rows.rs | 3 +- .../gateway-store/src/postgres/proxies.rs | 444 ++++++++++++++++++ .../gateway-store/tests/postgres/mod.rs | 2 + .../tests/postgres/provider_accounts/mod.rs | 114 ++++- .../gateway-store/tests/postgres/proxies.rs | 308 ++++++++++++ backend/crates/providers/openai/src/admin.rs | 7 +- .../providers/openai/src/credential/admin.rs | 20 +- .../providers/openai/src/transport/client.rs | 6 + .../crates/providers/openai/tests/admin.rs | 2 + .../openai/tests/credential/admin.rs | 60 ++- .../openai/tests/transport/websocket/mod.rs | 68 +++ backend/crates/providers/xai/src/admin.rs | 7 +- .../providers/xai/src/credential/import.rs | 15 +- backend/crates/providers/xai/tests/admin.rs | 1 + .../providers/xai/tests/credential/import.rs | 39 ++ backend/migrations/.frozen-sha256 | 1 + .../0005_managed_outbound_proxies.sql | 26 + docs/api.md | 49 +- docs/architecture.md | 11 +- frontend/src/api/index.ts | 3 +- frontend/src/api/modules/accounts.ts | 4 + frontend/src/api/modules/proxies.ts | 51 ++ frontend/src/composables/useProxyCatalog.ts | 33 ++ frontend/src/layout/components/AppSidebar.vue | 2 + frontend/src/router/routes.ts | 5 + .../components/AccountBatchEditModal.vue | 4 +- .../AccountCreateModal/AccountSetupFields.vue | 2 +- .../components/AccountCreateModal/index.vue | 2 +- .../components/AccountCreateModal/model.ts | 18 +- .../accounts/components/AccountEditModal.vue | 5 +- .../accounts/components/AccountProxyField.vue | 71 +-- .../components/AccountSettingsFields.vue | 5 +- .../composables/useAccountBatchEditor.ts | 14 +- .../accounts/composables/useAccountEditor.ts | 14 +- .../composables/useAccountOnboarding.ts | 28 +- frontend/src/views/accounts/index.vue | 8 +- frontend/src/views/proxies/index.vue | 275 +++++++++++ 73 files changed, 2955 insertions(+), 155 deletions(-) create mode 100644 backend/crates/gateway-admin/src/model/proxies.rs create mode 100644 backend/crates/gateway-admin/src/ports/proxy.rs create mode 100644 backend/crates/gateway-admin/src/use_case/proxies.rs create mode 100644 backend/crates/gateway-admin/tests/use_case/proxies.rs create mode 100644 backend/crates/gateway-api/src/admin/proxies.rs create mode 100644 backend/crates/gateway-api/tests/admin/proxies.rs create mode 100644 backend/crates/gateway-host/src/proxy_probe.rs create mode 100644 backend/crates/gateway-host/tests/proxy_probe.rs create mode 100644 backend/crates/gateway-store/src/postgres/proxies.rs create mode 100644 backend/crates/gateway-store/tests/postgres/proxies.rs create mode 100644 backend/migrations/0005_managed_outbound_proxies.sql create mode 100644 frontend/src/api/modules/proxies.ts create mode 100644 frontend/src/composables/useProxyCatalog.ts create mode 100644 frontend/src/views/proxies/index.vue diff --git a/README.md b/README.md index ae59aafc0..8add28386 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ ## 功能 -- **账号管理**：OAuth 授权、账号导入、凭据刷新与额度查看；支持独立出站代理及 sub2api OpenAI 账号代理导入。 +- **账号管理**：OAuth 授权、账号导入、凭据刷新与额度查看；代理管理支持独立保存、连接测试和账号绑定，以及 sub2api OpenAI 账号代理导入。 - **访问控制**：通过账号分组分配密钥权限，设置日/周 USD 限额、并发和请求频率。 - **客户端接入**：支持 Codex CLI、桌面端与 Responses API 客户端，提供 HTTP 流式响应和 WebSocket。 - **请求诊断**：查看请求记录、用量、费用、延迟和错误详情。 diff --git a/backend/apps/gateway/src/bootstrap.rs b/backend/apps/gateway/src/bootstrap.rs index 582fe4c52..c74b3131b 100644 --- a/backend/apps/gateway/src/bootstrap.rs +++ b/backend/apps/gateway/src/bootstrap.rs @@ -78,7 +78,7 @@ pub async fn run() -> Result<(), BootstrapError> { store.admin_ports(), vec![openai.admin_provider(), xai.admin_provider()], core.snapshot_control(), - core.account_probe(), + (core.account_probe(), host.proxy_probe()), host.client_distribution_resolver(), host.system_operations(), ) diff --git a/backend/apps/gateway/tests/architecture.rs b/backend/apps/gateway/tests/architecture.rs index 356a2af43..bba1cda1b 100644 --- a/backend/apps/gateway/tests/architecture.rs +++ b/backend/apps/gateway/tests/architecture.rs @@ -182,6 +182,7 @@ const ADAPTER_PUBLIC_MODULES: &[(&str, &[&str])] = &[ &[ "client_distribution", "config", + "proxy_probe", "serve", "system_update", "workers", diff --git a/backend/crates/gateway-admin/src/lib.rs b/backend/crates/gateway-admin/src/lib.rs index ccf107ca7..b5d6ea3df 100644 --- a/backend/crates/gateway-admin/src/lib.rs +++ b/backend/crates/gateway-admin/src/lib.rs @@ -25,7 +25,7 @@ pub use use_case::{ account_groups::AccountGroupService, accounts::AccountsService, auth::AuthService, backup::BackupService, client_distribution::ClientDistributionService, client_keys::ClientKeyService, observability::ObservabilityService, openai::OpenAiService, - settings::SettingsService, system::SystemService, xai::XaiService, + proxies::ProxiesService, settings::SettingsService, system::SystemService, xai::XaiService, }; use model::{AdminError, AdminErrorKind}; @@ -149,6 +149,7 @@ pub enum AdminConfigError { /// 字段全部私有；调用方经 accessor 直接调用能力，不需要命名内部 `use_case` 模块。 #[derive(Clone)] pub struct AdminServices { + proxies: Arc, auth: Arc, accounts: Arc, account_groups: Arc, @@ -163,6 +164,11 @@ pub struct AdminServices { } impl AdminServices { + #[must_use] + pub fn proxies(&self) -> &dyn ProxiesService { + self.proxies.as_ref() + } + #[must_use] pub fn auth(&self) -> &dyn AuthService { self.auth.as_ref() @@ -247,10 +253,11 @@ pub async fn initialize( store: AdminStorePorts, providers: Vec>, snapshot: Arc, - probe: Arc, + probes: (Arc, Arc), client_distribution: Arc, system: Arc, ) -> Result { + let (probe, proxy_probe) = probes; config .resolve_and_validate(Path::new(".")) .map_err(|error| AdminError::invalid(error.to_string()))?; @@ -290,6 +297,11 @@ pub async fn initialize( backup_ports.object_store(), ); let services = AdminServices { + proxies: Arc::new(use_case::proxies::DefaultProxiesService::new( + store.proxies(), + proxy_probe, + snapshot.clone(), + )), auth, accounts, account_groups: Arc::new(DefaultAccountGroupService::new( @@ -316,11 +328,13 @@ pub async fn initialize( openai: Arc::new(DefaultOpenAiService::new( openai, store.accounts(), + store.proxies(), snapshot.clone(), )), xai: Arc::new(DefaultXaiService::new( xai, store.accounts(), + store.proxies(), snapshot.clone(), )), backups, diff --git a/backend/crates/gateway-admin/src/model/accounts.rs b/backend/crates/gateway-admin/src/model/accounts.rs index 0e287609a..eb682b6d9 100644 --- a/backend/crates/gateway-admin/src/model/accounts.rs +++ b/backend/crates/gateway-admin/src/model/accounts.rs @@ -221,7 +221,7 @@ pub struct UpdateAccount { pub concurrency_limit: Option, pub weight: AccountWeight, pub group_ids: Vec, - pub outbound_proxy: Option>, + pub outbound_proxy: Option, } /// 账号更新结果。 @@ -239,7 +239,7 @@ pub struct BatchUpdateAccounts { pub concurrency_limit: Option, pub weight: AccountWeight, pub group_ids: Vec, - pub outbound_proxy: Option>, + pub outbound_proxy: Option, } /// 批量账号更新结果。 diff --git a/backend/crates/gateway-admin/src/model/mod.rs b/backend/crates/gateway-admin/src/model/mod.rs index c2fda018b..6fb10719f 100644 --- a/backend/crates/gateway-admin/src/model/mod.rs +++ b/backend/crates/gateway-admin/src/model/mod.rs @@ -10,6 +10,7 @@ pub mod client_distribution; pub mod client_keys; pub mod observability; pub mod provider_credentials; +pub mod proxies; pub mod settings; pub mod system; diff --git a/backend/crates/gateway-admin/src/model/provider_credentials.rs b/backend/crates/gateway-admin/src/model/provider_credentials.rs index 255a95fd6..8c080c54f 100644 --- a/backend/crates/gateway-admin/src/model/provider_credentials.rs +++ b/backend/crates/gateway-admin/src/model/provider_credentials.rs @@ -108,6 +108,7 @@ pub struct CredentialDetails { /// Provider 正式文档批量导入命令。 pub struct ImportCredentials { + pub outbound_proxy_id: Option, pub settings: Option, pub context: MutationContext, pub document: ProviderDocument, @@ -132,6 +133,7 @@ pub struct CredentialImportResult { /// Provider 解析导入文档时只接收不透明文档，不接触 revision 或审计上下文。 pub struct PrepareCredentialImport { + pub default_outbound_proxy: Option, pub document: ProviderDocument, } @@ -175,6 +177,7 @@ pub struct PreparedCredentialImport { /// Admin 交给 Store 的导入事务命令。 #[derive(Debug, Clone, PartialEq)] pub struct CredentialImportCommit { + pub outbound_proxy: Option, pub settings: Option, pub prepared: PreparedCredentialImport, } @@ -255,6 +258,7 @@ pub struct PendingAuthorizationMutation { target: AuthorizationMutationTarget, owner_binding: AuthorizationOwnerBinding, outbound_proxy: Option, + outbound_proxy_id: Option, } impl PendingAuthorizationMutation { @@ -269,6 +273,7 @@ impl PendingAuthorizationMutation { target, owner_binding, outbound_proxy: None, + outbound_proxy_id: None, } } @@ -299,6 +304,16 @@ impl PendingAuthorizationMutation { pub fn outbound_proxy(&self) -> Option<&gateway_core::account::OutboundProxy> { self.outbound_proxy.as_ref() } + + #[must_use] + pub fn with_outbound_proxy_id(mut self, id: Option) -> Self { + self.outbound_proxy_id = id; + self + } + + pub fn outbound_proxy_id(&self) -> Option<&str> { + self.outbound_proxy_id.as_deref() + } } impl PendingAuthorizationMutation { @@ -338,6 +353,9 @@ impl PendingAuthorizationMutation { Value::String(proxy.expose_url().to_owned()), ); } + if let Some(id) = &self.outbound_proxy_id { + document.insert("outbound_proxy_id".to_owned(), Value::String(id.clone())); + } document } @@ -382,7 +400,8 @@ impl PendingAuthorizationMutation { started_request_id: document.started_request_id, }, ) - .with_outbound_proxy(proxy)) + .with_outbound_proxy(proxy) + .with_outbound_proxy_id(document.outbound_proxy_id)) } } @@ -395,6 +414,8 @@ struct StoredAuthorizationMutationV1 { started_request_id: String, #[serde(default)] outbound_proxy_url: Option, + #[serde(default)] + outbound_proxy_id: Option, } #[derive(Deserialize)] @@ -418,7 +439,7 @@ pub struct StartAuthorization { pub context: MutationContext, pub name: String, pub reauthorization: Option, - pub outbound_proxy: Option, + pub outbound_proxy: Option, } /// OAuth 流程启动结果。 diff --git a/backend/crates/gateway-admin/src/model/proxies.rs b/backend/crates/gateway-admin/src/model/proxies.rs new file mode 100644 index 000000000..f5c11eaf0 --- /dev/null +++ b/backend/crates/gateway-admin/src/model/proxies.rs @@ -0,0 +1,81 @@ +//! Reusable account egress configurations and redacted connectivity observations. + +use chrono::{DateTime, Utc}; +use gateway_core::account::OutboundProxy; + +use super::{PageSize, Revision}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum AccountProxySelection { + Direct, + Url(OutboundProxy), + Saved(String), +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ImportProxyBinding { + pub id: String, + pub proxy: OutboundProxy, +} + +#[derive(Debug, Clone)] +pub struct ProxyListQuery { + pub page: u32, + pub page_size: PageSize, + pub search: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProxyAccountRef { + pub id: String, + pub name: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ProxyTestResult { + pub success: bool, + pub latency_ms: u64, + pub exit_ip: Option, + pub message: String, +} + +#[derive(Debug, Clone)] +pub struct ProxyRecord { + pub id: String, + pub name: String, + pub proxy: OutboundProxy, + pub revision: Revision, + pub accounts: Vec, + pub last_test_at: Option>, + pub last_test: Option, + pub created_at: DateTime, + pub updated_at: DateTime, +} + +#[derive(Debug, Clone)] +pub struct ProxyPage { + pub items: Vec, + pub total: u64, + pub page: u32, + pub page_size: u16, +} + +#[derive(Debug, Clone)] +pub struct NewProxy { + pub name: String, + pub proxy: OutboundProxy, +} + +#[derive(Debug, Clone)] +pub struct UpdateProxy { + pub id: String, + pub revision: Revision, + pub name: String, + pub proxy: Option, +} + +#[derive(Debug, Clone)] +pub struct ProxyMutation { + pub config_revision: Revision, + pub record: ProxyRecord, +} diff --git a/backend/crates/gateway-admin/src/ports/mod.rs b/backend/crates/gateway-admin/src/ports/mod.rs index 2b3101dde..fca5cbf02 100644 --- a/backend/crates/gateway-admin/src/ports/mod.rs +++ b/backend/crates/gateway-admin/src/ports/mod.rs @@ -3,5 +3,6 @@ pub mod backup; pub mod client_distribution; pub mod provider; +pub mod proxy; pub mod store; pub mod system; diff --git a/backend/crates/gateway-admin/src/ports/proxy.rs b/backend/crates/gateway-admin/src/ports/proxy.rs new file mode 100644 index 000000000..5e7ecba57 --- /dev/null +++ b/backend/crates/gateway-admin/src/ports/proxy.rs @@ -0,0 +1,45 @@ +use async_trait::async_trait; +use gateway_core::account::OutboundProxy; + +use super::store::AdminStoreResult; +use crate::model::{ + MutationContext, Revision, + proxies::{ + NewProxy, ProxyListQuery, ProxyMutation, ProxyPage, ProxyRecord, ProxyTestResult, + UpdateProxy, + }, +}; + +#[async_trait] +pub trait ProxyStore: Send + Sync { + async fn list(&self, query: ProxyListQuery) -> AdminStoreResult; + async fn get(&self, id: &str) -> AdminStoreResult; + async fn create( + &self, + command: NewProxy, + context: &MutationContext, + ) -> AdminStoreResult; + async fn update( + &self, + command: UpdateProxy, + context: &MutationContext, + ) -> AdminStoreResult; + async fn delete( + &self, + id: &str, + revision: Revision, + context: &MutationContext, + ) -> AdminStoreResult; + async fn record_test( + &self, + id: &str, + revision: Revision, + result: ProxyTestResult, + context: &MutationContext, + ) -> AdminStoreResult; +} + +#[async_trait] +pub trait ProxyProbe: Send + Sync { + async fn test(&self, proxy: &OutboundProxy) -> ProxyTestResult; +} diff --git a/backend/crates/gateway-admin/src/ports/store.rs b/backend/crates/gateway-admin/src/ports/store.rs index 181cfe95f..94f169957 100644 --- a/backend/crates/gateway-admin/src/ports/store.rs +++ b/backend/crates/gateway-admin/src/ports/store.rs @@ -382,6 +382,7 @@ pub struct AdminAccountStorePorts { accounts: Arc, runtime: Arc, groups: Arc, + proxies: Arc, } impl AdminAccountStorePorts { @@ -390,11 +391,13 @@ impl AdminAccountStorePorts { accounts: Arc, runtime: Arc, groups: Arc, + proxies: Arc, ) -> Self { Self { accounts, runtime, groups, + proxies, } } } @@ -447,6 +450,11 @@ impl AdminStorePorts { self.accounts.groups.clone() } + #[must_use] + pub fn proxies(&self) -> Arc { + self.accounts.proxies.clone() + } + #[must_use] pub fn auth(&self) -> Arc { self.auth.clone() diff --git a/backend/crates/gateway-admin/src/use_case/mod.rs b/backend/crates/gateway-admin/src/use_case/mod.rs index f2395b5f1..a200f1152 100644 --- a/backend/crates/gateway-admin/src/use_case/mod.rs +++ b/backend/crates/gateway-admin/src/use_case/mod.rs @@ -8,6 +8,7 @@ pub mod client_distribution; pub mod client_keys; pub mod observability; pub mod openai; +pub mod proxies; pub mod settings; pub mod system; pub mod xai; @@ -109,13 +110,33 @@ async fn required_credential( .ok_or_else(|| AdminError::not_found("Provider 凭据不存在")) } +async fn import_proxy_binding( + proxies: &dyn crate::ports::proxy::ProxyStore, + id: Option<&str>, +) -> Result, AdminError> { + let Some(id) = id else { return Ok(None) }; + let record = proxies + .get(id) + .await + .map_err(|error| map_store_error(error, "proxy"))?; + if !record.last_test.is_some_and(|test| test.success) { + return Err(AdminError::conflict("请先测试代理连接")); + } + Ok(Some(crate::model::proxies::ImportProxyBinding { + id: record.id, + proxy: record.proxy, + })) +} + async fn pending_authorization( accounts: &dyn AccountStore, + proxies: &dyn crate::ports::proxy::ProxyStore, provider_kind: &ProviderKind, command: &StartAuthorization, resource: &'static str, ) -> Result { - let mut proxy = command.outbound_proxy.clone(); + let proxy; + let mut proxy_id = None; let target = match &command.reauthorization { Some(account_id) => { proxy = required_credential(accounts, provider_kind, account_id, resource) @@ -126,9 +147,27 @@ async fn pending_authorization( account_id: account_id.clone(), } } - None => AuthorizationMutationTarget::Create { - name: command.name.clone(), - }, + None => { + use crate::model::proxies::AccountProxySelection; + proxy = match &command.outbound_proxy { + Some(AccountProxySelection::Url(value)) => Some(value.clone()), + Some(AccountProxySelection::Saved(id)) => { + let record = proxies + .get(id) + .await + .map_err(|error| map_store_error(error, "proxy"))?; + if !record.last_test.is_some_and(|test| test.success) { + return Err(AdminError::conflict("请先测试代理连接")); + } + proxy_id = Some(record.id); + Some(record.proxy) + } + Some(AccountProxySelection::Direct) | None => None, + }; + AuthorizationMutationTarget::Create { + name: command.name.clone(), + } + } }; Ok(PendingAuthorizationMutation::new( provider_kind.clone(), @@ -137,7 +176,8 @@ async fn pending_authorization( &command.context, ), ) - .with_outbound_proxy(proxy)) + .with_outbound_proxy(proxy) + .with_outbound_proxy_id(proxy_id)) } fn validate_prepared_import( diff --git a/backend/crates/gateway-admin/src/use_case/openai.rs b/backend/crates/gateway-admin/src/use_case/openai.rs index 09a602780..5923f2d4f 100644 --- a/backend/crates/gateway-admin/src/use_case/openai.rs +++ b/backend/crates/gateway-admin/src/use_case/openai.rs @@ -58,6 +58,7 @@ pub trait OpenAiService: Send + Sync { pub(crate) struct DefaultOpenAiService { provider: Arc, accounts: Arc, + proxies: Arc, snapshot: Arc, } @@ -66,11 +67,13 @@ impl DefaultOpenAiService { pub(crate) fn new( provider: Arc, accounts: Arc, + proxies: Arc, snapshot: Arc, ) -> Self { Self { provider, accounts, + proxies, snapshot, } } @@ -128,9 +131,17 @@ impl OpenAiService for DefaultOpenAiService { command: ImportCredentials, ) -> Result { let context = command.context; + let outbound_proxy = super::import_proxy_binding( + self.proxies.as_ref(), + command.outbound_proxy_id.as_deref(), + ) + .await?; let prepared = self .provider .prepare_import(PrepareCredentialImport { + default_outbound_proxy: outbound_proxy + .as_ref() + .map(|binding| binding.proxy.clone()), document: command.document, }) .await @@ -144,6 +155,7 @@ impl OpenAiService for DefaultOpenAiService { .accounts .commit_credential_import( CredentialImportCommit { + outbound_proxy, prepared, settings: command.settings, }, @@ -166,6 +178,7 @@ impl OpenAiService for DefaultOpenAiService { ) -> Result { let pending = pending_authorization( self.accounts.as_ref(), + self.proxies.as_ref(), self.provider.provider_kind(), &command, "OpenAI credential", diff --git a/backend/crates/gateway-admin/src/use_case/proxies.rs b/backend/crates/gateway-admin/src/use_case/proxies.rs new file mode 100644 index 000000000..a119c05aa --- /dev/null +++ b/backend/crates/gateway-admin/src/use_case/proxies.rs @@ -0,0 +1,151 @@ +use std::sync::Arc; + +use async_trait::async_trait; +use gateway_core::runtime::SnapshotControl; +use tokio::sync::Semaphore; + +use super::{map_store_error, publish_committed}; +use crate::{ + model::{AdminError, AdminErrorKind, MutationContext, Revision, proxies::*}, + ports::proxy::{ProxyProbe, ProxyStore}, +}; + +#[async_trait] +pub trait ProxiesService: Send + Sync { + async fn list(&self, query: ProxyListQuery) -> Result; + async fn create( + &self, + command: NewProxy, + context: &MutationContext, + ) -> Result; + async fn update( + &self, + command: UpdateProxy, + context: &MutationContext, + ) -> Result; + async fn delete( + &self, + id: &str, + revision: Revision, + context: &MutationContext, + ) -> Result; + async fn test( + &self, + id: &str, + revision: Revision, + context: &MutationContext, + ) -> Result; +} + +pub(crate) struct DefaultProxiesService { + store: Arc, + probe: Arc, + snapshot: Arc, + test_slots: Semaphore, +} + +impl DefaultProxiesService { + pub(crate) fn new( + store: Arc, + probe: Arc, + snapshot: Arc, + ) -> Self { + Self { + store, + probe, + snapshot, + test_slots: Semaphore::new(4), + } + } +} + +#[async_trait] +impl ProxiesService for DefaultProxiesService { + async fn list(&self, query: ProxyListQuery) -> Result { + if query.page == 0 || query.search.len() > 256 || query.search.chars().any(char::is_control) + { + return Err(AdminError::invalid("代理查询参数不合法")); + } + self.store + .list(query) + .await + .map_err(|error| map_store_error(error, "proxy")) + } + + async fn create( + &self, + mut command: NewProxy, + context: &MutationContext, + ) -> Result { + command.name = validate_name(&command.name)?; + let result = self + .store + .create(command, context) + .await + .map_err(|error| map_store_error(error, "proxy"))?; + publish_committed(self.snapshot.as_ref(), result.config_revision).await?; + Ok(result) + } + + async fn update( + &self, + mut command: UpdateProxy, + context: &MutationContext, + ) -> Result { + command.name = validate_name(&command.name)?; + let result = self + .store + .update(command, context) + .await + .map_err(|error| map_store_error(error, "proxy"))?; + publish_committed(self.snapshot.as_ref(), result.config_revision).await?; + Ok(result) + } + + async fn delete( + &self, + id: &str, + revision: Revision, + context: &MutationContext, + ) -> Result { + let result = self + .store + .delete(id, revision, context) + .await + .map_err(|error| map_store_error(error, "proxy"))?; + publish_committed(self.snapshot.as_ref(), result).await?; + Ok(result) + } + + async fn test( + &self, + id: &str, + revision: Revision, + context: &MutationContext, + ) -> Result { + let _permit = self.test_slots.try_acquire().map_err(|_| { + AdminError::new(AdminErrorKind::RateLimited, "代理测试繁忙，请稍后重试") + })?; + let record = self + .store + .get(id) + .await + .map_err(|error| map_store_error(error, "proxy"))?; + if record.revision != revision { + return Err(AdminError::conflict("代理已被修改，请刷新后重新测试")); + } + let result = self.probe.test(&record.proxy).await; + self.store + .record_test(id, revision, result, context) + .await + .map_err(|error| map_store_error(error, "proxy")) + } +} + +fn validate_name(value: &str) -> Result { + let value = value.trim(); + if value.is_empty() || value.chars().count() > 100 || value.chars().any(char::is_control) { + return Err(AdminError::invalid("代理名称需要 1 至 100 个字符")); + } + Ok(value.to_owned()) +} diff --git a/backend/crates/gateway-admin/src/use_case/xai.rs b/backend/crates/gateway-admin/src/use_case/xai.rs index f7faf523f..eaf3e5711 100644 --- a/backend/crates/gateway-admin/src/use_case/xai.rs +++ b/backend/crates/gateway-admin/src/use_case/xai.rs @@ -49,6 +49,7 @@ pub trait XaiService: Send + Sync { pub(crate) struct DefaultXaiService { provider: Arc, accounts: Arc, + proxies: Arc, snapshot: Arc, } @@ -57,11 +58,13 @@ impl DefaultXaiService { pub(crate) fn new( provider: Arc, accounts: Arc, + proxies: Arc, snapshot: Arc, ) -> Self { Self { provider, accounts, + proxies, snapshot, } } @@ -81,9 +84,17 @@ impl XaiService for DefaultXaiService { command: ImportCredentials, ) -> Result { let context = command.context; + let outbound_proxy = super::import_proxy_binding( + self.proxies.as_ref(), + command.outbound_proxy_id.as_deref(), + ) + .await?; let prepared = self .provider .prepare_import(PrepareCredentialImport { + default_outbound_proxy: outbound_proxy + .as_ref() + .map(|binding| binding.proxy.clone()), document: command.document, }) .await @@ -97,6 +108,7 @@ impl XaiService for DefaultXaiService { .accounts .commit_credential_import( CredentialImportCommit { + outbound_proxy, prepared, settings: command.settings, }, @@ -125,6 +137,7 @@ impl XaiService for DefaultXaiService { ) -> Result { let pending = pending_authorization( self.accounts.as_ref(), + self.proxies.as_ref(), self.provider.provider_kind(), &command, "xAI credential", diff --git a/backend/crates/gateway-admin/tests/model/provider_credentials.rs b/backend/crates/gateway-admin/tests/model/provider_credentials.rs index 0f9122891..3e54df85f 100644 --- a/backend/crates/gateway-admin/tests/model/provider_credentials.rs +++ b/backend/crates/gateway-admin/tests/model/provider_credentials.rs @@ -97,7 +97,21 @@ fn pending_mutation_v1_should_round_trip_all_targets_and_owners() { PendingAuthorizationMutation::from_storage_v1(Value::Object( expected.to_storage_v1() )), - Ok(expected), + Ok(expected.clone()), + ); + let with_proxy = expected + .with_outbound_proxy(Some( + gateway_core::account::OutboundProxy::parse( + "http://user:secret@proxy.example:8080", + ) + .unwrap(), + )) + .with_outbound_proxy_id(Some("proxy_saved".to_owned())); + assert_eq!( + PendingAuthorizationMutation::from_storage_v1(Value::Object( + with_proxy.to_storage_v1() + )), + Ok(with_proxy) ); } } diff --git a/backend/crates/gateway-admin/tests/use_case/mod.rs b/backend/crates/gateway-admin/tests/use_case/mod.rs index 18172dc64..2bb5bbb84 100644 --- a/backend/crates/gateway-admin/tests/use_case/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/mod.rs @@ -5,6 +5,7 @@ mod backup; mod client_keys; mod observability; mod openai; +mod proxies; mod settings; mod system; mod xai; @@ -196,6 +197,7 @@ impl AdminHarness { self.accounts, self.account_runtime, self.account_groups, + Arc::new(proxies::UnavailableProxies), ), self.auth, self.client_keys, @@ -205,7 +207,7 @@ impl AdminHarness { ), self.providers, Arc::new(NoopSnapshot), - self.probe, + (self.probe, Arc::new(proxies::UnavailableProxies)), Arc::new(NoopClientDistribution), self.system, ) diff --git a/backend/crates/gateway-admin/tests/use_case/openai.rs b/backend/crates/gateway-admin/tests/use_case/openai.rs index ba22e59d0..bee8fff33 100644 --- a/backend/crates/gateway-admin/tests/use_case/openai.rs +++ b/backend/crates/gateway-admin/tests/use_case/openai.rs @@ -50,6 +50,7 @@ async fn openai_import_should_prepare_before_atomic_store_commit() { services .openai() .import_document(ImportCredentials { + outbound_proxy_id: None, settings: Some(super::accounts::import_settings()), context: context("import-openai"), document: document(), @@ -107,6 +108,7 @@ async fn openai_import_should_expose_only_explicit_public_errors_without_committ let error = services .openai() .import_document(ImportCredentials { + outbound_proxy_id: None, settings: None, context: context("import-openai-pat-failure"), document: document(), @@ -132,6 +134,7 @@ async fn openai_import_should_refresh_quota_for_every_imported_account() { let result = services .openai() .import_document(ImportCredentials { + outbound_proxy_id: None, settings: None, context: context("import-openai-batch"), document: document(), @@ -176,6 +179,7 @@ async fn openai_import_should_remain_successful_when_quota_refresh_fails() { let result = services .openai() .import_document(ImportCredentials { + outbound_proxy_id: None, settings: None, context: context("import-openai-quota-failure"), document: document(), @@ -348,6 +352,7 @@ async fn openai_import_provider_error_should_not_touch_store_transaction() { services .openai() .import_document(ImportCredentials { + outbound_proxy_id: None, settings: None, context: context("import-openai-error"), document: document(), diff --git a/backend/crates/gateway-admin/tests/use_case/proxies.rs b/backend/crates/gateway-admin/tests/use_case/proxies.rs new file mode 100644 index 000000000..5d3ea1e95 --- /dev/null +++ b/backend/crates/gateway-admin/tests/use_case/proxies.rs @@ -0,0 +1,51 @@ +use async_trait::async_trait; +use gateway_admin::{ + model::{MutationContext, Revision, proxies::*}, + ports::{ + proxy::{ProxyProbe, ProxyStore}, + store::AdminStoreResult, + }, +}; +use gateway_core::account::OutboundProxy; + +pub(super) struct UnavailableProxies; + +#[async_trait] +impl ProxyStore for UnavailableProxies { + async fn list(&self, _: ProxyListQuery) -> AdminStoreResult { + Err(super::unavailable("proxy")) + } + async fn get(&self, _: &str) -> AdminStoreResult { + Err(super::unavailable("proxy")) + } + async fn create(&self, _: NewProxy, _: &MutationContext) -> AdminStoreResult { + Err(super::unavailable("proxy")) + } + async fn update(&self, _: UpdateProxy, _: &MutationContext) -> AdminStoreResult { + Err(super::unavailable("proxy")) + } + async fn delete( + &self, + _: &str, + _: Revision, + _: &MutationContext, + ) -> AdminStoreResult { + Err(super::unavailable("proxy")) + } + async fn record_test( + &self, + _: &str, + _: Revision, + _: ProxyTestResult, + _: &MutationContext, + ) -> AdminStoreResult { + Err(super::unavailable("proxy")) + } +} + +#[async_trait] +impl ProxyProbe for UnavailableProxies { + async fn test(&self, _: &OutboundProxy) -> ProxyTestResult { + panic!("unexpected proxy probe") + } +} diff --git a/backend/crates/gateway-admin/tests/use_case/xai.rs b/backend/crates/gateway-admin/tests/use_case/xai.rs index a58f98d6b..1164c5c50 100644 --- a/backend/crates/gateway-admin/tests/use_case/xai.rs +++ b/backend/crates/gateway-admin/tests/use_case/xai.rs @@ -20,6 +20,7 @@ async fn xai_import_should_prepare_before_atomic_store_commit() { services .xai() .import_document(ImportCredentials { + outbound_proxy_id: None, settings: Some(super::accounts::import_settings()), context: context("import-xai"), document: document(), @@ -61,6 +62,7 @@ async fn xai_import_should_refresh_quota_for_every_imported_account() { services .xai() .import_document(ImportCredentials { + outbound_proxy_id: None, settings: None, context: context("import-xai-batch"), document: document(), @@ -96,6 +98,7 @@ async fn xai_import_should_remain_successful_when_quota_refresh_fails() { let result = services .xai() .import_document(ImportCredentials { + outbound_proxy_id: None, settings: None, context: context("import-xai-quota-failure"), document: document(), diff --git a/backend/crates/gateway-api/src/admin/accounts/credentials.rs b/backend/crates/gateway-api/src/admin/accounts/credentials.rs index 9e06c9ab4..2f9d4d5c3 100644 --- a/backend/crates/gateway-api/src/admin/accounts/credentials.rs +++ b/backend/crates/gateway-api/src/admin/accounts/credentials.rs @@ -53,6 +53,7 @@ impl AccountImportSettingsRequest { #[derive(Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct AccountImportRequest { + pub outbound_proxy_id: Option, pub settings: Option, pub provider: String, pub data: Value, @@ -60,6 +61,9 @@ pub struct AccountImportRequest { impl AccountImportRequest { pub fn validate(&self) -> Result<(), WireValidationError> { + if let Some(id) = &self.outbound_proxy_id { + require_wire_id(id, "outboundProxyId")?; + } if let Some(settings) = &self.settings { settings.validate()?; } @@ -82,6 +86,7 @@ impl AccountImportRequest { Ok(( provider, ImportCredentials { + outbound_proxy_id: self.outbound_proxy_id, settings: self .settings .map(AccountImportSettingsRequest::into_settings) @@ -96,6 +101,7 @@ impl AccountImportRequest { #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct StartAccountAuthorizationRequest { + pub outbound_proxy_id: Option, pub provider: String, pub name: String, pub account_id: Option, @@ -129,7 +135,10 @@ impl StartAccountAuthorizationRequest { context, name: self.name, reauthorization, - outbound_proxy: self.outbound_proxy_url.and_then(|value| value.0), + outbound_proxy: super::wire::proxy_selection( + self.outbound_proxy_id, + self.outbound_proxy_url, + )?, }, )) } @@ -188,6 +197,7 @@ impl CompleteAccountAuthorizationRequest { #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct UpdateAccountRequest { + pub outbound_proxy_id: Option, pub outbound_proxy_url: Option, pub account_id: String, pub enabled: bool, @@ -209,7 +219,10 @@ impl UpdateAccountRequest { pub(super) fn into_command(self) -> Result { self.validate()?; Ok(UpdateAccount { - outbound_proxy: self.outbound_proxy_url.map(|value| value.0), + outbound_proxy: super::wire::proxy_selection( + self.outbound_proxy_id, + self.outbound_proxy_url, + )?, account_id: self.account_id, enabled: self.enabled, concurrency_limit: parse_concurrency_limit(self.concurrency_limit)?, diff --git a/backend/crates/gateway-api/src/admin/accounts/wire.rs b/backend/crates/gateway-api/src/admin/accounts/wire.rs index 80b7742de..e630a868f 100644 --- a/backend/crates/gateway-api/src/admin/accounts/wire.rs +++ b/backend/crates/gateway-api/src/admin/accounts/wire.rs @@ -17,6 +17,33 @@ impl<'de> Deserialize<'de> for AccountProxyUpdate { } } +pub(super) fn proxy_selection( + id: Option, + url: Option, +) -> Result, WireValidationError> { + use gateway_admin::model::proxies::AccountProxySelection; + match (id, url) { + (Some(_), Some(_)) => Err(WireValidationError::new("outboundProxyId")), + (Some(id), None) if id.is_empty() => Ok(Some(AccountProxySelection::Direct)), + (Some(id), None) => { + if id.len() > 128 + || !id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) + { + return Err(WireValidationError::new("outboundProxyId")); + } + Ok(Some(AccountProxySelection::Saved(id))) + } + (None, Some(value)) => Ok(Some( + value + .0 + .map_or(AccountProxySelection::Direct, AccountProxySelection::Url), + )), + (None, None) => Ok(None), + } +} + /// 账号列表查询参数。 #[derive(Debug, Clone, Default, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] @@ -34,6 +61,7 @@ pub struct ListQuery { #[derive(Debug, Clone, Deserialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] pub struct BatchUpdateAccountsRequest { + pub outbound_proxy_id: Option, pub outbound_proxy_url: Option, pub account_ids: Vec, pub enabled: bool, @@ -84,7 +112,7 @@ impl BatchUpdateAccountsRequest { pub(super) fn into_command(self) -> Result { self.validate()?; Ok(BatchUpdateAccounts { - outbound_proxy: self.outbound_proxy_url.map(|value| value.0), + outbound_proxy: proxy_selection(self.outbound_proxy_id, self.outbound_proxy_url)?, account_ids: self.account_ids, enabled: self.enabled, concurrency_limit: parse_concurrency_limit(self.concurrency_limit)?, diff --git a/backend/crates/gateway-api/src/admin/mod.rs b/backend/crates/gateway-api/src/admin/mod.rs index c2d6689ef..7da028332 100644 --- a/backend/crates/gateway-api/src/admin/mod.rs +++ b/backend/crates/gateway-api/src/admin/mod.rs @@ -16,6 +16,7 @@ pub mod client_keys; mod extract; pub mod observability; pub mod presenter; +pub mod proxies; pub mod settings; pub mod system; pub mod wire; @@ -34,6 +35,7 @@ where { Router::new() .merge(account_groups::router::()) + .merge(proxies::router::()) .merge(accounts::router::()) .merge(auth::router::()) .merge(backups::router::()) diff --git a/backend/crates/gateway-api/src/admin/proxies.rs b/backend/crates/gateway-api/src/admin/proxies.rs new file mode 100644 index 000000000..65e00a2a8 --- /dev/null +++ b/backend/crates/gateway-api/src/admin/proxies.rs @@ -0,0 +1,301 @@ +use axum::{ + Router, + extract::State, + http::StatusCode, + response::IntoResponse, + routing::{get, post}, +}; +use gateway_admin::model::{ + PageSize, Revision, + proxies::{NewProxy, ProxyListQuery, ProxyMutation, ProxyRecord, UpdateProxy}, +}; +use serde::{Deserialize, Serialize}; + +use super::{ + AdminAuth, AdminEnvelope, AdminError, AdminJson, AdminQuery, AdminResponse, AdminSessionState, + PageMeta, accounts::AccountProxyUpdate, +}; + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct ListQuery { + page: Option, + page_size: Option, + search: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct CreateRequest { + name: String, + proxy_url: AccountProxyUpdate, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct UpdateRequest { + id: String, + revision: u64, + name: String, + proxy_url: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct IdRequest { + id: String, + revision: u64, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ProxyTestView { + success: bool, + latency_ms: u64, + exit_ip: Option, + message: String, +} + +#[derive(Serialize)] +struct ProxyAccountView { + id: String, + name: String, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct ProxyView { + id: String, + name: String, + endpoint: String, + has_authentication: bool, + revision: u64, + accounts: Vec, + last_test_at: Option, + last_test: Option, + created_at: String, + updated_at: String, +} + +impl From for ProxyView { + fn from(record: ProxyRecord) -> Self { + let endpoint = record.proxy.endpoint(); + Self { + id: record.id, + name: record.name, + has_authentication: record.proxy.expose_url() != endpoint, + endpoint, + revision: record.revision.get(), + accounts: record + .accounts + .into_iter() + .map(|account| ProxyAccountView { + id: account.id, + name: account.name, + }) + .collect(), + last_test_at: record.last_test_at.map(|at| at.to_rfc3339()), + last_test: record.last_test.map(|result| ProxyTestView { + success: result.success, + latency_ms: result.latency_ms, + exit_ip: result.exit_ip.map(|ip| ip.to_string()), + message: result.message, + }), + created_at: record.created_at.to_rfc3339(), + updated_at: record.updated_at.to_rfc3339(), + } + } +} + +#[derive(Serialize)] +struct ProxyPageView { + items: Vec, + page: PageMeta, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct MutationView { + record: ProxyView, + config_revision: u64, +} + +impl From for MutationView { + fn from(value: ProxyMutation) -> Self { + Self { + record: value.record.into(), + config_revision: value.config_revision.get(), + } + } +} + +pub fn router() -> Router +where + S: AdminSessionState + Clone + Send + Sync + 'static, +{ + Router::new() + .route("/api/admin/proxies", get(list::)) + .route("/api/admin/proxies/create", post(create::)) + .route("/api/admin/proxies/update", post(update::)) + .route("/api/admin/proxies/delete", post(delete::)) + .route("/api/admin/proxies/test", post(test::)) +} + +fn revision(value: u64) -> Result { + if value > i64::MAX as u64 { + return Err(AdminError::bad_request("代理版本不合法")); + } + Revision::new(value).map_err(|_| AdminError::bad_request("代理版本不合法")) +} + +fn map_error(error: gateway_admin::model::AdminError) -> AdminError { + super::wire::map_admin_service_error(error) +} + +async fn list( + _: AdminAuth, + State(state): State, + AdminQuery(query): AdminQuery, +) -> Result +where + S: AdminSessionState + Send + Sync, +{ + let result = state + .admin_services() + .proxies() + .list(ProxyListQuery { + page: query.page.unwrap_or(1), + page_size: PageSize::new(query.page_size.unwrap_or(20)) + .map_err(|_| AdminError::bad_request("分页大小不合法"))?, + search: query.search.unwrap_or_default().trim().to_owned(), + }) + .await + .map_err(map_error)?; + let total_pages = result.total.div_ceil(u64::from(result.page_size)); + Ok(AdminResponse::new( + StatusCode::OK, + AdminEnvelope::ok(ProxyPageView { + items: result.items.into_iter().map(Into::into).collect(), + page: PageMeta::new( + result.page, + u32::from(result.page_size), + result.total, + u32::try_from(total_pages).unwrap_or(u32::MAX), + ), + }), + )) +} + +async fn create( + auth: AdminAuth, + State(state): State, + AdminJson(request): AdminJson, +) -> Result +where + S: AdminSessionState + Send + Sync, +{ + let proxy = request + .proxy_url + .0 + .ok_or_else(|| AdminError::bad_request("代理 URL 不能为空"))?; + let result = state + .admin_services() + .proxies() + .create( + NewProxy { + name: request.name, + proxy, + }, + &auth.context().mutation_context(), + ) + .await + .map_err(map_error)?; + Ok(AdminResponse::new( + StatusCode::CREATED, + AdminEnvelope::ok(MutationView::from(result)), + )) +} + +async fn update( + auth: AdminAuth, + State(state): State, + AdminJson(request): AdminJson, +) -> Result +where + S: AdminSessionState + Send + Sync, +{ + let proxy = request + .proxy_url + .map(|value| { + value + .0 + .ok_or_else(|| AdminError::bad_request("代理 URL 不能为空")) + }) + .transpose()?; + let result = state + .admin_services() + .proxies() + .update( + UpdateProxy { + id: request.id, + revision: revision(request.revision)?, + name: request.name, + proxy, + }, + &auth.context().mutation_context(), + ) + .await + .map_err(map_error)?; + Ok(AdminResponse::new( + StatusCode::OK, + AdminEnvelope::ok(MutationView::from(result)), + )) +} + +async fn delete( + auth: AdminAuth, + State(state): State, + AdminJson(request): AdminJson, +) -> Result +where + S: AdminSessionState + Send + Sync, +{ + let result = state + .admin_services() + .proxies() + .delete( + &request.id, + revision(request.revision)?, + &auth.context().mutation_context(), + ) + .await + .map_err(map_error)?; + Ok(AdminResponse::new( + StatusCode::OK, + AdminEnvelope::ok(serde_json::json!({"configRevision": result.get()})), + )) +} + +async fn test( + auth: AdminAuth, + State(state): State, + AdminJson(request): AdminJson, +) -> Result +where + S: AdminSessionState + Send + Sync, +{ + let result = state + .admin_services() + .proxies() + .test( + &request.id, + revision(request.revision)?, + &auth.context().mutation_context(), + ) + .await + .map_err(map_error)?; + Ok(AdminResponse::new( + StatusCode::OK, + AdminEnvelope::ok(ProxyView::from(result)), + )) +} diff --git a/backend/crates/gateway-api/tests/admin/mod.rs b/backend/crates/gateway-api/tests/admin/mod.rs index cdc9e339f..18c5c6e5f 100644 --- a/backend/crates/gateway-api/tests/admin/mod.rs +++ b/backend/crates/gateway-api/tests/admin/mod.rs @@ -82,6 +82,7 @@ mod auth; mod client_keys; mod errors; mod observability; +mod proxies; mod settings; mod system; mod wire; @@ -124,7 +125,12 @@ impl AdminTestFixture { dashboard_summary_range: Arc::clone(&dashboard_summary_range), }); let stores = AdminStorePorts::new( - AdminAccountStorePorts::new(unused.clone(), unused.clone(), account_groups.clone()), + AdminAccountStorePorts::new( + unused.clone(), + unused.clone(), + account_groups.clone(), + Arc::new(proxies::MemoryProxies::default()), + ), auth.clone(), client_keys.clone(), unused, @@ -144,7 +150,7 @@ impl AdminTestFixture { stores, providers, Arc::new(NoopSnapshot), - Arc::new(NoopProbe), + (Arc::new(NoopProbe), Arc::new(proxies::SuccessfulProbe)), Arc::new(StaticClientDistribution), system, ) diff --git a/backend/crates/gateway-api/tests/admin/proxies.rs b/backend/crates/gateway-api/tests/admin/proxies.rs new file mode 100644 index 000000000..35d5aff2f --- /dev/null +++ b/backend/crates/gateway-api/tests/admin/proxies.rs @@ -0,0 +1,280 @@ +use std::sync::Mutex; + +use async_trait::async_trait; +use axum::{ + body::{Body, to_bytes}, + http::{Request, StatusCode, header}, +}; +use chrono::Utc; +use gateway_admin::{ + model::{MutationContext, Revision, proxies::*}, + ports::{ + proxy::{ProxyProbe, ProxyStore}, + store::{AdminStoreError, AdminStoreErrorKind, AdminStoreResult}, + }, +}; +use gateway_core::account::OutboundProxy; +use serde_json::{Value, json}; +use tower::ServiceExt as _; + +use super::{AdminTestFixture, AdminTestState}; + +#[derive(Default)] +pub(super) struct MemoryProxies(Mutex>); + +fn missing() -> AdminStoreError { + AdminStoreError::new(AdminStoreErrorKind::NotFound, "proxy", "missing proxy") +} + +#[async_trait] +impl ProxyStore for MemoryProxies { + async fn list(&self, query: ProxyListQuery) -> AdminStoreResult { + let items: Vec<_> = self.0.lock().unwrap().iter().cloned().collect(); + Ok(ProxyPage { + total: u64::try_from(items.len()).unwrap(), + items, + page: query.page, + page_size: query.page_size.get(), + }) + } + async fn get(&self, _: &str) -> AdminStoreResult { + self.0.lock().unwrap().clone().ok_or_else(missing) + } + async fn create( + &self, + command: NewProxy, + _: &MutationContext, + ) -> AdminStoreResult { + let record = ProxyRecord { + id: "proxy_test".to_owned(), + name: command.name, + proxy: command.proxy, + revision: Revision::new(1).unwrap(), + accounts: vec![], + last_test: None, + last_test_at: None, + created_at: Utc::now(), + updated_at: Utc::now(), + }; + *self.0.lock().unwrap() = Some(record.clone()); + Ok(ProxyMutation { + config_revision: record.revision, + record, + }) + } + async fn update( + &self, + command: UpdateProxy, + _: &MutationContext, + ) -> AdminStoreResult { + let mut stored = self.0.lock().unwrap(); + let record = stored.as_mut().ok_or_else(missing)?; + record.name = command.name; + if let Some(proxy) = command.proxy { + record.proxy = proxy; + } + record.revision = Revision::new(record.revision.get() + 1).unwrap(); + Ok(ProxyMutation { + config_revision: record.revision, + record: record.clone(), + }) + } + async fn delete( + &self, + _: &str, + _: Revision, + _: &MutationContext, + ) -> AdminStoreResult { + self.0.lock().unwrap().take().ok_or_else(missing)?; + Ok(Revision::new(3).unwrap()) + } + async fn record_test( + &self, + _: &str, + _: Revision, + result: ProxyTestResult, + _: &MutationContext, + ) -> AdminStoreResult { + let mut stored = self.0.lock().unwrap(); + let record = stored.as_mut().ok_or_else(missing)?; + record.last_test = Some(result); + record.last_test_at = Some(Utc::now()); + Ok(record.clone()) + } +} + +pub(super) struct SuccessfulProbe; + +#[async_trait] +impl ProxyProbe for SuccessfulProbe { + async fn test(&self, proxy: &OutboundProxy) -> ProxyTestResult { + assert_eq!( + proxy.expose_url(), + "http://test-user:private-password@proxy.example:8080/" + ); + ProxyTestResult { + success: true, + latency_ms: 15, + exit_ip: Some("203.0.113.2".parse().unwrap()), + message: "Connected".to_owned(), + } + } +} + +async fn request( + fixture: &AdminTestFixture, + path: &str, + body: Option, + authenticated: bool, +) -> (StatusCode, Value) { + let mut builder = Request::builder() + .header("x-request-id", "req_proxy_tests") + .uri(path) + .method(if body.is_some() { "POST" } else { "GET" }); + if authenticated { + builder = builder.header(header::COOKIE, "cpr_admin_session=valid-session"); + } + let body = body.map_or_else(Body::empty, |body| Body::from(body.to_string())); + let response = gateway_api::admin::proxies::router::() + .with_state(fixture.state()) + .oneshot( + builder + .header(header::CONTENT_TYPE, "application/json") + .body(body) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let body = to_bytes(response.into_body(), 64 * 1024).await.unwrap(); + let text = std::str::from_utf8(&body).unwrap(); + assert!(!text.contains("private-password")); + assert!(!text.contains("test-user")); + (status, serde_json::from_slice(&body).unwrap()) +} + +#[tokio::test] +async fn proxy_routes_save_reload_test_rename_and_delete_without_exposing_credentials() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + let (status, created) = request( + &fixture, + "/api/admin/proxies/create", + Some(json!({ + "name": " Office ", "proxyUrl": "http://test-user:private-password@proxy.example:8080" + })), + true, + ) + .await; + assert_eq!(status, StatusCode::CREATED); + assert_eq!(created["data"]["record"]["name"], "Office"); + assert_eq!(created["data"]["record"]["hasAuthentication"], true); + assert_eq!( + created["data"]["record"]["endpoint"], + "http://proxy.example:8080/" + ); + assert!(created["data"]["record"].get("proxyUrl").is_none()); + + let (status, tested) = request( + &fixture, + "/api/admin/proxies/test", + Some(json!({"id": "proxy_test", "revision": 1})), + true, + ) + .await; + assert_eq!(status, StatusCode::OK); + assert_eq!(tested["data"]["lastTest"]["exitIp"], "203.0.113.2"); + let (_, listed) = request( + &fixture, + "/api/admin/proxies?page=1&pageSize=20", + None, + true, + ) + .await; + assert_eq!(listed["data"]["items"][0], tested["data"]); + + let (status, _) = request( + &fixture, + "/api/admin/proxies/update", + Some(json!({"id": "proxy_test", "revision": 1, "name": "Renamed"})), + true, + ) + .await; + assert_eq!(status, StatusCode::OK); + let (status, _) = request( + &fixture, + "/api/admin/proxies/test", + Some(json!({"id": "proxy_test", "revision": 1})), + true, + ) + .await; + assert_eq!(status, StatusCode::CONFLICT); + let (status, _) = request( + &fixture, + "/api/admin/proxies/test", + Some(json!({"id": "proxy_test", "revision": 2})), + true, + ) + .await; + assert_eq!(status, StatusCode::OK); + let (status, _) = request( + &fixture, + "/api/admin/proxies/delete", + Some(json!({"id": "proxy_test", "revision": 2})), + true, + ) + .await; + assert_eq!(status, StatusCode::OK); + let (_, listed) = request(&fixture, "/api/admin/proxies", None, true).await; + assert_eq!(listed["data"]["page"]["total"], 0); +} + +#[tokio::test] +async fn proxy_routes_require_auth_and_reject_invalid_input() { + let fixture = AdminTestFixture::new().await; + fixture.auth.insert_session("valid-session"); + assert_eq!( + request(&fixture, "/api/admin/proxies", None, false).await.0, + StatusCode::UNAUTHORIZED + ); + for action in ["create", "update", "delete", "test"] { + assert_eq!( + request( + &fixture, + &format!("/api/admin/proxies/{action}"), + Some(json!({})), + false + ) + .await + .0, + StatusCode::UNAUTHORIZED + ); + } + for query in ["page=0", "pageSize=0", "pageSize=201", "unknown=1"] { + assert_eq!( + request(&fixture, &format!("/api/admin/proxies?{query}"), None, true) + .await + .0, + StatusCode::BAD_REQUEST + ); + } + for body in [ + json!({"name":" ","proxyUrl":"http://proxy.example:8080"}), + json!({"name":"Invalid","proxyUrl":""}), + ] { + assert_eq!( + request(&fixture, "/api/admin/proxies/create", Some(body), true) + .await + .0, + StatusCode::BAD_REQUEST + ); + } + let (status, _) = request( + &fixture, + "/api/admin/proxies/create", + Some(json!({"name":"Invalid","proxyUrl":"ftp://test-user:private-password@proxy.example"})), + true, + ) + .await; + assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY); +} diff --git a/backend/crates/gateway-api/tests/architecture.rs b/backend/crates/gateway-api/tests/architecture.rs index 68ec244fd..6a549410d 100644 --- a/backend/crates/gateway-api/tests/architecture.rs +++ b/backend/crates/gateway-api/tests/architecture.rs @@ -50,6 +50,7 @@ fn source_tree_should_match_frozen_machine_manifest() { "src/admin/observability/routes.rs", "src/admin/observability/wire.rs", "src/admin/presenter.rs", + "src/admin/proxies.rs", "src/admin/settings.rs", "src/admin/system.rs", "src/admin/wire.rs", @@ -96,6 +97,7 @@ fn test_tree_should_match_frozen_rust_mirror() { "tests/admin/observability/mod.rs", "tests/admin/observability/query.rs", "tests/admin/observability/response.rs", + "tests/admin/proxies.rs", "tests/admin/settings.rs", "tests/admin/system.rs", "tests/admin/wire.rs", diff --git a/backend/crates/gateway-host/Cargo.toml b/backend/crates/gateway-host/Cargo.toml index adc0c71b9..15a21f413 100644 --- a/backend/crates/gateway-host/Cargo.toml +++ b/backend/crates/gateway-host/Cargo.toml @@ -17,7 +17,7 @@ futures = "0.3.32" gateway-admin = { path = "../gateway-admin" } gateway-core = { path = "../gateway-core" } hex = "0.4.3" -reqwest = { version = "=0.12.28", default-features = false, features = ["json", "stream", "rustls-tls-native-roots"] } +reqwest = { version = "=0.12.28", default-features = false, features = ["json", "stream", "rustls-tls-native-roots", "socks"] } scraper = "0.27.0" semver = "1.0.27" serde = { version = "1.0.228", features = ["derive"] } diff --git a/backend/crates/gateway-host/src/lib.rs b/backend/crates/gateway-host/src/lib.rs index ad9d214e8..37e0e8362 100644 --- a/backend/crates/gateway-host/src/lib.rs +++ b/backend/crates/gateway-host/src/lib.rs @@ -3,6 +3,7 @@ pub mod client_distribution; pub mod config; mod logging; +pub mod proxy_probe; pub mod serve; pub mod system_update; pub mod workers; @@ -81,6 +82,11 @@ impl HostBundle { self.client_distribution.clone() } + #[must_use] + pub fn proxy_probe(&self) -> Arc { + Arc::new(proxy_probe::HttpProxyProbe::default()) + } + #[must_use] pub fn connection_lifecycle(&self) -> Arc { self.connections.clone() diff --git a/backend/crates/gateway-host/src/proxy_probe.rs b/backend/crates/gateway-host/src/proxy_probe.rs new file mode 100644 index 000000000..61308cdc2 --- /dev/null +++ b/backend/crates/gateway-host/src/proxy_probe.rs @@ -0,0 +1,88 @@ +//! Bounded egress checks using the same explicit proxy protocols as provider requests. + +use std::{ + net::IpAddr, + time::{Duration, Instant}, +}; + +use async_trait::async_trait; +use gateway_admin::{model::proxies::ProxyTestResult, ports::proxy::ProxyProbe}; +use gateway_core::account::OutboundProxy; +use serde::Deserialize; + +pub struct HttpProxyProbe { + endpoint: String, +} + +impl Default for HttpProxyProbe { + fn default() -> Self { + Self::new("https://api.ipify.org?format=json") + } +} + +impl HttpProxyProbe { + #[must_use] + pub fn new(endpoint: impl Into) -> Self { + Self { + endpoint: endpoint.into(), + } + } + + async fn exit_ip(&self, proxy: &OutboundProxy) -> Result { + let proxy = reqwest::Proxy::all(proxy.expose_url()).map_err(|_| "代理地址不合法")?; + let client = reqwest::Client::builder() + .no_proxy() + .proxy(proxy) + .connect_timeout(Duration::from_secs(5)) + .timeout(Duration::from_secs(12)) + .redirect(reqwest::redirect::Policy::none()) + .build() + .map_err(|_| "无法创建代理连接")?; + let mut response = client.get(&self.endpoint).send().await.map_err(|error| { + if error.is_timeout() { + "代理连接超时" + } else { + "代理连接失败，请检查地址、认证和网络" + } + })?; + if !response.status().is_success() { + return Err( + if response.status() == reqwest::StatusCode::PROXY_AUTHENTICATION_REQUIRED { + "代理认证失败" + } else { + "出口检测服务返回错误状态" + }, + ); + } + let mut body = Vec::new(); + while let Some(chunk) = response.chunk().await.map_err(|_| "出口检测响应读取失败")? + { + if body.len() + chunk.len() > 1024 { + return Err("出口检测响应过大"); + } + body.extend_from_slice(&chunk); + } + #[derive(Deserialize)] + struct Response { + ip: IpAddr, + } + serde_json::from_slice::(&body) + .map(|response| response.ip) + .map_err(|_| "出口检测响应不合法") + } +} + +#[async_trait] +impl ProxyProbe for HttpProxyProbe { + async fn test(&self, proxy: &OutboundProxy) -> ProxyTestResult { + let started = Instant::now(); + let result = tokio::time::timeout(Duration::from_secs(15), self.exit_ip(proxy)).await; + let result = result.unwrap_or(Err("代理连接超时")); + ProxyTestResult { + success: result.is_ok(), + latency_ms: u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX), + exit_ip: result.as_ref().ok().copied(), + message: result.map_or_else(str::to_owned, |_| "连接成功".to_owned()), + } + } +} diff --git a/backend/crates/gateway-host/tests/main.rs b/backend/crates/gateway-host/tests/main.rs index 650ad1be0..689ff5d3f 100644 --- a/backend/crates/gateway-host/tests/main.rs +++ b/backend/crates/gateway-host/tests/main.rs @@ -1,6 +1,7 @@ mod client_distribution; mod config; mod logging; +mod proxy_probe; mod serve; mod system_update; mod workers; diff --git a/backend/crates/gateway-host/tests/proxy_probe.rs b/backend/crates/gateway-host/tests/proxy_probe.rs new file mode 100644 index 000000000..a253bf6cd --- /dev/null +++ b/backend/crates/gateway-host/tests/proxy_probe.rs @@ -0,0 +1,62 @@ +use gateway_admin::ports::proxy::ProxyProbe; +use gateway_core::account::OutboundProxy; +use gateway_host::proxy_probe::HttpProxyProbe; +use serde_json::json; +use wiremock::{ + Mock, MockServer, ResponseTemplate, + matchers::{any, header}, +}; + +#[tokio::test] +async fn proxy_probe_sends_authentication_through_explicit_proxy() { + let proxy_server = MockServer::start().await; + Mock::given(header("proxy-authorization", "Basic dXNlcjpwYXNzd29yZA==")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"ip": "203.0.113.8"}))) + .expect(1) + .mount(&proxy_server) + .await; + let proxy = + OutboundProxy::parse(&format!("http://user:password@{}", proxy_server.address())).unwrap(); + let result = HttpProxyProbe::new("http://unresolvable.invalid/ip") + .test(&proxy) + .await; + assert!(result.success); + assert_eq!(result.exit_ip.unwrap().to_string(), "203.0.113.8"); +} + +#[tokio::test] +async fn proxy_probe_rejects_auth_errors_redirects_and_invalid_or_oversized_responses() { + for response in [ + ResponseTemplate::new(407), + ResponseTemplate::new(302).insert_header("Location", "http://127.0.0.1/"), + ResponseTemplate::new(200).set_body_json(json!({"ip": "not-an-ip"})), + ResponseTemplate::new(200).set_body_string("a".repeat(1025)), + ] { + let proxy_server = MockServer::start().await; + Mock::given(any()) + .respond_with(response) + .expect(1) + .mount(&proxy_server) + .await; + let result = HttpProxyProbe::new("http://unresolvable.invalid/ip") + .test(&OutboundProxy::parse(&proxy_server.uri()).unwrap()) + .await; + assert!(!result.success); + assert!(result.exit_ip.is_none()); + } +} + +#[tokio::test] +async fn unavailable_proxy_never_falls_back_to_direct_connection() { + let target = MockServer::start().await; + Mock::given(any()) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({"ip":"203.0.113.8"}))) + .expect(0) + .mount(&target) + .await; + let unused = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy = OutboundProxy::parse(&format!("http://{}", unused.local_addr().unwrap())).unwrap(); + drop(unused); + let result = HttpProxyProbe::new(target.uri()).test(&proxy).await; + assert!(!result.success); +} diff --git a/backend/crates/gateway-store/src/bundle.rs b/backend/crates/gateway-store/src/bundle.rs index 1a954f502..9f6c839db 100644 --- a/backend/crates/gateway-store/src/bundle.rs +++ b/backend/crates/gateway-store/src/bundle.rs @@ -106,6 +106,7 @@ pub async fn initialize(mut config: StoreConfig) -> StoreResult { )), admin_account_runtime, Arc::new(postgres::PgAccountGroupRepository::new(pool.clone())), + Arc::new(postgres::PgProxyRepository::new(pool.clone())), ), Arc::new(AdminAuthStoreAdapter { security: postgres::PgAdminSecurityAuditRepository::new(pool.clone()), diff --git a/backend/crates/gateway-store/src/postgres/mod.rs b/backend/crates/gateway-store/src/postgres/mod.rs index 9c2550681..ca4307df0 100644 --- a/backend/crates/gateway-store/src/postgres/mod.rs +++ b/backend/crates/gateway-store/src/postgres/mod.rs @@ -22,6 +22,7 @@ mod execution_buffer; mod observability; mod ops_events; mod provider_accounts; +mod proxies; mod retention; mod runtime_settings; mod snapshot; @@ -38,6 +39,7 @@ pub use execution_buffer::*; pub use observability::*; pub use ops_events::*; pub use provider_accounts::*; +pub use proxies::PgProxyRepository; pub use retention::*; pub use runtime_settings::*; pub use snapshot::*; diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs index 4f06f43ab..2b98ec716 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/admin_adapter.rs @@ -216,6 +216,7 @@ impl PgAdminAccountStore { settings: Option, context: &MutationContext, action: &str, + outbound_proxy: Option, ) -> AdminStoreResult { let provider_kind = prepared.provider_kind.as_str().to_owned(); let accounts = prepared @@ -233,6 +234,7 @@ impl PgAdminAccountStore { .accounts .import_provider_accounts(ImportProviderAccounts { settings, + outbound_proxy, scope: ProviderAccountAdminScope { provider_kind: provider_kind.clone(), }, @@ -627,6 +629,7 @@ impl AccountStore for PgAdminAccountStore { command.settings, context, "import_document", + command.outbound_proxy, ) .await } @@ -650,6 +653,16 @@ impl AccountStore for PgAdminAccountStore { command.settings, context, "authorize", + command + .pending + .outbound_proxy_id() + .zip(command.pending.outbound_proxy()) + .map( + |(id, proxy)| gateway_admin::model::proxies::ImportProxyBinding { + id: id.to_owned(), + proxy: proxy.clone(), + }, + ), ) .await?; let [account_id]: [CoreProviderAccountId; 1] = diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/repository.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/repository.rs index d8da8566c..b98911cbe 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/repository.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/repository.rs @@ -133,15 +133,28 @@ impl ProviderAccountRepository for PgProviderAccountRepository { } else { CredentialState::Unknown }; + let mut transaction = self + .pool + .begin() + .await + .map_err(|_| postgres_unavailable("begin provider account insert"))?; + let proxy_id = match account.outbound_proxy.as_ref() { + Some(proxy) => Some( + super::super::proxies::ensure_proxy_for_url(&mut transaction, proxy, None) + .await? + .0, + ), + None => None, + }; sqlx::query( "insert into provider_accounts ( - outbound_proxy_url, id, provider_kind, name, email, upstream_user_id, + outbound_proxy_url, outbound_proxy_id, id, provider_kind, name, email, upstream_user_id, upstream_account_id, plan_type, authentication_kind, provider_credentials_json, credential_revision, has_refresh_token, access_token_expires_at, next_refresh_at, enabled, concurrency_limit, weight, credential_state, provider_quota_json, credential_observed_at, quota_access_observed_at, quota_observed_at, created_at, updated_at ) values ( - $18, $1, $2, $3, $4, $5, $6, $7, $8, $9, 1, $10, $11, $12, $13, + $18, $19, $1, $2, $3, $4, $5, $6, $7, $8, $9, 1, $10, $11, $12, $13, $14, $15, $16, null, $17, null, null, now(), greatest(now(), $17) )", ) @@ -163,9 +176,14 @@ impl ProviderAccountRepository for PgProviderAccountRepository { .bind(credential_state.as_str()) .bind(account.credential_observed_at) .bind(account.outbound_proxy.as_ref().map(|proxy| proxy.expose_url())) - .execute(&self.pool) + .bind(proxy_id) + .execute(&mut *transaction) .await .map_err(|_| postgres_unavailable("insert provider account"))?; + transaction + .commit() + .await + .map_err(|_| postgres_unavailable("commit provider account insert"))?; Ok(()) } @@ -443,6 +461,22 @@ impl ProviderAccountAdminRepository for PgProviderAccountRepository { .map_err(|_| postgres_unavailable("begin provider account admin import"))?; let result = async { let revision = bump_config_revision_in_transaction(&mut transaction).await?; + if let Some(binding) = &command.outbound_proxy { + let (_, current) = super::super::proxies::resolve_proxy_selection( + &mut transaction, + &gateway_admin::model::proxies::AccountProxySelection::Saved( + binding.id.clone(), + ), + ) + .await?; + if current.as_ref() != Some(&binding.proxy) { + return Err(StoreError::Conflict { + entity: "outbound proxy", + id: binding.id.clone(), + kind: ConflictKind::StaleRevision, + }); + } + } let mut account_ids = Vec::with_capacity(command.accounts.len()); for account in &command.accounts { account_ids @@ -701,15 +735,23 @@ pub(crate) async fn upsert_provider_account_in_transaction( } else { CredentialState::Unknown }; + let proxy_id = match account.outbound_proxy.as_ref() { + Some(proxy) => Some( + super::super::proxies::ensure_proxy_for_url(transaction, proxy, None) + .await? + .0, + ), + None => None, + }; let imported_id = sqlx::query_scalar::<_, String>( "insert into provider_accounts ( - outbound_proxy_url, id, provider_kind, name, email, upstream_user_id, + outbound_proxy_url, outbound_proxy_id, id, provider_kind, name, email, upstream_user_id, upstream_account_id, plan_type, authentication_kind, provider_credentials_json, credential_revision, has_refresh_token, access_token_expires_at, next_refresh_at, enabled, concurrency_limit, weight, credential_state, provider_quota_json, credential_observed_at, quota_access_observed_at, quota_observed_at, created_at, updated_at ) values ( - $18, $1, $2, $3, $4, $5, $6, $7, $8, $9, 1, $10, $11, $12, $13, + $18, $19, $1, $2, $3, $4, $5, $6, $7, $8, $9, 1, $10, $11, $12, $13, $14, $15, $16, null, $17, null, null, now(), greatest(now(), $17) ) on conflict ( @@ -723,6 +765,7 @@ pub(crate) async fn upsert_provider_account_in_transaction( authentication_kind = excluded.authentication_kind, provider_credentials_json = excluded.provider_credentials_json, outbound_proxy_url = coalesce(excluded.outbound_proxy_url, provider_accounts.outbound_proxy_url), + outbound_proxy_id = coalesce(excluded.outbound_proxy_id, provider_accounts.outbound_proxy_id), credential_revision = provider_accounts.credential_revision + 1, has_refresh_token = excluded.has_refresh_token, access_token_expires_at = excluded.access_token_expires_at, @@ -759,6 +802,7 @@ pub(crate) async fn upsert_provider_account_in_transaction( .bind(credential_state.as_str()) .bind(account.credential_observed_at) .bind(account.outbound_proxy.as_ref().map(|proxy| proxy.expose_url())) + .bind(proxy_id) .fetch_optional(&mut **transaction) .await .map_err(|error| { @@ -865,12 +909,19 @@ pub(crate) async fn update_provider_accounts_scheduling_in_transaction( enabled: bool, concurrency_limit: Option, weight: AccountWeight, - outbound_proxy: Option<&Option>, + outbound_proxy: Option<&gateway_admin::model::proxies::AccountProxySelection>, ) -> StoreResult<()> { + let (proxy_id, proxy) = match outbound_proxy { + Some(selection) => { + super::super::proxies::resolve_proxy_selection(transaction, selection).await? + } + None => (None, None), + }; let updated = sqlx::query_scalar::<_, String>( "update provider_accounts set enabled = $2, concurrency_limit = $3, weight = $4, updated_at = greatest(now(), updated_at), - outbound_proxy_url = case when $5 then $6 else outbound_proxy_url end + outbound_proxy_url = case when $5 then $6 else outbound_proxy_url end, + outbound_proxy_id = case when $5 then $7 else outbound_proxy_id end where id = any($1::text[]) returning id", ) @@ -879,11 +930,8 @@ pub(crate) async fn update_provider_accounts_scheduling_in_transaction( .bind(concurrency_limit.map(|limit| i64::from(limit.get()))) .bind(i16::try_from(weight.get()).map_err(|_| invalid("invalid weight"))?) .bind(outbound_proxy.is_some()) - .bind( - outbound_proxy - .and_then(Option::as_ref) - .map(|proxy| proxy.expose_url()), - ) + .bind(proxy.as_ref().map(gateway_core::account::OutboundProxy::expose_url)) + .bind(proxy_id) .fetch_all(&mut **transaction) .await .map_err(|_| postgres_unavailable("set provider accounts state in admin transaction"))? diff --git a/backend/crates/gateway-store/src/postgres/provider_accounts/rows.rs b/backend/crates/gateway-store/src/postgres/provider_accounts/rows.rs index ae9d1998e..c963a8462 100644 --- a/backend/crates/gateway-store/src/postgres/provider_accounts/rows.rs +++ b/backend/crates/gateway-store/src/postgres/provider_accounts/rows.rs @@ -233,6 +233,7 @@ impl NewProviderAccount { #[derive(Clone)] pub struct ImportProviderAccounts { pub settings: Option, + pub outbound_proxy: Option, pub scope: ProviderAccountAdminScope, pub accounts: Vec, pub audit: AdminAuditEvent, @@ -313,7 +314,7 @@ impl fmt::Debug for RotateProviderAccount { #[derive(Debug, Clone)] pub struct BatchUpdateProviderAccountsAdmin { - pub outbound_proxy: Option>, + pub outbound_proxy: Option, pub account_ids: Vec, pub enabled: bool, pub concurrency_limit: Option, diff --git a/backend/crates/gateway-store/src/postgres/proxies.rs b/backend/crates/gateway-store/src/postgres/proxies.rs new file mode 100644 index 000000000..248ee0678 --- /dev/null +++ b/backend/crates/gateway-store/src/postgres/proxies.rs @@ -0,0 +1,444 @@ +//! Named proxies own credentials; accounts retain the resolved URL for provider transports. + +use async_trait::async_trait; +use gateway_admin::{ + model::{MutationContext, Revision as AdminRevision, proxies::*}, + ports::{ + proxy::ProxyStore, + store::{AdminStoreError, AdminStoreResult}, + }, +}; +use gateway_core::account::OutboundProxy; +use sqlx::{PgPool, Postgres, QueryBuilder, Row as _, Transaction, postgres::PgRow}; + +use super::{append_admin_audit_event_in_transaction, bump_config_revision_in_transaction}; +use crate::{ + ConflictKind, Revision, StoreError, StoreResult, admin_revision, admin_store_error, + mutation_audit, postgres_unavailable, +}; + +const ENTITY: &str = "outbound proxy"; +const SELECT: &str = "select p.*, + array(select a.id from provider_accounts a where a.outbound_proxy_id = p.id order by a.id) as account_ids, + array(select a.name from provider_accounts a where a.outbound_proxy_id = p.id order by a.id) as account_names + from outbound_proxies p"; + +#[derive(Clone)] +pub struct PgProxyRepository { + pool: PgPool, +} + +impl PgProxyRepository { + #[must_use] + pub const fn new(pool: PgPool) -> Self { + Self { pool } + } +} + +fn store_error(error: StoreError) -> AdminStoreError { + admin_store_error(ENTITY, error) +} +fn unavailable() -> StoreError { + postgres_unavailable("outbound proxy operation") +} +fn conflict(id: &str) -> StoreError { + StoreError::Conflict { + entity: ENTITY, + id: id.to_owned(), + kind: ConflictKind::InvalidTransition, + } +} +fn invalid() -> StoreError { + StoreError::InvalidData { + entity: ENTITY, + message: "invalid proxy record".to_owned(), + } +} + +fn record(row: PgRow) -> StoreResult { + let success: Option = row.try_get("last_test_success").map_err(|_| invalid())?; + let ip: Option = row.try_get("last_test_ip").map_err(|_| invalid())?; + let latency: Option = row.try_get("last_test_latency_ms").map_err(|_| invalid())?; + let ids: Vec = row.try_get("account_ids").map_err(|_| invalid())?; + let names: Vec = row.try_get("account_names").map_err(|_| invalid())?; + Ok(ProxyRecord { + id: row.try_get("id").map_err(|_| invalid())?, + name: row.try_get("name").map_err(|_| invalid())?, + proxy: OutboundProxy::parse( + &row.try_get::("proxy_url") + .map_err(|_| invalid())?, + ) + .map_err(|_| invalid())?, + revision: AdminRevision::new( + u64::try_from(row.try_get::("revision").map_err(|_| invalid())?) + .map_err(|_| invalid())?, + ) + .map_err(|_| invalid())?, + accounts: ids + .into_iter() + .zip(names) + .map(|(id, name)| ProxyAccountRef { id, name }) + .collect(), + last_test_at: row.try_get("last_test_at").map_err(|_| invalid())?, + last_test: success + .map(|success| -> StoreResult<_> { + Ok(ProxyTestResult { + success, + latency_ms: u64::try_from(latency.ok_or_else(invalid)?) + .map_err(|_| invalid())?, + exit_ip: ip.map(|ip| ip.parse().map_err(|_| invalid())).transpose()?, + message: row + .try_get::, _>("last_test_message") + .map_err(|_| invalid())? + .unwrap_or_default(), + }) + }) + .transpose()?, + created_at: row.try_get("created_at").map_err(|_| invalid())?, + updated_at: row.try_get("updated_at").map_err(|_| invalid())?, + }) +} + +async fn lock_url( + transaction: &mut Transaction<'_, Postgres>, + proxy: &OutboundProxy, +) -> StoreResult<()> { + sqlx::query("select pg_advisory_xact_lock(hashtextextended($1, 739218))") + .bind(proxy.expose_url()) + .execute(&mut **transaction) + .await + .map_err(|_| unavailable())?; + Ok(()) +} + +/// Imports and legacy URL writes join the catalog in the same account transaction. +pub(crate) async fn ensure_proxy_for_url( + transaction: &mut Transaction<'_, Postgres>, + proxy: &OutboundProxy, + name: Option<&str>, +) -> StoreResult<(String, bool)> { + lock_url(transaction, proxy).await?; + if let Some(id) = sqlx::query_scalar::<_, String>("select id from outbound_proxies where proxy_url = $1 order by created_at, id limit 1 for share") + .bind(proxy.expose_url()).fetch_optional(&mut **transaction).await.map_err(|_| unavailable())? { + return Ok((id, false)); + } + let id = format!("proxy_{}", uuid::Uuid::now_v7().simple()); + let generated_name = format!("Imported proxy {}", &id[id.len() - 8..]); + sqlx::query("insert into outbound_proxies (id, name, proxy_url) values ($1, $2, $3)") + .bind(&id) + .bind(name.unwrap_or(&generated_name)) + .bind(proxy.expose_url()) + .execute(&mut **transaction) + .await + .map_err(|_| unavailable())?; + Ok((id, true)) +} + +pub(crate) async fn resolve_proxy_selection( + transaction: &mut Transaction<'_, Postgres>, + selection: &AccountProxySelection, +) -> StoreResult<(Option, Option)> { + match selection { + AccountProxySelection::Direct => Ok((None, None)), + AccountProxySelection::Url(proxy) => { + let (id, _) = ensure_proxy_for_url(transaction, proxy, None).await?; + Ok((Some(id), Some(proxy.clone()))) + } + AccountProxySelection::Saved(id) => { + let (value, tested) = sqlx::query_as::<_, (String, Option)>( + "select proxy_url, last_test_success from outbound_proxies where id = $1 for share", + ) + .bind(id) + .fetch_optional(&mut **transaction) + .await + .map_err(|_| unavailable())? + .ok_or_else(|| StoreError::NotFound { + entity: ENTITY, + id: id.clone(), + })?; + if tested != Some(true) { + return Err(conflict(id)); + } + Ok(( + Some(id.clone()), + Some(OutboundProxy::parse(&value).map_err(|_| invalid())?), + )) + } + } +} + +async fn audit( + transaction: &mut Transaction<'_, Postgres>, + context: &MutationContext, + action: &str, + id: &str, + fields: &[&str], + revision: Revision, +) -> StoreResult<()> { + append_admin_audit_event_in_transaction( + transaction, + mutation_audit( + context, + action, + "outbound_proxy", + id, + fields.iter().map(|value| (*value).to_owned()).collect(), + ), + revision, + ) + .await +} + +#[async_trait] +impl ProxyStore for PgProxyRepository { + async fn list(&self, query: ProxyListQuery) -> AdminStoreResult { + if query.page == 0 { + return Err(store_error(invalid())); + } + let total: i64 = sqlx::query_scalar( + "select count(*) from outbound_proxies where strpos(lower(name), lower($1)) > 0", + ) + .bind(&query.search) + .fetch_one(&self.pool) + .await + .map_err(|_| store_error(unavailable()))?; + let mut builder = QueryBuilder::::new(SELECT); + builder + .push(" where strpos(lower(p.name), lower(") + .push_bind(&query.search) + .push(")) > 0 order by p.created_at desc, p.id limit ") + .push_bind(i64::from(query.page_size.get())) + .push(" offset ") + .push_bind(i64::from(query.page - 1) * i64::from(query.page_size.get())); + let items = builder + .build() + .fetch_all(&self.pool) + .await + .map_err(|_| store_error(unavailable()))? + .into_iter() + .map(record) + .collect::>>() + .map_err(store_error)?; + Ok(ProxyPage { + items, + total: u64::try_from(total).map_err(|_| store_error(invalid()))?, + page: query.page, + page_size: query.page_size.get(), + }) + } + + async fn get(&self, id: &str) -> AdminStoreResult { + let mut builder = QueryBuilder::::new(SELECT); + builder.push(" where p.id = ").push_bind(id); + let row = builder + .build() + .fetch_optional(&self.pool) + .await + .map_err(|_| store_error(unavailable()))? + .ok_or_else(|| { + store_error(StoreError::NotFound { + entity: ENTITY, + id: id.to_owned(), + }) + })?; + record(row).map_err(store_error) + } + + async fn create( + &self, + command: NewProxy, + context: &MutationContext, + ) -> AdminStoreResult { + let mut transaction = self + .pool + .begin() + .await + .map_err(|_| store_error(unavailable()))?; + let revision = bump_config_revision_in_transaction(&mut transaction) + .await + .map_err(store_error)?; + let (id, created) = + ensure_proxy_for_url(&mut transaction, &command.proxy, Some(&command.name)) + .await + .map_err(store_error)?; + if !created { + return Err(store_error(conflict(&id))); + } + audit( + &mut transaction, + context, + "create", + &id, + &["name", "proxy_url"], + revision, + ) + .await + .map_err(store_error)?; + transaction + .commit() + .await + .map_err(|_| store_error(unavailable()))?; + Ok(ProxyMutation { + config_revision: admin_revision(revision)?, + record: self.get(&id).await?, + }) + } + + async fn update( + &self, + command: UpdateProxy, + context: &MutationContext, + ) -> AdminStoreResult { + let mut transaction = self + .pool + .begin() + .await + .map_err(|_| store_error(unavailable()))?; + let revision = bump_config_revision_in_transaction(&mut transaction) + .await + .map_err(store_error)?; + if let Some(proxy) = &command.proxy { + lock_url(&mut transaction, proxy) + .await + .map_err(store_error)?; + let duplicate: bool = sqlx::query_scalar( + "select exists(select 1 from outbound_proxies where proxy_url = $1 and id <> $2)", + ) + .bind(proxy.expose_url()) + .bind(&command.id) + .fetch_one(&mut *transaction) + .await + .map_err(|_| store_error(unavailable()))?; + if duplicate { + return Err(store_error(conflict(&command.id))); + } + } + let changed = sqlx::query( + "update outbound_proxies set name = $3, proxy_url = coalesce($4, proxy_url), revision = revision + 1, updated_at = now(), + last_test_at = case when $4 is not null and $4 <> proxy_url then null else last_test_at end, + last_test_success = case when $4 is not null and $4 <> proxy_url then null else last_test_success end, + last_test_latency_ms = case when $4 is not null and $4 <> proxy_url then null else last_test_latency_ms end, + last_test_ip = case when $4 is not null and $4 <> proxy_url then null else last_test_ip end, + last_test_message = case when $4 is not null and $4 <> proxy_url then null else last_test_message end + where id = $1 and revision = $2") + .bind(&command.id).bind(i64::try_from(command.revision.get()).map_err(|_| store_error(invalid()))?) + .bind(&command.name).bind(command.proxy.as_ref().map(OutboundProxy::expose_url)) + .execute(&mut *transaction).await.map_err(|_| store_error(unavailable()))?; + if changed.rows_affected() != 1 { + return Err(store_error(conflict(&command.id))); + } + sqlx::query("update provider_accounts a set outbound_proxy_url = p.proxy_url, updated_at = greatest(now(), a.updated_at) from outbound_proxies p where p.id = $1 and a.outbound_proxy_id = p.id and a.outbound_proxy_url is distinct from p.proxy_url") + .bind(&command.id).execute(&mut *transaction).await.map_err(|_| store_error(unavailable()))?; + audit( + &mut transaction, + context, + "update", + &command.id, + &["name", "proxy_url"], + revision, + ) + .await + .map_err(store_error)?; + transaction + .commit() + .await + .map_err(|_| store_error(unavailable()))?; + Ok(ProxyMutation { + config_revision: admin_revision(revision)?, + record: self.get(&command.id).await?, + }) + } + + async fn delete( + &self, + id: &str, + revision: AdminRevision, + context: &MutationContext, + ) -> AdminStoreResult { + let mut transaction = self + .pool + .begin() + .await + .map_err(|_| store_error(unavailable()))?; + let config_revision = bump_config_revision_in_transaction(&mut transaction) + .await + .map_err(store_error)?; + let result = sqlx::query("delete from outbound_proxies where id = $1 and revision = $2") + .bind(id) + .bind(i64::try_from(revision.get()).map_err(|_| store_error(invalid()))?) + .execute(&mut *transaction) + .await + .map_err(|error| { + // PostgreSQL 18 distinguishes RESTRICT from other foreign-key violations. + if error.as_database_error().is_some_and(|error| { + error.is_foreign_key_violation() || error.code().as_deref() == Some("23001") + }) { + store_error(conflict(id)) + } else { + store_error(unavailable()) + } + })?; + if result.rows_affected() != 1 { + return Err(store_error(conflict(id))); + } + audit( + &mut transaction, + context, + "delete", + id, + &[], + config_revision, + ) + .await + .map_err(store_error)?; + transaction + .commit() + .await + .map_err(|_| store_error(unavailable()))?; + admin_revision(config_revision) + } + + async fn record_test( + &self, + id: &str, + revision: AdminRevision, + result: ProxyTestResult, + context: &MutationContext, + ) -> AdminStoreResult { + let mut transaction = self + .pool + .begin() + .await + .map_err(|_| store_error(unavailable()))?; + let updated = sqlx::query("update outbound_proxies set last_test_at = now(), last_test_success = $3, last_test_latency_ms = $4, last_test_ip = $5, last_test_message = $6 where id = $1 and revision = $2") + .bind(id).bind(i64::try_from(revision.get()).map_err(|_| store_error(invalid()))?) + .bind(result.success).bind(i64::try_from(result.latency_ms).map_err(|_| store_error(invalid()))?) + .bind(result.exit_ip.map(|ip| ip.to_string())).bind(result.message) + .execute(&mut *transaction).await.map_err(|_| store_error(unavailable()))?; + if updated.rows_affected() != 1 { + return Err(store_error(conflict(id))); + } + let current: i64 = + sqlx::query_scalar("select config_revision from runtime_settings where id = 1") + .fetch_one(&mut *transaction) + .await + .map_err(|_| store_error(unavailable()))?; + let current = Revision::new(u64::try_from(current).map_err(|_| store_error(invalid()))?) + .map_err(store_error)?; + audit( + &mut transaction, + context, + "test", + id, + &["last_test"], + current, + ) + .await + .map_err(store_error)?; + transaction + .commit() + .await + .map_err(|_| store_error(unavailable()))?; + self.get(id).await + } +} diff --git a/backend/crates/gateway-store/tests/postgres/mod.rs b/backend/crates/gateway-store/tests/postgres/mod.rs index 14484a076..10c8abacf 100644 --- a/backend/crates/gateway-store/tests/postgres/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/mod.rs @@ -22,6 +22,7 @@ mod health; mod observability; mod ops_events; mod provider_accounts; +mod proxies; mod query_budget; mod retention; mod runtime_settings; @@ -235,6 +236,7 @@ async fn connect_and_migrate_should_apply_all_migrations_once_and_reopen_cleanly "client_key_charge_events", "model_requests", "ops_events", + "outbound_proxies", "provider_accounts", "runtime_settings", ] diff --git a/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs b/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs index 4283d53a4..cdaa93b84 100644 --- a/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs @@ -1224,7 +1224,7 @@ async fn account_proxy_edits_preserve_credentials_and_clear_egress_without_audit store .update_account( UpdateAccount { - outbound_proxy: Some(Some( + outbound_proxy: Some(gateway_admin::model::proxies::AccountProxySelection::Url( gateway_core::account::OutboundProxy::parse( "socks5h://next:new-secret@127.0.0.1:1080", ) @@ -1240,7 +1240,7 @@ async fn account_proxy_edits_preserve_credentials_and_clear_egress_without_audit store .update_account( UpdateAccount { - outbound_proxy: Some(None), + outbound_proxy: Some(gateway_admin::model::proxies::AccountProxySelection::Direct), ..command }, &context, @@ -1546,6 +1546,7 @@ async fn admin_import_updates_the_same_verified_identity_without_rebinding_it() let imported = repository .import_provider_accounts(ImportProviderAccounts { settings: None, + outbound_proxy: None, scope: scope.clone(), accounts: vec![account("acct_admin_upsert", "user-admin-upsert")], audit: audit("audit_admin_upsert_create", "import", "acct_admin_upsert"), @@ -1569,6 +1570,7 @@ async fn admin_import_updates_the_same_verified_identity_without_rebinding_it() let imported = repository .import_provider_accounts(ImportProviderAccounts { settings: None, + outbound_proxy: None, scope: scope.clone(), accounts: vec![updated], audit: audit("audit_admin_upsert_update", "import", "acct_admin_upsert"), @@ -1604,6 +1606,7 @@ async fn admin_import_updates_the_same_verified_identity_without_rebinding_it() repository .import_provider_accounts(ImportProviderAccounts { settings: None, + outbound_proxy: None, scope, accounts: vec![account("acct_admin_upsert", "another-user")], audit: audit("audit_admin_upsert_rebind", "import", "acct_admin_upsert"), @@ -1693,6 +1696,101 @@ async fn authorization_create_returns_existing_account_id_when_identity_is_upser database.close().await; } +#[tokio::test] +async fn authorization_import_rejects_a_saved_proxy_changed_during_oauth() { + use gateway_admin::{ + model::proxies::{NewProxy, ProxyTestResult, UpdateProxy}, + ports::proxy::ProxyStore, + }; + let Some(database) = TestDatabase::create("oauth_proxy_guard").await else { + return; + }; + let proxies = gateway_store::postgres::PgProxyRepository::new(database.pool.clone()); + let context = MutationContext { + actor: MutationActor::System, + request_id: "oauth-proxy".to_owned(), + }; + let original = gateway_core::account::OutboundProxy::parse("http://127.0.0.1:8080").unwrap(); + let saved = proxies + .create( + NewProxy { + name: "OAuth".to_owned(), + proxy: original.clone(), + }, + &context, + ) + .await + .unwrap() + .record; + let success = ProxyTestResult { + success: true, + latency_ms: 1, + exit_ip: Some("203.0.113.5".parse().unwrap()), + message: "Connected".to_owned(), + }; + proxies + .record_test(&saved.id, saved.revision, success.clone(), &context) + .await + .unwrap(); + let replacement = gateway_core::account::OutboundProxy::parse("http://127.0.0.1:9090").unwrap(); + let edited = proxies + .update( + UpdateProxy { + id: saved.id.clone(), + revision: saved.revision, + name: saved.name, + proxy: Some(replacement.clone()), + }, + &context, + ) + .await + .unwrap() + .record; + proxies + .record_test(&edited.id, edited.revision, success, &context) + .await + .unwrap(); + let repository = PgProviderAccountRepository::new(database.pool.clone()); + let mut candidate = account("acct_oauth_proxy", "oauth-user"); + candidate.outbound_proxy = Some(original); + let command = |candidate: NewProviderAccount| ImportProviderAccounts { + settings: None, + outbound_proxy: Some(gateway_admin::model::proxies::ImportProxyBinding { + id: saved.id.clone(), + proxy: candidate.outbound_proxy.clone().unwrap(), + }), + scope: ProviderAccountAdminScope { + provider_kind: "openai".to_owned(), + }, + accounts: vec![candidate], + audit: audit("audit_oauth_proxy", "authorize", "acct_oauth_proxy"), + }; + assert!( + repository + .import_provider_accounts(command(candidate.clone())) + .await + .is_err() + ); + assert_eq!(account_count(&database.pool, "acct_oauth_proxy").await, 0); + candidate.outbound_proxy = Some(replacement); + repository + .import_provider_accounts(command(candidate)) + .await + .unwrap(); + assert_eq!( + proxies.get(&saved.id).await.unwrap().accounts[0].id, + "acct_oauth_proxy" + ); + assert_eq!( + sqlx::query_scalar::<_, i64>("select count(*) from outbound_proxies") + .fetch_one(&database.pool) + .await + .unwrap(), + 1 + ); + database.close().await; +} + #[tokio::test] async fn core_refresh_cas_updates_profile_and_credential_under_one_revision() { let Some(database) = TestDatabase::create("provider_account_core_refresh").await else { @@ -1884,6 +1982,7 @@ async fn provider_account_admin_mutations_are_scoped_audited_and_atomic() { let imported = repository .import_provider_accounts(ImportProviderAccounts { settings: None, + outbound_proxy: None, scope: scope.clone(), accounts: vec![ account("acct_admin_a", "user-admin-a"), @@ -1912,6 +2011,7 @@ async fn provider_account_admin_mutations_are_scoped_audited_and_atomic() { repository .import_provider_accounts(ImportProviderAccounts { settings: None, + outbound_proxy: None, scope: scope.clone(), accounts: vec![ account("acct_admin_transient", "user-admin-transient"), @@ -2101,6 +2201,7 @@ async fn provider_account_import_and_reauthorization_preserve_existing_membershi let imported = repository .import_provider_accounts(ImportProviderAccounts { settings: None, + outbound_proxy: None, scope: scope.clone(), accounts: vec![account("acct_grouped_import", "user-grouped-import")], audit: audit("audit_grouped_import", "import", "acct_grouped_import"), @@ -2135,6 +2236,7 @@ async fn provider_account_import_and_reauthorization_preserve_existing_membershi repository .import_provider_accounts(ImportProviderAccounts { settings: None, + outbound_proxy: None, scope: scope.clone(), accounts: vec![account("acct_reimport_candidate", "user-grouped-import")], audit: audit("audit_grouped_reimport", "import", "acct_grouped_import"), @@ -2203,6 +2305,7 @@ async fn verified_credential_rotation_preserves_quota_exhaustion() { repository .import_provider_accounts(ImportProviderAccounts { settings: None, + outbound_proxy: None, scope: scope.clone(), accounts: vec![account("acct_rotation_quota", "user-rotation-quota")], audit: audit( @@ -2345,6 +2448,7 @@ async fn disabled_account_preserves_user_state_during_refresh_writes() { repository .import_provider_accounts(ImportProviderAccounts { settings: None, + outbound_proxy: None, scope: scope.clone(), accounts: vec![disabled], audit: audit( @@ -2398,7 +2502,7 @@ async fn disabled_account_preserves_user_state_during_refresh_writes() { database.close().await; } -fn account(id: &str, upstream_user_id: &str) -> NewProviderAccount { +pub(super) fn account(id: &str, upstream_user_id: &str) -> NewProviderAccount { NewProviderAccount { outbound_proxy: None, id: id.to_owned(), @@ -2604,7 +2708,7 @@ async fn proxy_edit_preserves_an_inflight_token_refresh() { concurrency_limit: None, weight: gateway_core::account::AccountWeight::DEFAULT, group_ids: vec![], - outbound_proxy: Some(Some( + outbound_proxy: Some(gateway_admin::model::proxies::AccountProxySelection::Url( gateway_core::account::OutboundProxy::parse("http://127.0.0.1:18080").unwrap(), )), }, @@ -2663,6 +2767,7 @@ async fn account_import_settings_apply_atomically_to_new_and_existing_identities let result = repository .import_provider_accounts(ImportProviderAccounts { settings: Some(settings.clone()), + outbound_proxy: None, scope: ProviderAccountAdminScope { provider_kind: "openai".to_owned(), }, @@ -2693,6 +2798,7 @@ async fn account_import_settings_apply_atomically_to_new_and_existing_identities .expect("existing account"); let failed = repository .import_provider_accounts(ImportProviderAccounts { + outbound_proxy: None, settings: Some(AccountImportSettings { group_ids: vec![ AccountGroupId::new("grp_00000000000000000000000000000092") diff --git a/backend/crates/gateway-store/tests/postgres/proxies.rs b/backend/crates/gateway-store/tests/postgres/proxies.rs new file mode 100644 index 000000000..d6eed5c49 --- /dev/null +++ b/backend/crates/gateway-store/tests/postgres/proxies.rs @@ -0,0 +1,308 @@ +use gateway_admin::{ + model::{ + MutationActor, MutationContext, PageSize, + accounts::{BatchUpdateAccounts, UpdateAccount}, + proxies::*, + }, + ports::{ + proxy::ProxyStore, + store::{AccountStore, AdminStoreErrorKind}, + }, +}; +use gateway_core::account::{AccountWeight, OutboundProxy}; +use gateway_store::postgres::{ + PgProviderAccountRepository, PgProxyRepository, ProviderAccountRepository, +}; + +use super::{TestDatabase, admin_account_store, provider_accounts::account}; + +fn context() -> MutationContext { + MutationContext { + actor: MutationActor::System, + request_id: "managed-proxy-test".to_owned(), + } +} + +fn success() -> ProxyTestResult { + ProxyTestResult { + success: true, + latency_ms: 10, + exit_ip: Some("203.0.113.5".parse().unwrap()), + message: "Connected".to_owned(), + } +} + +fn update(account_id: &str, selection: AccountProxySelection) -> UpdateAccount { + UpdateAccount { + account_id: account_id.to_owned(), + enabled: true, + concurrency_limit: None, + weight: AccountWeight::DEFAULT, + group_ids: vec![], + outbound_proxy: Some(selection), + } +} + +#[tokio::test] +async fn managed_proxies_persist_bind_update_all_accounts_and_protect_stale_tests() { + let Some(database) = TestDatabase::create("managed_proxies").await else { + return; + }; + let accounts = PgProviderAccountRepository::new(database.pool.clone()); + for id in ["acct_one", "acct_two"] { + accounts + .insert_provider_account(account(id, id)) + .await + .unwrap(); + } + let store = PgProxyRepository::new(database.pool.clone()); + let admin = admin_account_store(&database.pool); + let context = context(); + let old_proxy = OutboundProxy::parse("http://user:secret@127.0.0.1:8080").unwrap(); + let created = store + .create( + NewProxy { + name: "Office".to_owned(), + proxy: old_proxy.clone(), + }, + &context, + ) + .await + .unwrap() + .record; + assert_eq!(store.get(&created.id).await.unwrap().proxy, old_proxy); + assert!( + store + .create( + NewProxy { + name: "Duplicate".to_owned(), + proxy: old_proxy.clone() + }, + &context + ) + .await + .is_err() + ); + let selection = AccountProxySelection::Saved(created.id.clone()); + assert!( + admin + .update_account(update("acct_one", selection.clone()), &context) + .await + .is_err() + ); + let read = || async { + sqlx::query_as::<_, (String, Option, Option, i64)>("select id, outbound_proxy_id, outbound_proxy_url, credential_revision from provider_accounts order by id") + .fetch_all(&database.pool).await.unwrap() + }; + assert!( + read() + .await + .iter() + .all(|row| row.1.is_none() && row.2.is_none() && row.3 == 1) + ); + let tested = store + .record_test(&created.id, created.revision, success(), &context) + .await + .unwrap(); + assert!(tested.last_test_at.is_some()); + for id in ["acct_one", "acct_two"] { + admin + .update_account(update(id, selection.clone()), &context) + .await + .unwrap(); + } + assert_eq!(store.get(&created.id).await.unwrap().accounts.len(), 2); + assert_eq!( + store + .delete(&created.id, created.revision, &context) + .await + .unwrap_err() + .kind(), + AdminStoreErrorKind::Conflict + ); + assert!( + read() + .await + .iter() + .all(|row| row.1.as_deref() == Some(created.id.as_str()) + && row.2.as_deref() == Some(old_proxy.expose_url()) + && row.3 == 1) + ); + + let renamed = store + .update( + UpdateProxy { + id: created.id.clone(), + revision: created.revision, + name: "Renamed".to_owned(), + proxy: None, + }, + &context, + ) + .await + .unwrap() + .record; + assert_eq!(renamed.proxy, old_proxy); + assert_eq!(renamed.last_test, Some(success())); + assert!(read().await.iter().all(|row| row.3 == 1)); + let new_proxy = OutboundProxy::parse("socks5h://next:new-secret@127.0.0.1:1080").unwrap(); + let edited = store + .update( + UpdateProxy { + id: created.id.clone(), + revision: renamed.revision, + name: renamed.name, + proxy: Some(new_proxy.clone()), + }, + &context, + ) + .await + .unwrap() + .record; + assert!(edited.last_test.is_none()); + assert!(edited.last_test_at.is_none()); + assert!( + read() + .await + .iter() + .all(|row| row.2.as_deref() == Some(new_proxy.expose_url()) && row.3 == 1) + ); + assert!( + store + .record_test(&created.id, created.revision, success(), &context) + .await + .is_err() + ); + assert!(store.get(&created.id).await.unwrap().last_test.is_none()); + store + .record_test(&created.id, edited.revision, success(), &context) + .await + .unwrap(); + + for id in ["acct_one", "acct_two"] { + admin + .update_account(update(id, AccountProxySelection::Direct), &context) + .await + .unwrap(); + } + assert!( + read() + .await + .iter() + .all(|row| row.1.is_none() && row.2.is_none() && row.3 == 1) + ); + store + .delete(&created.id, edited.revision, &context) + .await + .unwrap(); + assert!(store.get(&created.id).await.is_err()); + let audits: Vec = + sqlx::query_scalar("select to_jsonb(a) from admin_audit_events a") + .fetch_all(&database.pool) + .await + .unwrap(); + assert!(!serde_json::to_string(&audits).unwrap().contains("secret")); + database.close().await; +} + +#[tokio::test] +async fn legacy_urls_join_one_catalog_entry_and_invalid_batch_rolls_back() { + let Some(database) = TestDatabase::create("proxy_catalog_legacy").await else { + return; + }; + let accounts = PgProviderAccountRepository::new(database.pool.clone()); + let proxy = OutboundProxy::parse("http://user:secret@127.0.0.1:8080").unwrap(); + for id in ["acct_one", "acct_two"] { + let mut seed = account(id, id); + seed.outbound_proxy = Some(proxy.clone()); + accounts.insert_provider_account(seed).await.unwrap(); + } + let store = PgProxyRepository::new(database.pool.clone()); + let page = store + .list(ProxyListQuery { + page: 1, + page_size: PageSize::new(20).unwrap(), + search: String::new(), + }) + .await + .unwrap(); + assert_eq!(page.total, 1); + assert_eq!(page.items[0].accounts.len(), 2); + let admin = admin_account_store(&database.pool); + assert!( + admin + .batch_update_accounts( + BatchUpdateAccounts { + account_ids: vec!["acct_one".to_owned(), "acct_missing".to_owned()], + enabled: false, + concurrency_limit: None, + weight: AccountWeight::DEFAULT, + group_ids: vec![], + outbound_proxy: Some(AccountProxySelection::Url( + OutboundProxy::parse("http://127.0.0.1:9090").unwrap() + )), + }, + &context() + ) + .await + .is_err() + ); + assert_eq!( + sqlx::query_scalar::<_, i64>("select count(*) from outbound_proxies") + .fetch_one(&database.pool) + .await + .unwrap(), + 1 + ); + assert_eq!( + sqlx::query_as::<_, (bool, String)>( + "select enabled, outbound_proxy_url from provider_accounts where id = 'acct_one'" + ) + .fetch_one(&database.pool) + .await + .unwrap(), + (true, proxy.expose_url().to_owned()) + ); + database.close().await; +} + +#[tokio::test] +async fn migration_backfills_shared_proxies_without_changing_credentials() { + let Some(database) = TestDatabase::create("proxy_backfill").await else { + return; + }; + let accounts = PgProviderAccountRepository::new(database.pool.clone()); + for id in ["acct_one", "acct_two", "acct_direct"] { + accounts + .insert_provider_account(account(id, id)) + .await + .unwrap(); + } + sqlx::raw_sql("alter table provider_accounts drop column outbound_proxy_id; drop table outbound_proxies; + update provider_accounts set outbound_proxy_url = 'http://user:secret@127.0.0.1:8080/' where id <> 'acct_direct';") + .execute(&database.pool).await.unwrap(); + sqlx::raw_sql(include_str!( + "../../../../migrations/0005_managed_outbound_proxies.sql" + )) + .execute(&database.pool) + .await + .unwrap(); + assert_eq!( + sqlx::query_scalar::<_, i64>("select count(*) from outbound_proxies") + .fetch_one(&database.pool) + .await + .unwrap(), + 1 + ); + assert_eq!(sqlx::query_scalar::<_, i64>("select count(*) from provider_accounts a join outbound_proxies p on a.outbound_proxy_id = p.id where a.outbound_proxy_url = p.proxy_url and a.credential_revision = 1").fetch_one(&database.pool).await.unwrap(), 2); + assert!( + sqlx::query_scalar::<_, Option>( + "select outbound_proxy_id from provider_accounts where id = 'acct_direct'" + ) + .fetch_one(&database.pool) + .await + .unwrap() + .is_none() + ); + database.close().await; +} diff --git a/backend/crates/providers/openai/src/admin.rs b/backend/crates/providers/openai/src/admin.rs index 9614986b8..0a57d313f 100644 --- a/backend/crates/providers/openai/src/admin.rs +++ b/backend/crates/providers/openai/src/admin.rs @@ -291,9 +291,10 @@ impl ProviderAdmin for OpenAiAdminProvider { ) -> Result { let prepared = self .credentials - .prepare_import_document(Value::Object( - command.document.into_provider_data().into_inner(), - )) + .prepare_import_document_with_proxy( + Value::Object(command.document.into_provider_data().into_inner()), + command.default_outbound_proxy.as_ref(), + ) .await .inspect_err(|error| { log_import_failure("prepare_document", credential_admin_error_code(error)); diff --git a/backend/crates/providers/openai/src/credential/admin.rs b/backend/crates/providers/openai/src/credential/admin.rs index 6b25284c1..0ac823dc4 100644 --- a/backend/crates/providers/openai/src/credential/admin.rs +++ b/backend/crates/providers/openai/src/credential/admin.rs @@ -824,6 +824,14 @@ impl CodexCredentialAdminService { pub async fn prepare_import_document( &self, payload: Value, + ) -> Result { + self.prepare_import_document_with_proxy(payload, None).await + } + + pub async fn prepare_import_document_with_proxy( + &self, + payload: Value, + default_proxy: Option<&gateway_core::account::OutboundProxy>, ) -> Result { if serde_json::to_vec(&payload) .map_err(|_| CodexCredentialAdminError::InvalidInput)? @@ -832,7 +840,7 @@ impl CodexCredentialAdminService { { return Err(CodexCredentialAdminError::InvalidInput); } - let candidates = parse_import_document(&payload)?; + let candidates = parse_import_document(&payload, default_proxy)?; if candidates.is_empty() || candidates.len() > MAX_BATCH { return Err(CodexCredentialAdminError::InvalidInput); } @@ -1041,6 +1049,7 @@ fn log_manual_refresh_failure(account_id: &ProviderAccountId, error: &RefreshFai fn parse_import_document( payload: &Value, + default_proxy: Option<&gateway_core::account::OutboundProxy>, ) -> Result, CodexCredentialAdminError> { let payload = payload .get("data") @@ -1052,7 +1061,14 @@ fn parse_import_document( continue; } let mut account = parse_oauth_import_account(value)?; - account.outbound_proxy = import_proxy(payload, value)?; + account.outbound_proxy = if ["outboundProxyUrl", "outbound_proxy_url", "proxy_key"] + .iter() + .any(|field| value.get(field).is_some()) + { + import_proxy(payload, value)? + } else { + default_proxy.cloned() + }; accounts.push(account); } Ok(accounts) diff --git a/backend/crates/providers/openai/src/transport/client.rs b/backend/crates/providers/openai/src/transport/client.rs index eacc51a12..92ed9ec19 100644 --- a/backend/crates/providers/openai/src/transport/client.rs +++ b/backend/crates/providers/openai/src/transport/client.rs @@ -820,6 +820,12 @@ pub(super) async fn read_error_response_body( pub(super) fn websocket_upstream_request(request: &CodexResponsesRequest) -> CodexResponsesRequest { let mut request = request.clone(); + // Upstream execution is streaming; downstream delivery keeps the original preference. + if !request.stream() { + request + .body_mut() + .insert("stream".to_owned(), Value::Bool(true)); + } project_websocket_client_metadata(&mut request); request } diff --git a/backend/crates/providers/openai/tests/admin.rs b/backend/crates/providers/openai/tests/admin.rs index 8e61ba2bd..ed4fdbe14 100644 --- a/backend/crates/providers/openai/tests/admin.rs +++ b/backend/crates/providers/openai/tests/admin.rs @@ -984,6 +984,7 @@ async fn openai_admin_provider_rejects_unprepared_mutations_before_store_commit( let admin = bundle.admin_provider(); let import_error = admin .prepare_import(PrepareCredentialImport { + default_outbound_proxy: None, document: ProviderDocument::new(OpaqueProviderData::new(Map::new())), }) .await @@ -1025,6 +1026,7 @@ async fn initialized_provider_reports_a_safe_pat_format_error_before_network_acc let error = bundle .admin_provider() .prepare_import(PrepareCredentialImport { + default_outbound_proxy: None, document: ProviderDocument::new(OpaqueProviderData::new(Map::from_iter([( "accessToken".to_owned(), json!("at-sensitive-token with whitespace"), diff --git a/backend/crates/providers/openai/tests/credential/admin.rs b/backend/crates/providers/openai/tests/credential/admin.rs index fcb8a661b..2cdd68d05 100644 --- a/backend/crates/providers/openai/tests/credential/admin.rs +++ b/backend/crates/providers/openai/tests/credential/admin.rs @@ -56,6 +56,54 @@ async fn sub2api_import_resolves_distinct_proxy_bindings_and_encodes_credentials assert!(!format!("{prepared:?}").contains("p:a/ss")); } +#[tokio::test] +async fn import_default_proxy_preserves_explicit_account_exits() { + let service = CodexCredentialAdminService::new( + Arc::new(UnusedRefresher), + Arc::new(TestLeaseCoordinator::default()), + runtime_policy(), + ); + let default_proxy = + gateway_core::account::OutboundProxy::parse("http://default.example:8080").unwrap(); + for (field, value, expected) in [ + ( + None, + serde_json::Value::Null, + Some("http://default.example:8080/"), + ), + ( + Some("outboundProxyUrl"), + serde_json::json!("socks5h://override.example:1080"), + Some("socks5h://override.example:1080"), + ), + (Some("outbound_proxy_url"), serde_json::json!(""), None), + (Some("outboundProxyUrl"), serde_json::Value::Null, None), + (Some("proxy_key"), serde_json::Value::Null, None), + ] { + let mut account = serde_json::json!({"access_token": test_jwt(serde_json::json!({ + "https://api.openai.com/auth": {"chatgpt_user_id": "default-proxy-user"} + }))}); + if let Some(field) = field { + account[field] = value; + } + let prepared = service + .prepare_import_document_with_proxy( + serde_json::json!({"data": {"accounts": [account]}}), + Some(&default_proxy), + ) + .await + .unwrap(); + assert_eq!( + prepared.accounts()[0] + .account + .outbound_proxy() + .map(|proxy| proxy.endpoint()) + .as_deref(), + expected + ); + } +} + #[tokio::test] async fn sub2api_invalid_proxy_bindings_are_rejected_before_any_token_refresh() { let service = CodexCredentialAdminService::new( @@ -746,7 +794,7 @@ async fn oauth_import_rejects_legacy_bare_token_field() { #[tokio::test] async fn pat_import_uses_account_proxy_without_direct_fallback() { - for status in [200, 503] { + for (status, use_default) in [(200, false), (503, false), (200, true), (503, true)] { let origin = MockServer::start().await; let proxy = MockServer::start().await; Mock::given(method("GET")) @@ -756,11 +804,13 @@ async fn pat_import_uses_account_proxy_without_direct_fallback() { .expect(1) .mount(&proxy) .await; + let default_proxy = gateway_core::account::OutboundProxy::parse(&proxy.uri()).unwrap(); + let mut document = serde_json::json!({"accessToken": "at-proxy-token"}); + if !use_default { + document["outboundProxyUrl"] = serde_json::json!(proxy.uri()); + } let imported = pat_service(&origin) - .prepare_import_document(serde_json::json!({ - "accessToken": "at-proxy-token", - "outboundProxyUrl": proxy.uri(), - })) + .prepare_import_document_with_proxy(document, use_default.then_some(&default_proxy)) .await; if status == 200 { let prepared = imported.expect("PAT validated through account proxy"); diff --git a/backend/crates/providers/openai/tests/transport/websocket/mod.rs b/backend/crates/providers/openai/tests/transport/websocket/mod.rs index ea738bf2b..800f7c6ff 100644 --- a/backend/crates/providers/openai/tests/transport/websocket/mod.rs +++ b/backend/crates/providers/openai/tests/transport/websocket/mod.rs @@ -193,6 +193,74 @@ fn websocket_connection_should_prepare_response_create_payload_text() { assert!(payload.get("store").is_none()); } +#[tokio::test] +async fn backend_websocket_should_stream_upstream_for_non_streaming_client_request() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let mut websocket = accept_codex_test_websocket(stream).await; + let Some(Ok(Message::Text(payload))) = websocket.next().await else { + panic!("client should send response.create"); + }; + let payload: Value = serde_json::from_str(payload.as_str()).unwrap(); + websocket + .send(Message::Text( + json!({ + "type": "response.completed", + "response": { + "id": "resp_non_streaming", + "object": "response", + "status": "completed", + "output": [], + "usage": {"input_tokens": 3, "output_tokens": 1, "total_tokens": 4} + } + }) + .to_string() + .into(), + )) + .await + .unwrap(); + payload + }); + let request = CodexResponsesRequest::from_body( + json!({ + "model": "gpt-test", + "instructions": "collect this response", + "input": [], + "stream": false, + "previous_response_id": "resp_previous" + }) + .as_object() + .unwrap() + .clone(), + ); + let backend = CodexBackendClient::new( + reqwest::Client::builder().no_proxy().build().unwrap(), + format!("http://{address}"), + test_wire_profile(), + ); + let response = timeout( + Duration::from_secs(5), + backend.create_response( + &request, + request_context("req_non_streaming_ws", Some("acct-non-streaming")), + ), + ) + .await + .unwrap() + .expect("upstream WebSocket response"); + let payload = server.await.unwrap(); + assert_eq!(response.transport, CodexBackendTransport::WebSocket); + assert!(response.body.contains("resp_non_streaming")); + assert_eq!(payload["stream"], true); + assert_eq!(payload["previous_response_id"], "resp_previous"); + assert!( + !request.stream(), + "downstream delivery remains non-streaming" + ); +} + #[test] fn websocket_connection_should_prepare_capture_payload_with_canonical_field_order() { let mut request = codex_request_with_prompt_cache_key( diff --git a/backend/crates/providers/xai/src/admin.rs b/backend/crates/providers/xai/src/admin.rs index 961ab2c6f..54d7fdce3 100644 --- a/backend/crates/providers/xai/src/admin.rs +++ b/backend/crates/providers/xai/src/admin.rs @@ -429,8 +429,11 @@ impl ProviderAdmin for XaiAdminProvider { command.document.into_provider_data().into_inner(), )) .map_err(|_| provider_error(ProviderAdminErrorKind::Invalid))?; - let document = GrokOAuthImportDocument::parse_json(&document) - .map_err(|_| provider_error(ProviderAdminErrorKind::Invalid))?; + let document = GrokOAuthImportDocument::parse_json_with_proxy( + &document, + command.default_outbound_proxy.as_ref(), + ) + .map_err(|_| provider_error(ProviderAdminErrorKind::Invalid))?; let mut subjects = BTreeSet::new(); let mut credentials = Vec::new(); // 逐条目独立成败：verify 会真实轮换 RT，后续条目的任何失败都不得 diff --git a/backend/crates/providers/xai/src/credential/import.rs b/backend/crates/providers/xai/src/credential/import.rs index 8794603f0..4c698191a 100644 --- a/backend/crates/providers/xai/src/credential/import.rs +++ b/backend/crates/providers/xai/src/credential/import.rs @@ -162,6 +162,13 @@ impl GrokOAuthImportDocument { /// 独立代理 URL 随账号导入；其他展示 metadata 不参与认证。 /// 实际 token 仍须通过官方 refresh/user-info 验证；API Key 不能混入 OAuth 条目。 pub fn parse_json(document: &[u8]) -> Result { + Self::parse_json_with_proxy(document, None) + } + + pub fn parse_json_with_proxy( + document: &[u8], + default_proxy: Option<&gateway_core::account::OutboundProxy>, + ) -> Result { if document.is_empty() || document.len() > MAX_IMPORT_DOCUMENT_BYTES { return Err(GrokOAuthImportError::InvalidField("document")); } @@ -174,7 +181,13 @@ impl GrokOAuthImportDocument { let now = Utc::now(); let mut entries = Vec::with_capacity(accounts.len()); for (index, account) in accounts.into_iter().enumerate() { - if let Some(entry) = parse_account_entry(account, index, now)? { + if let Some(mut entry) = parse_account_entry(account, index, now)? { + if !["outboundProxyUrl", "outbound_proxy_url", "proxy_key"] + .iter() + .any(|field| account.get(field).is_some()) + { + entry.outbound_proxy = default_proxy.cloned(); + } entries.push(entry); } } diff --git a/backend/crates/providers/xai/tests/admin.rs b/backend/crates/providers/xai/tests/admin.rs index bbfa3f616..a76d8aaf6 100644 --- a/backend/crates/providers/xai/tests/admin.rs +++ b/backend/crates/providers/xai/tests/admin.rs @@ -413,6 +413,7 @@ async fn xai_admin_provider_rejects_unprepared_mutations_before_store_commit() { let import_error = admin .prepare_import(PrepareCredentialImport { + default_outbound_proxy: None, document: ProviderDocument::new(OpaqueProviderData::new(Map::new())), }) .await diff --git a/backend/crates/providers/xai/tests/credential/import.rs b/backend/crates/providers/xai/tests/credential/import.rs index 0f32ad5b3..f6d0c70fd 100644 --- a/backend/crates/providers/xai/tests/credential/import.rs +++ b/backend/crates/providers/xai/tests/credential/import.rs @@ -204,6 +204,45 @@ fn strict_oauth_account_document_should_parse_without_exposing_credentials() { let _candidate = entry.into_candidate(); } +#[test] +fn import_default_proxy_preserves_explicit_account_exits() { + let default_proxy = + gateway_core::account::OutboundProxy::parse("http://default.example:8080").unwrap(); + for (field, value, expected) in [ + ( + None, + serde_json::Value::Null, + Some("http://default.example:8080/"), + ), + ( + Some("outboundProxyUrl"), + serde_json::json!("socks5h://override.example:1080"), + Some("socks5h://override.example:1080"), + ), + (Some("outbound_proxy_url"), serde_json::json!(""), None), + (Some("outboundProxyUrl"), serde_json::Value::Null, None), + (Some("proxy_key"), serde_json::Value::Null, None), + ] { + let mut document: serde_json::Value = + serde_json::from_slice(&oauth_account_document()).unwrap(); + if let Some(field) = field { + document["accounts"][0][field] = value; + } + let parsed = GrokOAuthImportDocument::parse_json_with_proxy( + &serde_json::to_vec(&document).unwrap(), + Some(&default_proxy), + ) + .unwrap(); + assert_eq!( + parsed.into_entries()[0] + .outbound_proxy() + .map(|proxy| proxy.endpoint()) + .as_deref(), + expected + ); + } +} + #[test] fn oauth_account_document_should_reject_api_key_credential_field() { let mut document: serde_json::Value = diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index ce50996c3..26cb42d5b 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -2,3 +2,4 @@ b7d3df7b48d835a00a3b74cbefc958a039d38fea5529cdad04bfbc703dca72aa 0001_initial.s 78c427c17d8b5759405f1629a80ceb624012cb6b31371b762078b000266d86c6 0002_client_key_budgets.sql 46f493ff410cf77550114542e0418475876ee2b2c975436290f0f72440e098c4 0003_account_outbound_proxy.sql 5e4bfd0b47a4de5e99b7ae1d89e0294de61856a40fba3480572db765147c3b3a 0004_remove_client_charge_reconciliation.sql +2741fabb6ec23e7a19962e364ffdc054f81584c868e24453e33644b96ccca19b 0005_managed_outbound_proxies.sql diff --git a/backend/migrations/0005_managed_outbound_proxies.sql b/backend/migrations/0005_managed_outbound_proxies.sql new file mode 100644 index 000000000..a1da877d7 --- /dev/null +++ b/backend/migrations/0005_managed_outbound_proxies.sql @@ -0,0 +1,26 @@ +create table outbound_proxies ( + id text primary key, + name text not null check (char_length(name) between 1 and 100), + proxy_url text not null, + revision bigint not null default 1 check (revision > 0), + last_test_at timestamptz, + last_test_success boolean, + last_test_latency_ms bigint, + last_test_ip text, + last_test_message text, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now() +); + +alter table provider_accounts add column outbound_proxy_id text + references outbound_proxies(id) on delete restrict; +create index provider_accounts_outbound_proxy_id_idx on provider_accounts(outbound_proxy_id); + +insert into outbound_proxies (id, name, proxy_url) +select 'proxy_' || gen_random_uuid()::text, + 'Imported proxy ' || row_number() over (order by proxy_url), proxy_url +from (select distinct outbound_proxy_url as proxy_url from provider_accounts + where outbound_proxy_url is not null) existing; + +update provider_accounts a set outbound_proxy_id = p.id +from outbound_proxies p where p.proxy_url = a.outbound_proxy_url; diff --git a/docs/api.md b/docs/api.md index d806c3511..a1c228919 100644 --- a/docs/api.md +++ b/docs/api.md @@ -171,12 +171,12 @@ Responses wire 之间的协议转换层，转换只在 xAI Provider 内完成。 | `GET` | `/api/admin/accounts` | `page`、`pageSize`、`provider`、`groupId`、`search`、`status`、排序字段 | 分页查询账号与汇总 | | `GET` | `/api/admin/accounts/detail` | `accountId` | 查询账号详情、额度和本地用量 | | `GET` | `/api/admin/accounts/export` | `accountIds`、`confirm=export_sensitive_accounts` | 显式导出最多 200 个账号的敏感 Provider 文档 | -| `POST` | `/api/admin/accounts/import` | `{ provider, data, settings? }` | 导入或按上游身份更新账号，可同时应用调度与分组设置 | +| `POST` | `/api/admin/accounts/import` | `{ provider, data, settings?, outboundProxyId? }` | 导入或按上游身份更新账号，可同时应用调度、分组设置与默认代理 | | `POST` | `/api/admin/accounts/refresh` | `{ accountId }` | 手工刷新 OAuth credential（`idToken` / `accessToken` / `refreshToken`），不刷新额度 | | `POST` | `/api/admin/accounts/recover` | `{ accountId }` | 管理员显式清除该账号的本地错误/额度/cooldown 事实并重新启用，不访问上游 | | `POST` | `/api/admin/accounts/rotate` | OpenAI rotation 字段 | 手工替换 OpenAI OAuth token | -| `POST` | `/api/admin/accounts/update` | `{ accountId, enabled, concurrencyLimit, weight, groupIds, outboundProxyUrl? }` | 一次更新账号调度状态、并发上限（`null` 表示继承运行参数）、权重（1–100）、所属分组与出站代理 | -| `POST` | `/api/admin/accounts/batch-update` | `{ accountIds, enabled, concurrencyLimit, weight, groupIds, outboundProxyUrl? }` | 一次事务统一更新所选账号的调度字段、完整分组集合与可选代理 | +| `POST` | `/api/admin/accounts/update` | `{ accountId, enabled, concurrencyLimit, weight, groupIds, outboundProxyId?, outboundProxyUrl? }` | 一次更新账号调度状态、并发上限（`null` 表示继承运行参数）、权重（1–100）、所属分组与出站代理 | +| `POST` | `/api/admin/accounts/batch-update` | `{ accountIds, enabled, concurrencyLimit, weight, groupIds, outboundProxyId?, outboundProxyUrl? }` | 一次事务统一更新所选账号的调度字段、完整分组集合与可选代理 | | `POST` | `/api/admin/accounts/delete` | `{ provider, accountIds }` | 批量删除 1–200 个账号 | | `GET` | `/api/admin/accounts/quota` | `accountId` | 读取当前额度，不强制访问上游 | | `POST` | `/api/admin/accounts/quota/refresh` | `{ accountId }` | 访问 Provider 并刷新额度，同时同步额度所属状态 | @@ -186,7 +186,7 @@ Responses wire 之间的协议转换层，转换只在 xAI Provider 内完成。 | `GET` | `/api/admin/accounts/models` | `accountId` | 优先读取该 Provider + 套餐的模型 cache，缺失时有限实时拉取 | | `POST` | `/api/admin/accounts/models/refresh` | `{ accountId }` | 强制拉取最新模型并覆盖 cache | | `GET` | `/api/admin/accounts/connection-test` | `accountId`、`modelId` | 通过 SSE 返回实时连接测试事件，不作为业务 Responses 用量记录 | -| `POST` | `/api/admin/accounts/oauth/start` | `{ provider, name, accountId?, outboundProxyUrl? }` | 创建 OpenAI 或 xAI OAuth flow；`accountId` 表示重新授权 | +| `POST` | `/api/admin/accounts/oauth/start` | `{ provider, name, accountId?, outboundProxyId?, outboundProxyUrl? }` | 创建 OpenAI 或 xAI OAuth flow；`accountId` 表示重新授权 | | `POST` | `/api/admin/accounts/oauth/complete` | `{ provider, flowId, callbackUrl, settings? }` | 消费 OAuth callback；首次授权可附带账号设置，重新授权保留原设置 | 账号列表支持以下稳定值： @@ -203,13 +203,48 @@ OpenAI 的 `self_serve_business_prolite` 等 Team 套餐显示为 `Business`； 账号套餐为空或 `unknown` 时，后端优先用已保存的上游额度响应 中的明确套餐值补全 `planType` 和 `planTypeDisplay`；两处均无套餐信息时才显示“未知套餐”。 -`outboundProxyUrl` 接受 HTTP、HTTPS、SOCKS5、SOCKS5H 代理 URL，可带用户名和密码。 +`outboundProxyId` 绑定已保存且最近测试成功的代理；省略或 `null` 保留当前绑定，空字符串清除绑定。 +`outboundProxyUrl` 兼容 HTTP、HTTPS、SOCKS5、SOCKS5H 代理 URL，可带用户名和密码；不能与 ID 同时设置。 编辑时省略或 `null` 表示保持原配置，空字符串表示清除代理并直连。列表和详情只返回 不含认证信息的 `outboundProxyEndpoint`（直连时为 `null`）；只有显式敏感导出包含完整 URL。 指定代理后，推理、OAuth 服务端交换/刷新及账号辅助请求使用同一出口；代理失败不会退回直连。 浏览器打开的第三方 OAuth 授权页仍使用浏览器自身网络。 账号出口与连接隔离见 [架构说明](architecture.md#账号出站代理)。 +### 独立代理管理 / Managed Proxies + +所有端点要求管理员身份。所有响应只返回去掉认证信息的 `endpoint`，不会返回完整 URL。 +All endpoints require admin authentication and redact proxy credentials from responses. + +| 方法 / Method | 路径 / Path | 请求 / Request | 结果 / Result | +| --- | --- | --- | --- | +| `GET` | `/api/admin/proxies` | `page`、`pageSize`（1-200）、`search`（名称） | `{ items, page }` | +| `POST` | `/api/admin/proxies/create` | `{ name, proxyUrl }` | `201 { record, configRevision }` | +| `POST` | `/api/admin/proxies/update` | `{ id, revision, name, proxyUrl? }` | `{ record, configRevision }` | +| `POST` | `/api/admin/proxies/test` | `{ id, revision }` | 最新代理记录 / Proxy record with test result | +| `POST` | `/api/admin/proxies/delete` | `{ id, revision }` | `{ configRevision }` | + +`record` 包含 `id`、`name`、`endpoint`、`hasAuthentication`、`revision`、`accounts: [{ id, name }]`、 +`lastTestAt`、`lastTest: { success, latencyMs, exitIp, message }`、`createdAt`、`updatedAt`。 +未测试时 `lastTestAt` / `lastTest` 为 `null`。连通性失败返回 HTTP 200 和 `lastTest.success=false`； +记录版本过期、重复 URL、删除已绑定的代理返回 409，并发测试满载返回 429。 + +更新省略 `proxyUrl` 保留认证；连接配置改变时清除测试结果并更新所有绑定账号。 +Omit `proxyUrl` to preserve credentials. Connection changes invalidate the previous test and update all bound accounts. +Tests persist only when the requested revision still matches. Connectivity failures use HTTP 200 with +`lastTest.success=false`; stale revisions, duplicate URLs and deleting an in-use proxy return 409. +The test concurrency limit returns 429. + +测试固定经代理访问 `https://api.ipify.org?format=json`，超时 15 秒，每进程最多同时测试 4 条。 +出口测试通过不表示 Provider 账号权限或额度可用；账号可用性使用账号连接测试。 +导入请求可以携带顶层 `outboundProxyId`，在令牌交换前解析为默认出口；文件中显式的代理配置优先。 +OAuth 或导入期间若所选代理被删除、连接配置改变或测试失败，提交会拒绝过期配置。 + +Tests reach `https://api.ipify.org?format=json` through the configured proxy, with a 15-second timeout +and four concurrent tests per process. Provider access still requires the account connection test. +Imports accept a top-level `outboundProxyId` as the default exit before token exchange; explicit per-account +settings in the document take precedence. OAuth and import commits reject a deleted, changed or failed proxy. + ### 账号连接测试 SSE `GET /api/admin/accounts/connection-test` 固定探测请求指定的账号，不参与普通账号轮换。成功流沿用 @@ -468,6 +503,10 @@ HTTP 返回 `429`，`error.code` 为 `key_daily_budget_exceeded` 或 `key_weekly 自动结算按网关请求 ID 幂等执行。账本独立于使用统计日志，记录保留至删除 Key， 不受 `usageRetentionDays` 影响。 +English: Daily and weekly budgets use automatically recorded costs. Missing usage or interrupted requests +do not block a Key, and no manual reconciliation is required. New requests receive `429` once recorded +costs reach the daily or weekly limit; requests already admitted can finish above that threshold. + ## 8. 运行设置 | 方法 | 路由 | 说明 | diff --git a/docs/architecture.md b/docs/architecture.md index 4b6e63c3b..4efeb1e27 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -189,11 +189,20 @@ Core 只理解 `Operation`、能力要求、Provider 候选、稳定错误和 ca 连接池按出口隔离。修改代理推进配置 revision，使后续请求使用新出口；已执行请求可沿原连接完成。 代理变更不推进 `credential_revision`，不会使进行中的令牌刷新因凭据版本冲突而丢失结果。 -导入器在认证交换前解析并校验账号出口。sub2api 的代理引用转换为账号上的代理 URL，不创建共享代理资源； +代理独立保存和测试，通过 `outboundProxyId` 绑定账号；账号保留解析后的 URL 供 Provider 使用。 +连接配置改变时在同一事务内同步关联账号并清除旧测试结果，已绑定的代理不可删除。 +迁移 `0005_managed_outbound_proxies.sql` 将已有 URL 按完整连接信息合并为共享代理，保留认证及账号绑定。 + +导入器在认证交换前解析并校验账号出口。sub2api 的代理引用按完整 URL 登记为共享代理并绑定账号； 其 SOCKS5 配置按远端 DNS 语义转换为 SOCKS5H。账号导入只迁移账号及出口，不迁移下游 Key、余额或历史用量。 代理认证信息仅通过敏感账号导出返回，列表、详情、Debug 和普通审计不得暴露；数据库及备份按凭据保护。 输入字段、协议支持与导入校验见 [账号 API](api.md#5-账号)。 +Managed proxies store names, credentials and connectivity results independently. Accounts bind by ID and retain +the resolved URL for provider transports. Connection edits update all bound accounts atomically and invalidate +old tests; in-use proxies cannot be deleted. Migration 0005 preserves existing credentials and account bindings. +Proxy changes advance the runtime configuration revision without invalidating in-flight token refreshes. + ### Codex 原生生图与认证 管理端导出的 Codex 配置使用代理 Bearer 密钥，并声明服务端托管账号认证。 diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index 5769fca9c..514a4c0b5 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -3,7 +3,6 @@ export * from './modules/account-groups' // 账号管理 export * from './modules/accounts' - // API Keys 管理 export * from './modules/api-keys' @@ -16,6 +15,8 @@ export * from './modules/backups' // Dashboard 聚合 export * from './modules/dashboard' +export * from './modules/proxies' + // 设置管理 export * from './modules/settings' diff --git a/frontend/src/api/modules/accounts.ts b/frontend/src/api/modules/accounts.ts index 21775943e..b1aaabe11 100644 --- a/frontend/src/api/modules/accounts.ts +++ b/frontend/src/api/modules/accounts.ts @@ -291,6 +291,7 @@ interface AccountResetCreditConsumeParam extends AccountIdParam { interface AccountUpdateParam { outboundProxyUrl?: string + outboundProxyId?: string accountId: string enabled: boolean concurrencyLimit: number | null @@ -300,6 +301,7 @@ interface AccountUpdateParam { interface AccountBatchUpdateParam { outboundProxyUrl?: string + outboundProxyId?: string accountIds: string[] enabled: boolean concurrencyLimit: number | null @@ -320,6 +322,7 @@ interface AccountImportSettings { } interface AccountImportParam { + outboundProxyId?: string settings?: AccountImportSettings provider: string data: unknown @@ -327,6 +330,7 @@ interface AccountImportParam { interface AccountOAuthStartParam { outboundProxyUrl?: string + outboundProxyId?: string provider: string name: string accountId?: string diff --git a/frontend/src/api/modules/proxies.ts b/frontend/src/api/modules/proxies.ts new file mode 100644 index 000000000..9ee7d5fc9 --- /dev/null +++ b/frontend/src/api/modules/proxies.ts @@ -0,0 +1,51 @@ +import request from '../request' + +export interface OutboundProxyTest { + success: boolean + latencyMs: number + exitIp: string | null + message: string +} + +export interface OutboundProxyRecord { + id: string + name: string + endpoint: string + hasAuthentication: boolean + revision: number + accounts: { id: string, name: string }[] + lastTestAt: string | null + lastTest: OutboundProxyTest | null + createdAt: string + updatedAt: string +} + +interface ProxyPage { + items: OutboundProxyRecord[] + page: { page: number, pageSize: number, total: number, totalPages: number } +} + +interface ProxyMutation { + record: OutboundProxyRecord + configRevision: number +} + +export function getProxies(params: { page: number, pageSize: number, search?: string }) { + return request({ url: '/api/admin/proxies', method: 'GET', params }) +} + +export function createProxy(data: { name: string, proxyUrl: string }) { + return request({ url: '/api/admin/proxies/create', method: 'POST', data }) +} + +export function updateProxy(data: { id: string, revision: number, name: string, proxyUrl?: string }) { + return request({ url: '/api/admin/proxies/update', method: 'POST', data }) +} + +export function deleteProxy(data: { id: string, revision: number }) { + return request<{ configRevision: number }>({ url: '/api/admin/proxies/delete', method: 'POST', data }) +} + +export function testProxy(data: { id: string, revision: number }) { + return request({ url: '/api/admin/proxies/test', method: 'POST', data, timeout: 25000 }) +} diff --git a/frontend/src/composables/useProxyCatalog.ts b/frontend/src/composables/useProxyCatalog.ts new file mode 100644 index 000000000..4b362c4b2 --- /dev/null +++ b/frontend/src/composables/useProxyCatalog.ts @@ -0,0 +1,33 @@ +import type { OutboundProxyRecord } from '@/api' +import { onMounted, shallowRef } from 'vue' +import { getProxies } from '@/api' +import { toast } from '@/components/base/BaseToast' +import { errorMessage } from '@/utils/async' + +export function useProxyCatalog() { + const proxies = shallowRef([]) + const loading = shallowRef(false) + + async function loadProxies() { + if (loading.value) + return + loading.value = true + try { + const first = await getProxies({ page: 1, pageSize: 200 }) + const items = [...first.items] + for (let page = 2; page <= first.page.totalPages; page += 1) { + items.push(...(await getProxies({ page, pageSize: 200 })).items) + } + proxies.value = items + } + catch (error) { + toast.error(errorMessage(error, '代理列表加载失败')) + } + finally { + loading.value = false + } + } + + onMounted(() => void loadProxies()) + return { proxies, loading, loadProxies } +} diff --git a/frontend/src/layout/components/AppSidebar.vue b/frontend/src/layout/components/AppSidebar.vue index 4b3980095..17ace9d97 100644 --- a/frontend/src/layout/components/AppSidebar.vue +++ b/frontend/src/layout/components/AppSidebar.vue @@ -8,6 +8,7 @@ import { LayoutDashboard, LogOut, Moon, + Network, Palette, PanelLeftClose, PanelLeftOpen, @@ -58,6 +59,7 @@ const preferredMotion = usePreferredReducedMotion() const navItems = [ { label: '概览', icon: LayoutDashboard, path: '/' }, { label: '账号管理', icon: Users, path: '/accounts' }, + { label: '代理管理', icon: Network, path: '/proxies' }, { label: '分组管理', icon: FolderTree, path: '/account-groups' }, { label: 'API 密钥', icon: KeyRound, path: '/api-keys' }, { label: '使用统计', icon: ChartNoAxesColumn, path: '/usage' }, diff --git a/frontend/src/router/routes.ts b/frontend/src/router/routes.ts index 1272e31f6..41bd2fa3d 100644 --- a/frontend/src/router/routes.ts +++ b/frontend/src/router/routes.ts @@ -20,6 +20,11 @@ export const routes: RouteRecordRaw[] = [ name: 'accounts', component: () => import('@/views/accounts/index.vue'), }, + { + path: 'proxies', + name: 'proxies', + component: () => import('@/views/proxies/index.vue'), + }, { path: 'account-groups', name: 'account-groups', diff --git a/frontend/src/views/accounts/components/AccountBatchEditModal.vue b/frontend/src/views/accounts/components/AccountBatchEditModal.vue index 7f3921d4a..94fca00ae 100644 --- a/frontend/src/views/accounts/components/AccountBatchEditModal.vue +++ b/frontend/src/views/accounts/components/AccountBatchEditModal.vue @@ -21,7 +21,7 @@ const enabled = defineModel('enabled', { required: true }) const concurrencyLimit = defineModel('concurrencyLimit', { required: true }) const weight = defineModel('weight', { required: true }) const proxyMode = defineModel('proxyMode', { required: true }) -const proxyUrl = defineModel('proxyUrl', { required: true }) +const proxyId = defineModel('proxyId', { required: true }) const selectedGroupIds = defineModel('selectedGroupIds', { required: true }) @@ -39,7 +39,7 @@ const selectedGroupIds = defineModel('selectedGroupIds', { required: t v-model:weight="weight" v-model:selected-group-ids="selectedGroupIds" v-model:proxy-mode="proxyMode" - v-model:proxy-url="proxyUrl" + v-model:proxy-id="proxyId" :groups="groups" :groups-loading="groupsLoading" :disabled="saving" diff --git a/frontend/src/views/accounts/components/AccountCreateModal/AccountSetupFields.vue b/frontend/src/views/accounts/components/AccountCreateModal/AccountSetupFields.vue index 16d653721..f788c7200 100644 --- a/frontend/src/views/accounts/components/AccountCreateModal/AccountSetupFields.vue +++ b/frontend/src/views/accounts/components/AccountCreateModal/AccountSetupFields.vue @@ -27,7 +27,7 @@ const form = defineModel({ required: true }) v-model:weight="form.weight" v-model:selected-group-ids="form.groupIds" v-model:proxy-mode="form.proxyMode" - v-model:proxy-url="form.proxyUrl" + v-model:proxy-id="form.proxyId" :groups="groups" :groups-loading="groupsLoading" :preserve-proxy="false" diff --git a/frontend/src/views/accounts/components/AccountCreateModal/index.vue b/frontend/src/views/accounts/components/AccountCreateModal/index.vue index 37c68aadb..cde101bcc 100644 --- a/frontend/src/views/accounts/components/AccountCreateModal/index.vue +++ b/frontend/src/views/accounts/components/AccountCreateModal/index.vue @@ -81,7 +81,7 @@ function continueToImport() { :disabled="busy" :groups="groups" :groups-loading="groupsLoading" - :proxy-error="form.proxyUrl.trim() ? proxyError : undefined" + :proxy-error="form.proxyId.trim() ? proxyError : undefined" />