feat: add managed account proxies and fix non-streaming WebSocket requests

This commit is contained in:
Yanshan Chen
2026-09-11 11:55:09 +08:00
parent 2b98839c29
commit 704b652f86
73 changed files with 2955 additions and 155 deletions
+1 -1
View File
@@ -21,7 +21,7 @@
## 功能
- **账号管理**:OAuth 授权、账号导入、凭据刷新与额度查看;支持独立出站代理及 sub2api OpenAI 账号代理导入。
- **账号管理**:OAuth 授权、账号导入、凭据刷新与额度查看;代理管理支持独立保存、连接测试和账号绑定,以及 sub2api OpenAI 账号代理导入。
- **访问控制**:通过账号分组分配密钥权限,设置日/周 USD 限额、并发和请求频率。
- **客户端接入**:支持 Codex CLI、桌面端与 Responses API 客户端,提供 HTTP 流式响应和 WebSocket。
- **请求诊断**:查看请求记录、用量、费用、延迟和错误详情。
+1 -1
View File
@@ -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(),
)
@@ -182,6 +182,7 @@ const ADAPTER_PUBLIC_MODULES: &[(&str, &[&str])] = &[
&[
"client_distribution",
"config",
"proxy_probe",
"serve",
"system_update",
"workers",
+16 -2
View File
@@ -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<dyn ProxiesService>,
auth: Arc<dyn AuthService>,
accounts: Arc<dyn AccountsService>,
account_groups: Arc<dyn AccountGroupService>,
@@ -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<Arc<dyn ProviderAdmin>>,
snapshot: Arc<dyn SnapshotControl>,
probe: Arc<dyn AccountProbe>,
probes: (Arc<dyn AccountProbe>, Arc<dyn ports::proxy::ProxyProbe>),
client_distribution: Arc<dyn ClientDistributionResolver>,
system: Arc<dyn SystemOperations>,
) -> Result<AdminBundle, AdminError> {
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,
@@ -221,7 +221,7 @@ pub struct UpdateAccount {
pub concurrency_limit: Option<AccountConcurrencyLimit>,
pub weight: AccountWeight,
pub group_ids: Vec<gateway_core::routing::AccountGroupId>,
pub outbound_proxy: Option<Option<gateway_core::account::OutboundProxy>>,
pub outbound_proxy: Option<super::proxies::AccountProxySelection>,
}
/// 账号更新结果。
@@ -239,7 +239,7 @@ pub struct BatchUpdateAccounts {
pub concurrency_limit: Option<AccountConcurrencyLimit>,
pub weight: AccountWeight,
pub group_ids: Vec<gateway_core::routing::AccountGroupId>,
pub outbound_proxy: Option<Option<gateway_core::account::OutboundProxy>>,
pub outbound_proxy: Option<super::proxies::AccountProxySelection>,
}
/// 批量账号更新结果。
@@ -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;
@@ -108,6 +108,7 @@ pub struct CredentialDetails {
/// Provider 正式文档批量导入命令。
pub struct ImportCredentials {
pub outbound_proxy_id: Option<String>,
pub settings: Option<AccountImportSettings>,
pub context: MutationContext,
pub document: ProviderDocument,
@@ -132,6 +133,7 @@ pub struct CredentialImportResult {
/// Provider 解析导入文档时只接收不透明文档,不接触 revision 或审计上下文。
pub struct PrepareCredentialImport {
pub default_outbound_proxy: Option<gateway_core::account::OutboundProxy>,
pub document: ProviderDocument,
}
@@ -175,6 +177,7 @@ pub struct PreparedCredentialImport {
/// Admin 交给 Store 的导入事务命令。
#[derive(Debug, Clone, PartialEq)]
pub struct CredentialImportCommit {
pub outbound_proxy: Option<super::proxies::ImportProxyBinding>,
pub settings: Option<AccountImportSettings>,
pub prepared: PreparedCredentialImport,
}
@@ -255,6 +258,7 @@ pub struct PendingAuthorizationMutation {
target: AuthorizationMutationTarget,
owner_binding: AuthorizationOwnerBinding,
outbound_proxy: Option<gateway_core::account::OutboundProxy>,
outbound_proxy_id: Option<String>,
}
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<String>) -> 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<String>,
#[serde(default)]
outbound_proxy_id: Option<String>,
}
#[derive(Deserialize)]
@@ -418,7 +439,7 @@ pub struct StartAuthorization {
pub context: MutationContext,
pub name: String,
pub reauthorization: Option<ProviderAccountId>,
pub outbound_proxy: Option<gateway_core::account::OutboundProxy>,
pub outbound_proxy: Option<super::proxies::AccountProxySelection>,
}
/// OAuth 流程启动结果。
@@ -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<std::net::IpAddr>,
pub message: String,
}
#[derive(Debug, Clone)]
pub struct ProxyRecord {
pub id: String,
pub name: String,
pub proxy: OutboundProxy,
pub revision: Revision,
pub accounts: Vec<ProxyAccountRef>,
pub last_test_at: Option<DateTime<Utc>>,
pub last_test: Option<ProxyTestResult>,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
#[derive(Debug, Clone)]
pub struct ProxyPage {
pub items: Vec<ProxyRecord>,
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<OutboundProxy>,
}
#[derive(Debug, Clone)]
pub struct ProxyMutation {
pub config_revision: Revision,
pub record: ProxyRecord,
}
@@ -3,5 +3,6 @@
pub mod backup;
pub mod client_distribution;
pub mod provider;
pub mod proxy;
pub mod store;
pub mod system;
@@ -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<ProxyPage>;
async fn get(&self, id: &str) -> AdminStoreResult<ProxyRecord>;
async fn create(
&self,
command: NewProxy,
context: &MutationContext,
) -> AdminStoreResult<ProxyMutation>;
async fn update(
&self,
command: UpdateProxy,
context: &MutationContext,
) -> AdminStoreResult<ProxyMutation>;
async fn delete(
&self,
id: &str,
revision: Revision,
context: &MutationContext,
) -> AdminStoreResult<Revision>;
async fn record_test(
&self,
id: &str,
revision: Revision,
result: ProxyTestResult,
context: &MutationContext,
) -> AdminStoreResult<ProxyRecord>;
}
#[async_trait]
pub trait ProxyProbe: Send + Sync {
async fn test(&self, proxy: &OutboundProxy) -> ProxyTestResult;
}
@@ -382,6 +382,7 @@ pub struct AdminAccountStorePorts {
accounts: Arc<dyn AccountStore>,
runtime: Arc<dyn AccountRuntimeStore>,
groups: Arc<dyn AccountGroupStore>,
proxies: Arc<dyn super::proxy::ProxyStore>,
}
impl AdminAccountStorePorts {
@@ -390,11 +391,13 @@ impl AdminAccountStorePorts {
accounts: Arc<dyn AccountStore>,
runtime: Arc<dyn AccountRuntimeStore>,
groups: Arc<dyn AccountGroupStore>,
proxies: Arc<dyn super::proxy::ProxyStore>,
) -> Self {
Self {
accounts,
runtime,
groups,
proxies,
}
}
}
@@ -447,6 +450,11 @@ impl AdminStorePorts {
self.accounts.groups.clone()
}
#[must_use]
pub fn proxies(&self) -> Arc<dyn super::proxy::ProxyStore> {
self.accounts.proxies.clone()
}
#[must_use]
pub fn auth(&self) -> Arc<dyn AuthStore> {
self.auth.clone()
@@ -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<Option<crate::model::proxies::ImportProxyBinding>, 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<PendingAuthorizationMutation, AdminError> {
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(
@@ -58,6 +58,7 @@ pub trait OpenAiService: Send + Sync {
pub(crate) struct DefaultOpenAiService {
provider: Arc<dyn ProviderAdmin>,
accounts: Arc<dyn AccountStore>,
proxies: Arc<dyn crate::ports::proxy::ProxyStore>,
snapshot: Arc<dyn SnapshotControl>,
}
@@ -66,11 +67,13 @@ impl DefaultOpenAiService {
pub(crate) fn new(
provider: Arc<dyn ProviderAdmin>,
accounts: Arc<dyn AccountStore>,
proxies: Arc<dyn crate::ports::proxy::ProxyStore>,
snapshot: Arc<dyn SnapshotControl>,
) -> Self {
Self {
provider,
accounts,
proxies,
snapshot,
}
}
@@ -128,9 +131,17 @@ impl OpenAiService for DefaultOpenAiService {
command: ImportCredentials,
) -> Result<CredentialImportResult, AdminError> {
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<AuthorizationStarted, AdminError> {
let pending = pending_authorization(
self.accounts.as_ref(),
self.proxies.as_ref(),
self.provider.provider_kind(),
&command,
"OpenAI credential",
@@ -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<ProxyPage, AdminError>;
async fn create(
&self,
command: NewProxy,
context: &MutationContext,
) -> Result<ProxyMutation, AdminError>;
async fn update(
&self,
command: UpdateProxy,
context: &MutationContext,
) -> Result<ProxyMutation, AdminError>;
async fn delete(
&self,
id: &str,
revision: Revision,
context: &MutationContext,
) -> Result<Revision, AdminError>;
async fn test(
&self,
id: &str,
revision: Revision,
context: &MutationContext,
) -> Result<ProxyRecord, AdminError>;
}
pub(crate) struct DefaultProxiesService {
store: Arc<dyn ProxyStore>,
probe: Arc<dyn ProxyProbe>,
snapshot: Arc<dyn SnapshotControl>,
test_slots: Semaphore,
}
impl DefaultProxiesService {
pub(crate) fn new(
store: Arc<dyn ProxyStore>,
probe: Arc<dyn ProxyProbe>,
snapshot: Arc<dyn SnapshotControl>,
) -> Self {
Self {
store,
probe,
snapshot,
test_slots: Semaphore::new(4),
}
}
}
#[async_trait]
impl ProxiesService for DefaultProxiesService {
async fn list(&self, query: ProxyListQuery) -> Result<ProxyPage, AdminError> {
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<ProxyMutation, AdminError> {
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<ProxyMutation, AdminError> {
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<Revision, AdminError> {
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<ProxyRecord, AdminError> {
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<String, AdminError> {
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())
}
@@ -49,6 +49,7 @@ pub trait XaiService: Send + Sync {
pub(crate) struct DefaultXaiService {
provider: Arc<dyn ProviderAdmin>,
accounts: Arc<dyn AccountStore>,
proxies: Arc<dyn crate::ports::proxy::ProxyStore>,
snapshot: Arc<dyn SnapshotControl>,
}
@@ -57,11 +58,13 @@ impl DefaultXaiService {
pub(crate) fn new(
provider: Arc<dyn ProviderAdmin>,
accounts: Arc<dyn AccountStore>,
proxies: Arc<dyn crate::ports::proxy::ProxyStore>,
snapshot: Arc<dyn SnapshotControl>,
) -> Self {
Self {
provider,
accounts,
proxies,
snapshot,
}
}
@@ -81,9 +84,17 @@ impl XaiService for DefaultXaiService {
command: ImportCredentials,
) -> Result<CredentialImportResult, AdminError> {
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<AuthorizationStarted, AdminError> {
let pending = pending_authorization(
self.accounts.as_ref(),
self.proxies.as_ref(),
self.provider.provider_kind(),
&command,
"xAI credential",
@@ -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)
);
}
}
@@ -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,
)
@@ -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(),
@@ -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<ProxyPage> {
Err(super::unavailable("proxy"))
}
async fn get(&self, _: &str) -> AdminStoreResult<ProxyRecord> {
Err(super::unavailable("proxy"))
}
async fn create(&self, _: NewProxy, _: &MutationContext) -> AdminStoreResult<ProxyMutation> {
Err(super::unavailable("proxy"))
}
async fn update(&self, _: UpdateProxy, _: &MutationContext) -> AdminStoreResult<ProxyMutation> {
Err(super::unavailable("proxy"))
}
async fn delete(
&self,
_: &str,
_: Revision,
_: &MutationContext,
) -> AdminStoreResult<Revision> {
Err(super::unavailable("proxy"))
}
async fn record_test(
&self,
_: &str,
_: Revision,
_: ProxyTestResult,
_: &MutationContext,
) -> AdminStoreResult<ProxyRecord> {
Err(super::unavailable("proxy"))
}
}
#[async_trait]
impl ProxyProbe for UnavailableProxies {
async fn test(&self, _: &OutboundProxy) -> ProxyTestResult {
panic!("unexpected proxy probe")
}
}
@@ -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(),
@@ -53,6 +53,7 @@ impl AccountImportSettingsRequest {
#[derive(Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct AccountImportRequest {
pub outbound_proxy_id: Option<String>,
pub settings: Option<AccountImportSettingsRequest>,
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<String>,
pub provider: String,
pub name: String,
pub account_id: Option<String>,
@@ -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<String>,
pub outbound_proxy_url: Option<super::wire::AccountProxyUpdate>,
pub account_id: String,
pub enabled: bool,
@@ -209,7 +219,10 @@ impl UpdateAccountRequest {
pub(super) fn into_command(self) -> Result<UpdateAccount, WireValidationError> {
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)?,
@@ -17,6 +17,33 @@ impl<'de> Deserialize<'de> for AccountProxyUpdate {
}
}
pub(super) fn proxy_selection(
id: Option<String>,
url: Option<AccountProxyUpdate>,
) -> Result<Option<gateway_admin::model::proxies::AccountProxySelection>, 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<String>,
pub outbound_proxy_url: Option<AccountProxyUpdate>,
pub account_ids: Vec<String>,
pub enabled: bool,
@@ -84,7 +112,7 @@ impl BatchUpdateAccountsRequest {
pub(super) fn into_command(self) -> Result<BatchUpdateAccounts, WireValidationError> {
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)?,
@@ -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::<S>())
.merge(proxies::router::<S>())
.merge(accounts::router::<S>())
.merge(auth::router::<S>())
.merge(backups::router::<S>())
@@ -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<u32>,
page_size: Option<u16>,
search: Option<String>,
}
#[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<AccountProxyUpdate>,
}
#[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<String>,
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<ProxyAccountView>,
last_test_at: Option<String>,
last_test: Option<ProxyTestView>,
created_at: String,
updated_at: String,
}
impl From<ProxyRecord> 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<ProxyView>,
page: PageMeta,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct MutationView {
record: ProxyView,
config_revision: u64,
}
impl From<ProxyMutation> for MutationView {
fn from(value: ProxyMutation) -> Self {
Self {
record: value.record.into(),
config_revision: value.config_revision.get(),
}
}
}
pub fn router<S>() -> Router<S>
where
S: AdminSessionState + Clone + Send + Sync + 'static,
{
Router::new()
.route("/api/admin/proxies", get(list::<S>))
.route("/api/admin/proxies/create", post(create::<S>))
.route("/api/admin/proxies/update", post(update::<S>))
.route("/api/admin/proxies/delete", post(delete::<S>))
.route("/api/admin/proxies/test", post(test::<S>))
}
fn revision(value: u64) -> Result<Revision, AdminError> {
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<S>(
_: AdminAuth,
State(state): State<S>,
AdminQuery(query): AdminQuery<ListQuery>,
) -> Result<impl IntoResponse, AdminError>
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<S>(
auth: AdminAuth,
State(state): State<S>,
AdminJson(request): AdminJson<CreateRequest>,
) -> Result<impl IntoResponse, AdminError>
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<S>(
auth: AdminAuth,
State(state): State<S>,
AdminJson(request): AdminJson<UpdateRequest>,
) -> Result<impl IntoResponse, AdminError>
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<S>(
auth: AdminAuth,
State(state): State<S>,
AdminJson(request): AdminJson<IdRequest>,
) -> Result<impl IntoResponse, AdminError>
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<S>(
auth: AdminAuth,
State(state): State<S>,
AdminJson(request): AdminJson<IdRequest>,
) -> Result<impl IntoResponse, AdminError>
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)),
))
}
@@ -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,
)
@@ -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<Option<ProxyRecord>>);
fn missing() -> AdminStoreError {
AdminStoreError::new(AdminStoreErrorKind::NotFound, "proxy", "missing proxy")
}
#[async_trait]
impl ProxyStore for MemoryProxies {
async fn list(&self, query: ProxyListQuery) -> AdminStoreResult<ProxyPage> {
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<ProxyRecord> {
self.0.lock().unwrap().clone().ok_or_else(missing)
}
async fn create(
&self,
command: NewProxy,
_: &MutationContext,
) -> AdminStoreResult<ProxyMutation> {
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<ProxyMutation> {
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<Revision> {
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<ProxyRecord> {
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<Value>,
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::<AdminTestState>()
.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);
}
@@ -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",
+1 -1
View File
@@ -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"] }
+6
View File
@@ -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<dyn gateway_admin::ports::proxy::ProxyProbe> {
Arc::new(proxy_probe::HttpProxyProbe::default())
}
#[must_use]
pub fn connection_lifecycle(&self) -> Arc<dyn ConnectionLifecycle> {
self.connections.clone()
@@ -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<String>) -> Self {
Self {
endpoint: endpoint.into(),
}
}
async fn exit_ip(&self, proxy: &OutboundProxy) -> Result<IpAddr, &'static str> {
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::<Response>(&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()),
}
}
}
@@ -1,6 +1,7 @@
mod client_distribution;
mod config;
mod logging;
mod proxy_probe;
mod serve;
mod system_update;
mod workers;
@@ -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);
}
@@ -106,6 +106,7 @@ pub async fn initialize(mut config: StoreConfig) -> StoreResult<StoreBundle> {
)),
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()),
@@ -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::*;
@@ -216,6 +216,7 @@ impl PgAdminAccountStore {
settings: Option<AccountImportSettings>,
context: &MutationContext,
action: &str,
outbound_proxy: Option<gateway_admin::model::proxies::ImportProxyBinding>,
) -> AdminStoreResult<CredentialImportResult> {
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] =
@@ -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<AccountConcurrencyLimit>,
weight: AccountWeight,
outbound_proxy: Option<&Option<gateway_core::account::OutboundProxy>>,
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"))?
@@ -233,6 +233,7 @@ impl NewProviderAccount {
#[derive(Clone)]
pub struct ImportProviderAccounts {
pub settings: Option<AccountImportSettings>,
pub outbound_proxy: Option<gateway_admin::model::proxies::ImportProxyBinding>,
pub scope: ProviderAccountAdminScope,
pub accounts: Vec<NewProviderAccount>,
pub audit: AdminAuditEvent,
@@ -313,7 +314,7 @@ impl fmt::Debug for RotateProviderAccount {
#[derive(Debug, Clone)]
pub struct BatchUpdateProviderAccountsAdmin {
pub outbound_proxy: Option<Option<gateway_core::account::OutboundProxy>>,
pub outbound_proxy: Option<gateway_admin::model::proxies::AccountProxySelection>,
pub account_ids: Vec<String>,
pub enabled: bool,
pub concurrency_limit: Option<AccountConcurrencyLimit>,
@@ -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<ProxyRecord> {
let success: Option<bool> = row.try_get("last_test_success").map_err(|_| invalid())?;
let ip: Option<String> = row.try_get("last_test_ip").map_err(|_| invalid())?;
let latency: Option<i64> = row.try_get("last_test_latency_ms").map_err(|_| invalid())?;
let ids: Vec<String> = row.try_get("account_ids").map_err(|_| invalid())?;
let names: Vec<String> = 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::<String, _>("proxy_url")
.map_err(|_| invalid())?,
)
.map_err(|_| invalid())?,
revision: AdminRevision::new(
u64::try_from(row.try_get::<i64, _>("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::<Option<String>, _>("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<String>, Option<OutboundProxy>)> {
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<bool>)>(
"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<ProxyPage> {
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::<Postgres>::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::<StoreResult<Vec<_>>>()
.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<ProxyRecord> {
let mut builder = QueryBuilder::<Postgres>::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<ProxyMutation> {
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<ProxyMutation> {
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<AdminRevision> {
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<ProxyRecord> {
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
}
}
@@ -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",
]
@@ -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")
@@ -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<String>, Option<String>, 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<serde_json::Value> =
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<String>>(
"select outbound_proxy_id from provider_accounts where id = 'acct_direct'"
)
.fetch_one(&database.pool)
.await
.unwrap()
.is_none()
);
database.close().await;
}
+4 -3
View File
@@ -291,9 +291,10 @@ impl ProviderAdmin for OpenAiAdminProvider {
) -> Result<PreparedCredentialImport, ProviderAdminError> {
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));
@@ -824,6 +824,14 @@ impl CodexCredentialAdminService {
pub async fn prepare_import_document(
&self,
payload: Value,
) -> Result<PreparedCodexAccountImport, CodexCredentialAdminError> {
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<PreparedCodexAccountImport, CodexCredentialAdminError> {
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<Vec<ParsedCodexImportAccount>, 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)
@@ -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
}
@@ -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"),
@@ -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");
@@ -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(
+5 -2
View File
@@ -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,后续条目的任何失败都不得
@@ -162,6 +162,13 @@ impl GrokOAuthImportDocument {
/// 独立代理 URL 随账号导入;其他展示 metadata 不参与认证。
/// 实际 token 仍须通过官方 refresh/user-info 验证;API Key 不能混入 OAuth 条目。
pub fn parse_json(document: &[u8]) -> Result<Self, GrokOAuthImportError> {
Self::parse_json_with_proxy(document, None)
}
pub fn parse_json_with_proxy(
document: &[u8],
default_proxy: Option<&gateway_core::account::OutboundProxy>,
) -> Result<Self, GrokOAuthImportError> {
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);
}
}
@@ -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
@@ -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 =
+1
View File
@@ -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
@@ -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;
+44 -5
View File
@@ -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. 运行设置
| 方法 | 路由 | 说明 |
+10 -1
View File
@@ -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 密钥,并声明服务端托管账号认证。
+2 -1
View File
@@ -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'
+4
View File
@@ -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
+51
View File
@@ -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<ProxyPage>({ url: '/api/admin/proxies', method: 'GET', params })
}
export function createProxy(data: { name: string, proxyUrl: string }) {
return request<ProxyMutation>({ url: '/api/admin/proxies/create', method: 'POST', data })
}
export function updateProxy(data: { id: string, revision: number, name: string, proxyUrl?: string }) {
return request<ProxyMutation>({ 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<OutboundProxyRecord>({ url: '/api/admin/proxies/test', method: 'POST', data, timeout: 25000 })
}
@@ -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<OutboundProxyRecord[]>([])
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 }
}
@@ -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' },
+5
View File
@@ -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',
@@ -21,7 +21,7 @@ const enabled = defineModel<boolean>('enabled', { required: true })
const concurrencyLimit = defineModel<string>('concurrencyLimit', { required: true })
const weight = defineModel<string>('weight', { required: true })
const proxyMode = defineModel<string>('proxyMode', { required: true })
const proxyUrl = defineModel<string>('proxyUrl', { required: true })
const proxyId = defineModel<string>('proxyId', { required: true })
const selectedGroupIds = defineModel<string[]>('selectedGroupIds', { required: true })
</script>
@@ -39,7 +39,7 @@ const selectedGroupIds = defineModel<string[]>('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"
@@ -27,7 +27,7 @@ const form = defineModel<AccountCreateForm>({ 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"
@@ -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"
/>
<p v-if="view.configuring && !scheduling.valid" class="m-0 text-xs text-cp-error" role="alert">
{{ scheduling.message }}
@@ -17,7 +17,7 @@ export interface AccountCreateForm {
oauthAuthUrl: string
oauthCallback: string
proxyMode: string
proxyUrl: string
proxyId: string
}
export function emptyAccountCreateForm(): AccountCreateForm {
@@ -34,25 +34,15 @@ export function emptyAccountCreateForm(): AccountCreateForm {
oauthAuthUrl: '',
oauthCallback: '',
proxyMode: 'direct',
proxyUrl: '',
proxyId: '',
}
}
export function accountProxyError(form: AccountCreateForm): string | undefined {
if (form.proxyMode !== 'proxy')
return undefined
if (!form.proxyUrl.trim())
return '请输入代理 URL'
try {
const url = new URL(form.proxyUrl.trim())
if (!['http:', 'https:', 'socks5:', 'socks5h:'].includes(url.protocol) || !url.hostname)
return '请输入 HTTP、HTTPS 或 SOCKS5 代理地址'
if (url.protocol.startsWith('socks5') && !url.port)
return 'SOCKS5 代理地址需要填写端口'
}
catch {
return '请输入有效的代理 URL'
}
if (!form.proxyId.trim())
return '请选择已通过测试的代理'
return undefined
}
@@ -25,7 +25,7 @@ const enabled = defineModel<boolean>('enabled', { required: true })
const concurrencyLimit = defineModel<string>('concurrencyLimit', { required: true })
const weight = defineModel<string>('weight', { required: true })
const proxyMode = defineModel<string>('proxyMode', { required: true })
const proxyUrl = defineModel<string>('proxyUrl', { required: true })
const proxyId = defineModel<string>('proxyId', { required: true })
const selectedGroupIds = defineModel<string[]>('selectedGroupIds', { required: true })
</script>
@@ -61,11 +61,12 @@ const selectedGroupIds = defineModel<string[]>('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"
:endpoint="account.outboundProxyEndpoint"
:account-id="account.id"
/>
</div>
@@ -1,50 +1,51 @@
<script setup lang="ts">
import { ExternalLink, RefreshCw } from '@lucide/vue'
import { computed } from 'vue'
import BaseFormItem from '@/components/base/BaseForm/FormItem.vue'
import BaseInput from '@/components/base/BaseInput.vue'
import BaseIconButton from '@/components/base/BaseIconButton.vue'
import BaseSelect from '@/components/base/BaseSelect.vue'
import { useProxyCatalog } from '@/composables/useProxyCatalog'
const props = withDefaults(
defineProps<{
endpoint?: string | null
disabled?: boolean
preserve?: boolean
error?: string
}>(),
{ preserve: true },
)
const props = withDefaults(defineProps<{ accountId?: string, endpoint?: string | null, disabled?: boolean, preserve?: boolean, error?: string }>(), { preserve: true })
const mode = defineModel<string>('mode', { required: true })
const url = defineModel<string>('url', { required: true })
const proxyId = defineModel<string>('proxyId', { required: true })
const { proxies, loading, loadProxies } = useProxyCatalog()
const current = computed(() => proxies.value.find(proxy => proxy.accounts.some(account => account.id === props.accountId)))
const options = computed(() => [
...(props.preserve ? [{ label: '保持当前', value: 'preserve' }] : []),
...(props.preserve ? [{ label: current.value ? `保持当前:${current.value.name}` : '保持当前', value: 'preserve' }] : []),
{ label: '直连', value: 'direct' },
{ label: '指定代理', value: 'proxy' },
...proxies.value.map(proxy => ({
label: `${proxy.name}${proxy.lastTest?.success ? '' : proxy.lastTest ? '(测试失败)' : '(未测试)'}`,
value: proxy.id,
disabled: proxy.lastTest?.success !== true,
})),
])
const selection = computed({
get: () => mode.value === 'proxy' ? proxyId.value : mode.value,
set: (value: string) => {
const nextMode = value === 'preserve' || value === 'direct' ? value : 'proxy'
mode.value = nextMode
proxyId.value = nextMode === 'proxy' ? value : ''
},
})
</script>
<template>
<div class="grid gap-5">
<BaseFormItem label="出站隧道">
<BaseSelect
v-model="mode"
class="w-full"
:options="options"
:disabled="disabled"
aria-label="出站隧道"
/>
<p v-if="endpoint && mode === 'preserve'" class="mt-2 mb-0 font-mono text-xs break-all text-cp-text-quaternary">
{{ endpoint }}
</p>
</BaseFormItem>
<BaseFormItem v-if="mode === 'proxy'" label="代理 URL" required :error="error">
<BaseInput
v-model="url"
type="password"
autocomplete="new-password"
aria-label="代理 URL"
placeholder="例如:http://127.0.0.1:7890"
:disabled="disabled"
/>
<div class="grid min-w-0 grid-cols-1 gap-3">
<BaseFormItem label="出站隧道" :error="error">
<div class="flex min-w-0 items-center gap-2">
<BaseSelect v-model="selection" class="min-w-0 flex-1" :options="options" :disabled="disabled || loading" aria-label="出站隧道" />
<BaseIconButton label="刷新可选代理" :disabled="disabled || loading" @click="loadProxies()">
<RefreshCw class="size-4" />
</BaseIconButton>
<a href="/proxies" target="_blank" rel="noopener noreferrer" class="inline-flex size-8 shrink-0 items-center justify-center text-cp-link" aria-label="打开代理管理" title="打开代理管理"><ExternalLink class="size-4" /></a>
</div>
</BaseFormItem>
<p v-if="endpoint && mode === 'preserve'" class="m-0 font-mono text-xs break-all text-cp-text-tertiary">
{{ endpoint }}
</p>
<p v-if="mode === 'proxy'" class="m-0 break-all font-mono text-xs text-cp-text-secondary">
{{ proxies.find(proxy => proxy.id === proxyId)?.endpoint }}
</p>
</div>
</template>
@@ -11,6 +11,7 @@ withDefaults(defineProps<{
groupsLoading: boolean
disabled: boolean
endpoint?: string | null
accountId?: string
preserveProxy?: boolean
proxyError?: string
}>(), { preserveProxy: true })
@@ -19,7 +20,7 @@ const enabled = defineModel<boolean>('enabled', { required: true })
const concurrencyLimit = defineModel<string>('concurrencyLimit', { required: true })
const weight = defineModel<string>('weight', { required: true })
const proxyMode = defineModel<string>('proxyMode', { required: true })
const proxyUrl = defineModel<string>('proxyUrl', { required: true })
const proxyId = defineModel<string>('proxyId', { required: true })
const selectedGroupIds = defineModel<string[]>('selectedGroupIds', { required: true })
</script>
@@ -67,6 +68,6 @@ const selectedGroupIds = defineModel<string[]>('selectedGroupIds', { required: t
:disabled="disabled"
/>
</BaseFormItem>
<AccountProxyField v-model:mode="proxyMode" v-model:url="proxyUrl" :preserve="preserveProxy" :error="proxyError" :endpoint="endpoint" :disabled="disabled" />
<AccountProxyField v-model:mode="proxyMode" v-model:proxy-id="proxyId" :preserve="preserveProxy" :error="proxyError" :endpoint="endpoint" :account-id="accountId" :disabled="disabled" />
</div>
</template>
@@ -21,7 +21,7 @@ export function useAccountBatchEditor(options: {
const concurrencyLimit = shallowRef('')
const weight = shallowRef('1')
const proxyMode = shallowRef('preserve')
const proxyUrl = shallowRef('')
const proxyId = shallowRef('')
const selectedGroupIds = ref<string[]>([])
const saveAction = useAsyncAction()
const saving = saveAction.loading
@@ -33,7 +33,7 @@ export function useAccountBatchEditor(options: {
schedulingEnabled.value = accounts.every(account => account.enabled)
proxyMode.value = 'preserve'
proxyUrl.value = ''
proxyId.value = ''
concurrencyLimit.value = sharedConcurrencyLimit(accounts)
weight.value = sharedWeight(accounts)
selectedGroupIds.value = sharedGroupIds(accounts)
@@ -44,8 +44,8 @@ export function useAccountBatchEditor(options: {
if (saving.value || options.selectedIds.value.size === 0)
return
const scheduling = parseAccountSchedulingForm(concurrencyLimit.value, weight.value)
if (proxyMode.value === 'proxy' && !proxyUrl.value.trim()) {
toast.warning('请输入代理 URL')
if (proxyMode.value === 'proxy' && !proxyId.value.trim()) {
toast.warning('请选择已通过测试的代理')
return
}
if (!scheduling.valid) {
@@ -57,7 +57,7 @@ export function useAccountBatchEditor(options: {
const accountIds = selectedAccounts().map(account => account.id)
await batchUpdateAccounts({
accountIds,
outboundProxyUrl: proxyMode.value === 'preserve' ? undefined : proxyMode.value === 'direct' ? '' : proxyUrl.value.trim(),
outboundProxyId: proxyMode.value === 'preserve' ? undefined : proxyMode.value === 'direct' ? '' : proxyId.value.trim(),
enabled: schedulingEnabled.value,
concurrencyLimit: scheduling.values.concurrencyLimit,
weight: scheduling.values.weight,
@@ -99,7 +99,7 @@ export function useAccountBatchEditor(options: {
return
schedulingEnabled.value = true
proxyMode.value = 'preserve'
proxyUrl.value = ''
proxyId.value = ''
concurrencyLimit.value = ''
weight.value = '1'
selectedGroupIds.value = []
@@ -111,7 +111,7 @@ export function useAccountBatchEditor(options: {
concurrencyLimit,
weight,
proxyMode,
proxyUrl,
proxyId,
selectedGroupIds,
saving,
open,
@@ -20,7 +20,7 @@ export function useAccountEditor(options: {
const concurrencyLimit = shallowRef('')
const weight = shallowRef('1')
const proxyMode = shallowRef('preserve')
const proxyUrl = shallowRef('')
const proxyId = shallowRef('')
const selectedGroupIds = ref<string[]>([])
const saveAction = useAsyncAction()
const saving = saveAction.loading
@@ -34,7 +34,7 @@ export function useAccountEditor(options: {
function open(account: AccountRow) {
editingAccountId.value = account.id
proxyMode.value = 'preserve'
proxyUrl.value = ''
proxyId.value = ''
schedulingEnabled.value = account.enabled
concurrencyLimit.value = concurrencyLimitInput(account.concurrencyLimit)
weight.value = String(account.weight)
@@ -47,8 +47,8 @@ export function useAccountEditor(options: {
if (!accountId || saving.value)
return
const scheduling = parseAccountSchedulingForm(concurrencyLimit.value, weight.value)
if (proxyMode.value === 'proxy' && !proxyUrl.value.trim()) {
toast.warning('请输入代理 URL')
if (proxyMode.value === 'proxy' && !proxyId.value.trim()) {
toast.warning('请选择已通过测试的代理')
return
}
if (!scheduling.valid) {
@@ -59,7 +59,7 @@ export function useAccountEditor(options: {
await saveAction.run(async () => {
await updateAccount({
accountId,
outboundProxyUrl: proxyMode.value === 'preserve' ? undefined : proxyMode.value === 'direct' ? '' : proxyUrl.value.trim(),
outboundProxyId: proxyMode.value === 'preserve' ? undefined : proxyMode.value === 'direct' ? '' : proxyId.value.trim(),
enabled: schedulingEnabled.value,
concurrencyLimit: scheduling.values.concurrencyLimit,
weight: scheduling.values.weight,
@@ -76,7 +76,7 @@ export function useAccountEditor(options: {
return
editingAccountId.value = null
proxyMode.value = 'preserve'
proxyUrl.value = ''
proxyId.value = ''
schedulingEnabled.value = true
concurrencyLimit.value = ''
weight.value = '1'
@@ -90,7 +90,7 @@ export function useAccountEditor(options: {
concurrencyLimit,
weight,
proxyMode,
proxyUrl,
proxyId,
selectedGroupIds,
saving,
open,
@@ -82,7 +82,7 @@ export function useAccountOnboarding(options: {
throw new Error(proxyError)
const result = await startAccountOAuth({
...input,
outboundProxyUrl: !account && createForm.value.proxyMode === 'proxy' ? createForm.value.proxyUrl.trim() : undefined,
outboundProxyId: !account && createForm.value.proxyMode === 'proxy' ? createForm.value.proxyId.trim() : undefined,
...(account
? {
accountId: account.id,
@@ -180,9 +180,8 @@ export function useAccountOnboarding(options: {
const result = await importAccounts({
provider,
settings: accountImportSettings(createForm.value),
data: createForm.value.proxyMode === 'proxy'
? withDefaultImportProxy(entry.document, createForm.value.proxyUrl.trim())
: entry.document,
outboundProxyId: createForm.value.proxyMode === 'proxy' ? createForm.value.proxyId.trim() : undefined,
data: entry.document,
})
importedCount += result.importedCount
}
@@ -199,9 +198,8 @@ export function useAccountOnboarding(options: {
const result = await importAccounts({
provider: entry.provider,
settings: accountImportSettings(createForm.value),
data: createForm.value.proxyMode === 'proxy'
? withDefaultImportProxy(entry.document, createForm.value.proxyUrl.trim())
: entry.document,
outboundProxyId: createForm.value.proxyMode === 'proxy' ? createForm.value.proxyId.trim() : undefined,
data: entry.document,
})
importedCount += result.importedCount
}
@@ -245,7 +243,7 @@ export function useAccountOnboarding(options: {
watch(
[
() => createForm.value.proxyMode,
() => createForm.value.proxyMode === 'proxy' ? createForm.value.proxyUrl.trim() : '',
() => createForm.value.proxyMode === 'proxy' ? createForm.value.proxyId.trim() : '',
],
() => {
createForm.value.oauthFlowId = ''
@@ -268,20 +266,6 @@ export function useAccountOnboarding(options: {
}
}
function withDefaultImportProxy(document: Record<string, unknown>, proxyUrl: string): Record<string, unknown> {
if (isRecord(document.data) && Array.isArray(document.data.accounts))
return { ...document, data: withDefaultImportProxy(document.data, proxyUrl) }
if (Array.isArray(document.accounts)) {
return {
...document,
accounts: document.accounts.map(account => isRecord(account) ? withDefaultImportProxy(account, proxyUrl) : account),
}
}
if (['outboundProxyUrl', 'outbound_proxy_url', 'proxy_key'].some(key => Object.hasOwn(document, key)))
return document
return { ...document, outboundProxyUrl: proxyUrl }
}
function parseImportJson(value: string) {
try {
return JSON.parse(value)
+4 -4
View File
@@ -129,7 +129,7 @@ const {
concurrencyLimit: batchConcurrencyLimit,
weight: batchWeight,
proxyMode: batchProxyMode,
proxyUrl: batchProxyUrl,
proxyId: batchProxyId,
selectedGroupIds: batchGroupIds,
saving: savingBatchEdit,
open: openBatchEdit,
@@ -148,7 +148,7 @@ const {
concurrencyLimit: editingConcurrencyLimit,
weight: editingWeight,
proxyMode: editingProxyMode,
proxyUrl: editingProxyUrl,
proxyId: editingProxyId,
selectedGroupIds: editingGroupIds,
saving: savingAccountEdit,
open: openAccountEdit,
@@ -350,7 +350,7 @@ const {
v-model:concurrency-limit="editingConcurrencyLimit"
v-model:weight="editingWeight"
v-model:proxy-mode="editingProxyMode"
v-model:proxy-url="editingProxyUrl"
v-model:proxy-id="editingProxyId"
v-model:selected-group-ids="editingGroupIds"
:account="editingAccount"
:groups="groups"
@@ -365,7 +365,7 @@ const {
v-model:concurrency-limit="batchConcurrencyLimit"
v-model:weight="batchWeight"
v-model:proxy-mode="batchProxyMode"
v-model:proxy-url="batchProxyUrl"
v-model:proxy-id="batchProxyId"
v-model:selected-group-ids="batchGroupIds"
:selected-count="selectedIds.size"
:groups="groups"
+275
View File
@@ -0,0 +1,275 @@
<script setup lang="ts">
import type { OutboundProxyRecord } from '@/api'
import { Eye, EyeOff, LockKeyhole, Pencil, Plus, RefreshCw, Save, Search, Trash2, Users, Wifi } from '@lucide/vue'
import { watchDebounced } from '@vueuse/core'
import { computed, onMounted, reactive, ref, shallowRef, watch } from 'vue'
import { RouterLink } from 'vue-router'
import { createProxy, deleteProxy, getProxies, testProxy, updateProxy } from '@/api'
import BaseButton from '@/components/base/BaseButton.vue'
import BaseConfirmModal from '@/components/base/BaseConfirmModal.vue'
import BaseFormItem from '@/components/base/BaseForm/FormItem.vue'
import BaseIconButton from '@/components/base/BaseIconButton.vue'
import BaseInput from '@/components/base/BaseInput.vue'
import BaseModal from '@/components/base/BaseModal/index.vue'
import BasePageHeader from '@/components/base/BasePageHeader.vue'
import BaseSwitch from '@/components/base/BaseSwitch.vue'
import BaseTablePagination from '@/components/base/BaseTable/BaseTablePagination.vue'
import { defineTableColumns } from '@/components/base/BaseTable/columns'
import BaseTable from '@/components/base/BaseTable/index.vue'
import { toast } from '@/components/base/BaseToast'
import { useAsyncAction } from '@/composables/useAsyncAction'
import { usePagedQuery } from '@/composables/usePagedQuery'
import { errorMessage } from '@/utils/async'
import { formatDateTime } from '@/utils/date'
const search = shallowRef('')
const query = usePagedQuery({
initialPageSize: 20,
load: pagination => getProxies({ ...pagination, search: search.value.trim() || undefined }),
onError: error => toast.error(errorMessage(error, '代理列表加载失败')),
})
const { items: proxies, loading } = query
const pagination = computed(() => ({ currentPage: query.page.value, pageSize: query.pageSize.value, total: query.total.value }))
const columns = defineTableColumns<OutboundProxyRecord>([
{ key: 'identity', label: '代理', kind: 'identity' },
{ key: 'status', label: '测试状态', kind: 'custom', size: 'lg' },
{ key: 'exitIp', label: '出口 IP', kind: 'custom' },
{ key: 'latency', label: '耗时', kind: 'custom', size: 'sm' },
{ key: 'accounts', label: '关联账号', kind: 'custom', size: 'sm' },
{ key: 'testedAt', label: '测试时间', kind: 'datetime' },
{ key: 'actions', label: '操作', kind: 'actions', size: 'lg' },
])
const showForm = shallowRef(false)
const editing = shallowRef<OutboundProxyRecord | null>(null)
const form = reactive({ name: '', proxyUrl: '' })
const replaceConnection = ref(true)
const showSecret = ref(false)
const saveAction = useAsyncAction()
const { loading: saving } = saveAction
const deleteAction = useAsyncAction()
const { loading: deleting } = deleteAction
const showDelete = shallowRef(false)
const pendingDelete = shallowRef<OutboundProxyRecord | null>(null)
const testingIds = ref(new Set<string>())
const showAccounts = shallowRef(false)
const inspected = shallowRef<OutboundProxyRecord | null>(null)
function openForm(proxy: OutboundProxyRecord | null = null) {
editing.value = proxy
form.name = proxy?.name ?? ''
form.proxyUrl = ''
replaceConnection.value = !proxy
showSecret.value = false
showForm.value = true
}
async function checkProxy(proxy: OutboundProxyRecord) {
if (testingIds.value.has(proxy.id))
return
testingIds.value.add(proxy.id)
try {
const result = await testProxy({ id: proxy.id, revision: proxy.revision })
if (result.lastTest?.success)
toast.success(`${result.name}:连接成功`)
else
toast.error(result.lastTest?.message ?? '代理测试失败')
}
catch (error) {
toast.error(errorMessage(error, '代理测试失败'))
}
finally {
testingIds.value.delete(proxy.id)
await query.execute({ silent: true })
}
}
async function save(testAfter: boolean) {
if (saving.value)
return
const name = form.name.trim()
const proxyUrl = form.proxyUrl.trim()
if (!name || (replaceConnection.value && !proxyUrl)) {
toast.warning('请填写代理名称和连接地址')
return
}
await saveAction.run(async () => {
const result = editing.value
? await updateProxy({ id: editing.value.id, revision: editing.value.revision, name, proxyUrl: replaceConnection.value ? proxyUrl : undefined })
: await createProxy({ name, proxyUrl })
showForm.value = false
form.proxyUrl = ''
toast.success('代理已保存')
search.value = ''
query.page.value = 1
await query.execute()
if (testAfter)
await checkProxy(result.record)
}, { errorText: '代理保存失败' })
}
function requestDelete(proxy: OutboundProxyRecord) {
pendingDelete.value = proxy
showDelete.value = true
}
async function confirmDelete() {
const proxy = pendingDelete.value
if (!proxy || deleting.value)
return
await deleteAction.run(async () => {
await deleteProxy({ id: proxy.id, revision: proxy.revision })
showDelete.value = false
await query.execute()
toast.success('代理已删除')
}, { errorText: '代理删除失败,请确认没有账号使用该代理' })
}
function setPage(page: number) {
query.page.value = page
void query.execute()
}
function setPageSize(size: number) {
query.pageSize.value = size
setPage(1)
}
watch(showForm, (open) => {
if (!open) {
form.proxyUrl = ''
showSecret.value = false
}
})
watchDebounced(search, () => setPage(1), { debounce: 300 })
onMounted(() => void query.execute())
</script>
<template>
<div class="flex h-full min-h-0 w-full flex-col overflow-hidden">
<BasePageHeader title="代理管理" />
<section class="mt-5 flex min-h-125 flex-1 flex-col bg-cp-bg-container p-5">
<div class="mb-4 flex flex-wrap items-center justify-between gap-3">
<BaseInput v-model="search" class="w-full sm:max-w-80" aria-label="搜索代理" placeholder="搜索代理名称">
<template #prefix>
<Search class="size-4" />
</template>
</BaseInput>
<div class="ml-auto flex items-center gap-2">
<BaseIconButton label="刷新代理列表" :disabled="loading" @click="query.execute()">
<RefreshCw class="size-4" />
</BaseIconButton>
<BaseButton variant="primary" @click="openForm()">
<Plus class="size-4" />新增代理
</BaseButton>
</div>
</div>
<BaseTable class="min-h-0 flex-1" :columns="columns" :rows="proxies" :loading="loading" empty-text="暂无代理">
<template #identity="{ row }">
<div class="grid min-w-0 gap-1">
<strong class="truncate text-cp text-cp-text" :title="row.name">{{ row.name }}</strong>
<span class="flex min-w-0 items-center gap-1 text-cp-xs text-cp-text-secondary">
<LockKeyhole v-if="row.hasAuthentication" class="size-3 shrink-0" aria-label="已保存代理认证" />
<span class="truncate font-mono" :title="row.endpoint">{{ row.endpoint }}</span>
</span>
</div>
</template>
<template #status="{ row }">
<div class="grid min-w-0 gap-1">
<span class="text-cp-sm font-semibold" :class="row.lastTest?.success ? 'text-cp-success' : row.lastTest ? 'text-cp-error' : 'text-cp-text-quaternary'">
{{ testingIds.has(row.id) ? '测试中' : row.lastTest?.success ? '通过' : row.lastTest ? '失败' : '未测试' }}
</span>
<span v-if="row.lastTest && !row.lastTest.success" class="truncate text-cp-xs text-cp-text-secondary" :title="row.lastTest.message">{{ row.lastTest.message }}</span>
</div>
</template>
<template #exitIp="{ row }">
<span class="break-all font-mono text-cp-xs">{{ row.lastTest?.exitIp ?? '-' }}</span>
</template>
<template #latency="{ row }">
{{ row.lastTest ? `${row.lastTest.latencyMs} ms` : '-' }}
</template>
<template #accounts="{ row }">
<BaseButton variant="ghost" :aria-label="`查看 ${row.name} 的关联账号`" @click="inspected = row; showAccounts = true">
<Users class="size-4" />{{ row.accounts.length }}
</BaseButton>
</template>
<template #testedAt="{ row }">
{{ row.lastTestAt ? formatDateTime(row.lastTestAt) : '-' }}
</template>
<template #actions="{ row }">
<div class="flex items-center gap-1">
<BaseIconButton label="测试代理" :loading="testingIds.has(row.id)" :disabled="testingIds.has(row.id)" @click="checkProxy(row)">
<Wifi class="size-4 text-cp-link" />
</BaseIconButton>
<BaseIconButton label="编辑代理" :disabled="testingIds.has(row.id)" @click="openForm(row)">
<Pencil class="size-4 text-cp-link" />
</BaseIconButton>
<BaseIconButton :label="row.accounts.length ? '代理正在被账号使用' : '删除代理'" :disabled="row.accounts.length > 0 || testingIds.has(row.id)" @click="requestDelete(row)">
<Trash2 class="size-4 text-cp-error" />
</BaseIconButton>
</div>
</template>
</BaseTable>
<BaseTablePagination :pagination="pagination" :loading="loading" @page-change="setPage" @page-size-change="setPageSize" />
</section>
<BaseModal v-model="showForm" :title="editing ? '编辑代理' : '新增代理'" size="md" :dismissible="!saving">
<div class="grid gap-5">
<BaseFormItem label="代理名称" required>
<BaseInput v-model="form.name" maxlength="100" :disabled="saving" aria-label="代理名称" />
</BaseFormItem>
<template v-if="editing">
<p class="m-0 break-all font-mono text-cp-sm text-cp-text-secondary">
{{ editing.endpoint }}
</p>
<div class="flex items-center justify-between gap-3">
<span class="text-cp text-cp-text-secondary">更换连接配置</span>
<BaseSwitch v-model="replaceConnection" label="更换连接配置" :disabled="saving" />
</div>
</template>
<BaseFormItem v-if="replaceConnection" label="代理 URL" required>
<BaseInput v-model="form.proxyUrl" :type="showSecret ? 'text' : 'password'" autocomplete="new-password" :disabled="saving" aria-label="代理 URL" placeholder="socks5h://user:password@host:1080">
<template #suffix>
<BaseIconButton :label="showSecret ? '隐藏代理地址' : '显示代理地址'" @click="showSecret = !showSecret">
<EyeOff v-if="showSecret" class="size-4" /><Eye v-else class="size-4" />
</BaseIconButton>
</template>
</BaseInput>
</BaseFormItem>
<p v-if="editing?.accounts.length && replaceConnection" class="m-0 text-cp-sm text-cp-warning-text">
将更新 {{ editing.accounts.length }} 个关联账号的出口。
</p>
</div>
<template #footer>
<BaseButton variant="ghost" :disabled="saving" @click="showForm = false">
取消
</BaseButton>
<BaseButton variant="secondary" :loading="saving" @click="save(false)">
<Save class="size-4" />保存代理
</BaseButton>
<BaseButton variant="primary" :loading="saving" @click="save(true)">
<Wifi class="size-4" />保存并测试
</BaseButton>
</template>
</BaseModal>
<BaseConfirmModal v-model="showDelete" title="删除代理" destructive :loading="deleting" @confirm="confirmDelete">
<p class="m-0">
确定删除“{{ pendingDelete?.name }}”吗?
</p>
</BaseConfirmModal>
<BaseModal v-model="showAccounts" title="关联账号" size="md">
<ul v-if="inspected?.accounts.length" class="m-0 grid list-none gap-3 p-0">
<li v-for="account in inspected.accounts" :key="account.id" class="break-all text-cp">
{{ account.name }}
</li>
</ul>
<p v-else class="m-0 text-cp-text-secondary">
暂无关联账号
</p>
<template #footer>
<RouterLink to="/accounts" class="text-cp-link">
账号管理
</RouterLink>
</template>
</BaseModal>
</div>
</template>