diff --git a/backend/Cargo.lock b/backend/Cargo.lock index d950f6fa7..344983e7a 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -744,6 +744,7 @@ dependencies = [ "tempfile", "thiserror", "tokio", + "tokio-rustls", ] [[package]] diff --git a/backend/apps/gateway/Cargo.toml b/backend/apps/gateway/Cargo.toml index 6f9131b78..fdfca651c 100644 --- a/backend/apps/gateway/Cargo.toml +++ b/backend/apps/gateway/Cargo.toml @@ -28,6 +28,7 @@ config = { version = "0.15.23", default-features = false, features = ["yaml"] } serde_json = "1" syn = { version = "3.0.3", features = ["full", "visit"] } tempfile = "3.27.0" +tokio-rustls = { version = "=0.26.4", default-features = false } [lints] workspace = true diff --git a/backend/apps/gateway/src/bootstrap.rs b/backend/apps/gateway/src/bootstrap.rs index c74b3131b..b01a91c75 100644 --- a/backend/apps/gateway/src/bootstrap.rs +++ b/backend/apps/gateway/src/bootstrap.rs @@ -78,7 +78,10 @@ pub async fn run() -> Result<(), BootstrapError> { store.admin_ports(), vec![openai.admin_provider(), xai.admin_provider()], core.snapshot_control(), - (core.account_probe(), host.proxy_probe()), + ( + core.account_probe(), + host.proxy_probe(provider_openai::build_reqwest_client_with_custom_ca), + ), host.client_distribution_resolver(), host.system_operations(), ) diff --git a/backend/apps/gateway/tests/bootstrap.rs b/backend/apps/gateway/tests/bootstrap.rs index 1b57b9211..4f694a115 100644 --- a/backend/apps/gateway/tests/bootstrap.rs +++ b/backend/apps/gateway/tests/bootstrap.rs @@ -9,6 +9,151 @@ const REDIS_PASSWORD: &str = "222222222222222222222222222222222222222222222222"; const ADMIN_PASSWORD: &str = "test-admin-password"; const TOPOLOGY_CHILD_ENV: &str = "CPR_TEST_TOPOLOGY_CHILD"; +#[tokio::test] +async fn proxy_probe_should_use_provider_custom_ca_for_https_proxies() { + use gateway_admin::ports::proxy::ProxyProbe; + use gateway_core::account::OutboundProxy; + use gateway_host::proxy_probe::HttpProxyProbe; + use std::{sync::Arc, time::Duration}; + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + use tokio_rustls::{ + TlsAcceptor, + rustls::{ + ServerConfig, + pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject as _}, + }, + }; + + const CHILD_ENV: &str = "CPR_TEST_PROXY_TLS_DIRECTORY"; + let Ok(directory) = std::env::var(CHILD_ENV) else { + let directory = tempfile::tempdir().unwrap(); + generate_proxy_test_certificates(directory.path()); + // 使用子进程隔离环境变量,避免并行测试读取到临时 CA 配置。 + for ca_env in ["CODEX_CA_CERTIFICATE", "SSL_CERT_FILE"] { + let mut child = Command::new(std::env::current_exe().unwrap()); + child + .args([ + "--exact", + "bootstrap::proxy_probe_should_use_provider_custom_ca_for_https_proxies", + "--nocapture", + ]) + .env(CHILD_ENV, directory.path()) + .env_remove("CODEX_CA_CERTIFICATE") + .env_remove("SSL_CERT_FILE") + .env(ca_env, directory.path().join("ca.pem")); + if ca_env == "CODEX_CA_CERTIFICATE" { + child.env( + "SSL_CERT_FILE", + directory.path().join("missing-fallback.pem"), + ); + } + let output = child.output().unwrap(); + assert!( + output.status.success(), + "{ca_env}: {}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + } + return; + }; + provider_openai::ensure_rustls_provider(); + let directory = std::path::Path::new(&directory); + let certificate = CertificateDer::from_pem_file(directory.join("server.pem")).unwrap(); + let key = PrivateKeyDer::from_pem_file(directory.join("server.key")).unwrap(); + let acceptor = TlsAcceptor::from(Arc::new( + ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(vec![certificate], key) + .unwrap(), + )); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy = + OutboundProxy::parse(&format!("https://{}", listener.local_addr().unwrap())).unwrap(); + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut stream = acceptor.accept(socket).await.unwrap(); + let mut request = Vec::new(); + while !request.ends_with(b"\r\n\r\n") { + assert!(request.len() < 8192); + request.push(stream.read_u8().await.unwrap()); + } + assert!(request.starts_with(b"GET http://unresolvable.invalid/ip ")); + let body = "{\"ip\":\"203.0.113.8\"}"; + stream.write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()).await.unwrap(); + stream.shutdown().await.unwrap(); + }); + let probe = HttpProxyProbe::new("http://unresolvable.invalid/ip") + .with_client_builder(provider_openai::build_reqwest_client_with_custom_ca); + let result = probe.test(&proxy).await; + assert!(result.success, "{}", result.message); + assert_eq!(result.exit_ip.unwrap().to_string(), "203.0.113.8"); + tokio::time::timeout(Duration::from_secs(5), server) + .await + .unwrap() + .unwrap(); +} + +fn generate_proxy_test_certificates(directory: &std::path::Path) { + let openssl = |args: &[&str]| { + let output = Command::new("openssl") + .args(args) + .current_dir(directory) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + }; + openssl(&[ + "req", + "-x509", + "-newkey", + "rsa:2048", + "-nodes", + "-keyout", + "ca.key", + "-out", + "ca.pem", + "-days", + "1", + "-subj", + "/CN=Proxy Test CA", + ]); + openssl(&[ + "req", + "-newkey", + "rsa:2048", + "-nodes", + "-keyout", + "server.key", + "-out", + "server.csr", + "-subj", + "/CN=localhost", + ]); + fs::write(directory.join("extensions"), "subjectAltName=IP:127.0.0.1\nbasicConstraints=critical,CA:FALSE\nkeyUsage=digitalSignature,keyEncipherment\nextendedKeyUsage=serverAuth\n").unwrap(); + openssl(&[ + "x509", + "-req", + "-in", + "server.csr", + "-CA", + "ca.pem", + "-CAkey", + "ca.key", + "-CAcreateserial", + "-out", + "server.pem", + "-days", + "1", + "-extfile", + "extensions", + ]); +} + #[test] fn config_loader_should_load_complete_terminal_example() { parse_config(&valid_config()).expect("terminal config example"); diff --git a/backend/crates/gateway-admin/src/model/proxies.rs b/backend/crates/gateway-admin/src/model/proxies.rs index f5c11eaf0..56fd4e006 100644 --- a/backend/crates/gateway-admin/src/model/proxies.rs +++ b/backend/crates/gateway-admin/src/model/proxies.rs @@ -1,4 +1,4 @@ -//! Reusable account egress configurations and redacted connectivity observations. +//! 可复用的账号出口配置,以及脱敏后的连通性测试结果。 use chrono::{DateTime, Utc}; use gateway_core::account::OutboundProxy; diff --git a/backend/crates/gateway-admin/src/ports/proxy.rs b/backend/crates/gateway-admin/src/ports/proxy.rs index 5e7ecba57..d58ad8429 100644 --- a/backend/crates/gateway-admin/src/ports/proxy.rs +++ b/backend/crates/gateway-admin/src/ports/proxy.rs @@ -5,13 +5,15 @@ use super::store::AdminStoreResult; use crate::model::{ MutationContext, Revision, proxies::{ - NewProxy, ProxyListQuery, ProxyMutation, ProxyPage, ProxyRecord, ProxyTestResult, - UpdateProxy, + ImportProxyBinding, NewProxy, ProxyListQuery, ProxyMutation, ProxyPage, ProxyRecord, + ProxyTestResult, UpdateProxy, }, }; #[async_trait] pub trait ProxyStore: Send + Sync { + /// 在凭据交换到提交期间保护选定代理的连接配置和测试结果。 + async fn reserve_import(&self, id: &str) -> AdminStoreResult; async fn list(&self, query: ProxyListQuery) -> AdminStoreResult; async fn get(&self, id: &str) -> AdminStoreResult; async fn create( @@ -39,6 +41,14 @@ pub trait ProxyStore: Send + Sync { ) -> AdminStoreResult; } +/// 离开作用域时释放保护,错误返回和请求取消也遵循相同规则。 +pub trait ProxyImportGuard: Send + Sync {} + +pub struct ProxyImportReservation { + pub binding: ImportProxyBinding, + pub guard: Box, +} + #[async_trait] pub trait ProxyProbe: Send + Sync { async fn test(&self, proxy: &OutboundProxy) -> ProxyTestResult; diff --git a/backend/crates/gateway-admin/src/use_case/mod.rs b/backend/crates/gateway-admin/src/use_case/mod.rs index a200f1152..6afc9288a 100644 --- a/backend/crates/gateway-admin/src/use_case/mod.rs +++ b/backend/crates/gateway-admin/src/use_case/mod.rs @@ -113,19 +113,13 @@ async fn required_credential( async fn import_proxy_binding( proxies: &dyn crate::ports::proxy::ProxyStore, id: Option<&str>, -) -> Result, AdminError> { +) -> Result, AdminError> { let Some(id) = id else { return Ok(None) }; - let record = proxies - .get(id) + let reservation = proxies + .reserve_import(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, - })) + Ok(Some(reservation)) } async fn pending_authorization( diff --git a/backend/crates/gateway-admin/src/use_case/openai.rs b/backend/crates/gateway-admin/src/use_case/openai.rs index 5923f2d4f..cf2bc1e59 100644 --- a/backend/crates/gateway-admin/src/use_case/openai.rs +++ b/backend/crates/gateway-admin/src/use_case/openai.rs @@ -131,11 +131,14 @@ impl OpenAiService for DefaultOpenAiService { command: ImportCredentials, ) -> Result { let context = command.context; - let outbound_proxy = super::import_proxy_binding( + let proxy_reservation = super::import_proxy_binding( self.proxies.as_ref(), command.outbound_proxy_id.as_deref(), ) .await?; + let outbound_proxy = proxy_reservation + .as_ref() + .map(|reservation| reservation.binding.clone()); let prepared = self .provider .prepare_import(PrepareCredentialImport { @@ -163,6 +166,7 @@ impl OpenAiService for DefaultOpenAiService { ) .await .map_err(|error| map_store_error(error, "OpenAI credential import"))?; + drop(proxy_reservation); self.provider .account_facts_changed(&result.credential_ids) .await; diff --git a/backend/crates/gateway-admin/src/use_case/xai.rs b/backend/crates/gateway-admin/src/use_case/xai.rs index eaf3e5711..b042ce6a4 100644 --- a/backend/crates/gateway-admin/src/use_case/xai.rs +++ b/backend/crates/gateway-admin/src/use_case/xai.rs @@ -84,11 +84,14 @@ impl XaiService for DefaultXaiService { command: ImportCredentials, ) -> Result { let context = command.context; - let outbound_proxy = super::import_proxy_binding( + let proxy_reservation = super::import_proxy_binding( self.proxies.as_ref(), command.outbound_proxy_id.as_deref(), ) .await?; + let outbound_proxy = proxy_reservation + .as_ref() + .map(|reservation| reservation.binding.clone()); let prepared = self .provider .prepare_import(PrepareCredentialImport { @@ -116,6 +119,7 @@ impl XaiService for DefaultXaiService { ) .await .map_err(|error| map_store_error(error, "xAI credential import"))?; + drop(proxy_reservation); publish_committed(self.snapshot.as_ref(), result.config_revision).await?; // 导入已经提交,额度属于可重建观察事实;单个账号查询失败由周期任务重试。 for account_id in &result.credential_ids { diff --git a/backend/crates/gateway-admin/tests/use_case/mod.rs b/backend/crates/gateway-admin/tests/use_case/mod.rs index 2bb5bbb84..b5b693b50 100644 --- a/backend/crates/gateway-admin/tests/use_case/mod.rs +++ b/backend/crates/gateway-admin/tests/use_case/mod.rs @@ -83,6 +83,7 @@ pub(super) struct AdminHarness { default_password: String, session_ttl_minutes: u64, accounts: Arc, + proxies: Arc, account_runtime: Arc, account_groups: Arc, auth: Arc, @@ -102,6 +103,7 @@ impl AdminHarness { default_password: "strong-test-password".to_owned(), session_ttl_minutes: 60, accounts: unavailable.clone(), + proxies: Arc::new(proxies::TestProxies::default()), account_runtime: unavailable.clone(), account_groups: Arc::new(UnavailableAccountGroupStore), auth: Arc::new(BootstrapAuthStore::default()), @@ -180,6 +182,14 @@ impl AdminHarness { self } + pub(super) fn proxies( + mut self, + proxies: Arc, + ) -> Self { + self.proxies = proxies; + self + } + pub(super) fn system(mut self, system: Arc) -> Self { self.system = system; self @@ -197,7 +207,7 @@ impl AdminHarness { self.accounts, self.account_runtime, self.account_groups, - Arc::new(proxies::UnavailableProxies), + self.proxies, ), self.auth, self.client_keys, @@ -207,7 +217,7 @@ impl AdminHarness { ), self.providers, Arc::new(NoopSnapshot), - (self.probe, Arc::new(proxies::UnavailableProxies)), + (self.probe, Arc::new(proxies::TestProxies::default())), Arc::new(NoopClientDistribution), self.system, ) diff --git a/backend/crates/gateway-admin/tests/use_case/proxies.rs b/backend/crates/gateway-admin/tests/use_case/proxies.rs index 5d3ea1e95..38bf02b2a 100644 --- a/backend/crates/gateway-admin/tests/use_case/proxies.rs +++ b/backend/crates/gateway-admin/tests/use_case/proxies.rs @@ -8,10 +8,40 @@ use gateway_admin::{ }; use gateway_core::account::OutboundProxy; -pub(super) struct UnavailableProxies; +#[derive(Default)] +pub(super) struct TestProxies { + pub events: Option, +} + +struct ImportGuard(super::accounts::EventLog); + +impl gateway_admin::ports::proxy::ProxyImportGuard for ImportGuard {} + +impl Drop for ImportGuard { + fn drop(&mut self) { + self.0.lock().unwrap().push("proxy.release"); + } +} #[async_trait] -impl ProxyStore for UnavailableProxies { +impl ProxyStore for TestProxies { + async fn reserve_import( + &self, + id: &str, + ) -> AdminStoreResult { + let events = self + .events + .as_ref() + .ok_or_else(|| super::unavailable("proxy"))?; + events.lock().unwrap().push("proxy.reserve"); + Ok(gateway_admin::ports::proxy::ProxyImportReservation { + binding: ImportProxyBinding { + id: id.to_owned(), + proxy: OutboundProxy::parse("http://127.0.0.1:8080").unwrap(), + }, + guard: Box::new(ImportGuard(events.clone())), + }) + } async fn list(&self, _: ProxyListQuery) -> AdminStoreResult { Err(super::unavailable("proxy")) } @@ -44,8 +74,64 @@ impl ProxyStore for UnavailableProxies { } #[async_trait] -impl ProxyProbe for UnavailableProxies { +impl ProxyProbe for TestProxies { async fn test(&self, _: &OutboundProxy) -> ProxyTestResult { panic!("unexpected proxy probe") } } + +#[tokio::test] +async fn credential_import_keeps_proxy_reserved_until_commit_and_releases_on_errors() { + use super::accounts::{ + FakeAccountStore, FakeProviderAdmin, context, document, events, recorded, + }; + use gateway_admin::model::provider_credentials::ImportCredentials; + use gateway_admin::ports::provider::ProviderAdminErrorKind; + use std::sync::Arc; + + for kind in ["openai", "xai"] { + for failure in [None, Some("prepare"), Some("commit")] { + let events = events(); + let provider = FakeProviderAdmin::new(kind, events.clone()); + let store = FakeAccountStore::new(kind, events.clone()); + if failure == Some("prepare") { + provider.fail_next(ProviderAdminErrorKind::Unavailable); + } + if failure == Some("commit") { + store.fail_next_commit(); + } + let services = super::AdminHarness::new() + .provider(provider) + .accounts(store) + .proxies(Arc::new(TestProxies { + events: Some(events.clone()), + })) + .build() + .await; + let command = ImportCredentials { + outbound_proxy_id: Some("proxy_import".to_owned()), + settings: None, + context: context("reserved-import"), + document: document(), + }; + let result = if kind == "openai" { + services.openai().import_document(command).await + } else { + services.xai().import_document(command).await + }; + assert_eq!(result.is_err(), failure.is_some()); + let events = recorded(&events); + let expected = if failure == Some("prepare") { + vec!["proxy.reserve", "provider.prepare_import", "proxy.release"] + } else { + vec![ + "proxy.reserve", + "provider.prepare_import", + "store.commit_import", + "proxy.release", + ] + }; + assert_eq!(&events[..expected.len()], expected); + } + } +} diff --git a/backend/crates/gateway-api/tests/admin/proxies.rs b/backend/crates/gateway-api/tests/admin/proxies.rs index 35d5aff2f..9fa3cfce5 100644 --- a/backend/crates/gateway-api/tests/admin/proxies.rs +++ b/backend/crates/gateway-api/tests/admin/proxies.rs @@ -28,6 +28,12 @@ fn missing() -> AdminStoreError { #[async_trait] impl ProxyStore for MemoryProxies { + async fn reserve_import( + &self, + _: &str, + ) -> AdminStoreResult { + Err(missing()) + } async fn list(&self, query: ProxyListQuery) -> AdminStoreResult { let items: Vec<_> = self.0.lock().unwrap().iter().cloned().collect(); Ok(ProxyPage { diff --git a/backend/crates/gateway-host/src/lib.rs b/backend/crates/gateway-host/src/lib.rs index 37e0e8362..92d523855 100644 --- a/backend/crates/gateway-host/src/lib.rs +++ b/backend/crates/gateway-host/src/lib.rs @@ -83,8 +83,14 @@ impl HostBundle { } #[must_use] - pub fn proxy_probe(&self) -> Arc { - Arc::new(proxy_probe::HttpProxyProbe::default()) + pub fn proxy_probe( + &self, + build_client: impl Fn(reqwest::ClientBuilder) -> Result + + Send + + Sync + + 'static, + ) -> Arc { + Arc::new(proxy_probe::HttpProxyProbe::default().with_client_builder(build_client)) } #[must_use] diff --git a/backend/crates/gateway-host/src/proxy_probe.rs b/backend/crates/gateway-host/src/proxy_probe.rs index 61308cdc2..772486503 100644 --- a/backend/crates/gateway-host/src/proxy_probe.rs +++ b/backend/crates/gateway-host/src/proxy_probe.rs @@ -1,7 +1,8 @@ -//! Bounded egress checks using the same explicit proxy protocols as provider requests. +//! 使用与 Provider 请求一致的显式代理协议,执行有超时和响应大小限制的出口测试。 use std::{ net::IpAddr, + sync::Arc, time::{Duration, Instant}, }; @@ -12,8 +13,12 @@ use serde::Deserialize; pub struct HttpProxyProbe { endpoint: String, + build_client: Arc, } +type ProxyClientBuilder = + dyn Fn(reqwest::ClientBuilder) -> Result + Send + Sync; + impl Default for HttpProxyProbe { fn default() -> Self { Self::new("https://api.ipify.org?format=json") @@ -25,19 +30,31 @@ impl HttpProxyProbe { pub fn new(endpoint: impl Into) -> Self { Self { endpoint: endpoint.into(), + build_client: Arc::new(|builder| builder.build().map_err(|_| "无法创建代理连接")), } } + /// 由组合根注入与 Provider 请求一致的证书信任策略。 + #[must_use] + pub fn with_client_builder( + mut self, + build: impl Fn(reqwest::ClientBuilder) -> Result + Send + Sync + 'static, + ) -> Self { + self.build_client = Arc::new(move |builder| { + build(builder).map_err(|_| "无法创建代理连接,请检查证书信任配置") + }); + self + } + async fn exit_ip(&self, proxy: &OutboundProxy) -> Result { let proxy = reqwest::Proxy::all(proxy.expose_url()).map_err(|_| "代理地址不合法")?; - let client = reqwest::Client::builder() + let builder = 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(|_| "无法创建代理连接")?; + .redirect(reqwest::redirect::Policy::none()); + let client = (self.build_client)(builder)?; let mut response = client.get(&self.endpoint).send().await.map_err(|error| { if error.is_timeout() { "代理连接超时" diff --git a/backend/crates/gateway-host/tests/proxy_probe.rs b/backend/crates/gateway-host/tests/proxy_probe.rs index a253bf6cd..6845b9726 100644 --- a/backend/crates/gateway-host/tests/proxy_probe.rs +++ b/backend/crates/gateway-host/tests/proxy_probe.rs @@ -60,3 +60,19 @@ async fn unavailable_proxy_never_falls_back_to_direct_connection() { let result = HttpProxyProbe::new(target.uri()).test(&proxy).await; assert!(!result.success); } + +#[tokio::test] +async fn invalid_certificate_configuration_should_not_fall_back_or_expose_details() { + let target = MockServer::start().await; + Mock::given(any()) + .respond_with(ResponseTemplate::new(200)) + .expect(0) + .mount(&target) + .await; + let result = HttpProxyProbe::new(target.uri()) + .with_client_builder(|_| Err("private-certificate-path")) + .test(&OutboundProxy::parse(&target.uri()).unwrap()) + .await; + assert!(!result.success); + assert!(!result.message.contains("private-certificate-path")); +} diff --git a/backend/crates/gateway-store/src/postgres/proxies.rs b/backend/crates/gateway-store/src/postgres/proxies.rs index 248ee0678..c04cb371d 100644 --- a/backend/crates/gateway-store/src/postgres/proxies.rs +++ b/backend/crates/gateway-store/src/postgres/proxies.rs @@ -1,15 +1,18 @@ -//! Named proxies own credentials; accounts retain the resolved URL for provider transports. +//! 命名代理持有认证信息,账号保留解析后的 URL 供 Provider 传输使用。 + +use std::sync::Arc; use async_trait::async_trait; use gateway_admin::{ model::{MutationContext, Revision as AdminRevision, proxies::*}, ports::{ - proxy::ProxyStore, + proxy::{ProxyImportGuard, ProxyImportReservation, ProxyStore}, store::{AdminStoreError, AdminStoreResult}, }, }; use gateway_core::account::OutboundProxy; -use sqlx::{PgPool, Postgres, QueryBuilder, Row as _, Transaction, postgres::PgRow}; +use sqlx::{PgConnection, PgPool, Postgres, QueryBuilder, Row as _, Transaction, postgres::PgRow}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use super::{append_admin_audit_event_in_transaction, bump_config_revision_in_transaction}; use crate::{ @@ -26,15 +29,44 @@ const SELECT: &str = "select p.*, #[derive(Clone)] pub struct PgProxyRepository { pool: PgPool, + import_slots: Arc, } impl PgProxyRepository { #[must_use] - pub const fn new(pool: PgPool) -> Self { - Self { pool } + pub fn new(pool: PgPool) -> Self { + Self { + pool, + import_slots: Arc::new(Semaphore::new(4)), + } } } +struct PgProxyImportGuard { + // 独立会话持有咨询锁,避免空闲事务超时,也不给提交事务占用的连接池制造死锁。 + // 每个仓储最多增加四个会话;取消请求或进程退出时,关闭连接会自动释放锁。 + _connection: PgConnection, + _slot: OwnedSemaphorePermit, +} + +impl ProxyImportGuard for PgProxyImportGuard {} + +async fn exclude_active_imports( + transaction: &mut Transaction<'_, Postgres>, + id: &str, +) -> StoreResult<()> { + let acquired: bool = + sqlx::query_scalar("select pg_try_advisory_xact_lock(hashtextextended($1, 739219))") + .bind(id) + .fetch_one(&mut **transaction) + .await + .map_err(|_| unavailable())?; + if !acquired { + return Err(conflict(id)); + } + Ok(()) +} + fn store_error(error: StoreError) -> AdminStoreError { admin_store_error(ENTITY, error) } @@ -111,7 +143,7 @@ async fn lock_url( Ok(()) } -/// Imports and legacy URL writes join the catalog in the same account transaction. +/// 导入和旧版 URL 写入在账号事务内登记到共享代理目录。 pub(crate) async fn ensure_proxy_for_url( transaction: &mut Transaction<'_, Postgres>, proxy: &OutboundProxy, @@ -191,6 +223,43 @@ async fn audit( #[async_trait] impl ProxyStore for PgProxyRepository { + async fn reserve_import(&self, id: &str) -> AdminStoreResult { + let slot = self + .import_slots + .clone() + .try_acquire_owned() + .map_err(|_| store_error(conflict(id)))?; + let mut connection = self + .pool + .acquire() + .await + .map_err(|_| store_error(unavailable()))? + .detach(); + let acquired: bool = + sqlx::query_scalar("select pg_try_advisory_lock_shared(hashtextextended($1, 739219))") + .bind(id) + .fetch_one(&mut connection) + .await + .map_err(|_| store_error(unavailable()))?; + if !acquired { + return Err(store_error(conflict(id))); + } + let record = self.get(id).await?; + if !record.last_test.is_some_and(|test| test.success) { + return Err(store_error(conflict(id))); + } + Ok(ProxyImportReservation { + binding: ImportProxyBinding { + id: record.id, + proxy: record.proxy, + }, + guard: Box::new(PgProxyImportGuard { + _connection: connection, + _slot: slot, + }), + }) + } + async fn list(&self, query: ProxyListQuery) -> AdminStoreResult { if query.page == 0 { return Err(store_error(invalid())); @@ -294,6 +363,9 @@ impl ProxyStore for PgProxyRepository { .begin() .await .map_err(|_| store_error(unavailable()))?; + exclude_active_imports(&mut transaction, &command.id) + .await + .map_err(store_error)?; let revision = bump_config_revision_in_transaction(&mut transaction) .await .map_err(store_error)?; @@ -360,6 +432,9 @@ impl ProxyStore for PgProxyRepository { .begin() .await .map_err(|_| store_error(unavailable()))?; + exclude_active_imports(&mut transaction, id) + .await + .map_err(store_error)?; let config_revision = bump_config_revision_in_transaction(&mut transaction) .await .map_err(store_error)?; @@ -369,7 +444,7 @@ impl ProxyStore for PgProxyRepository { .execute(&mut *transaction) .await .map_err(|error| { - // PostgreSQL 18 distinguishes RESTRICT from other foreign-key violations. + // PostgreSQL 18 为 RESTRICT 返回不同于普通外键违规的错误码。 if error.as_database_error().is_some_and(|error| { error.is_foreign_key_violation() || error.code().as_deref() == Some("23001") }) { @@ -410,6 +485,9 @@ impl ProxyStore for PgProxyRepository { .begin() .await .map_err(|_| store_error(unavailable()))?; + exclude_active_imports(&mut transaction, id) + .await + .map_err(store_error)?; 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()))?) diff --git a/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs b/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs index cdaa93b84..3208a9353 100644 --- a/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/provider_accounts/mod.rs @@ -2562,7 +2562,7 @@ fn plaintext_credential(marker: &str) -> PlaintextCredential { ) } -fn audit(id: &str, action: &str, entity_ref: &str) -> AdminAuditEvent { +pub(super) fn audit(id: &str, action: &str, entity_ref: &str) -> AdminAuditEvent { AdminAuditEvent { id: id.to_owned(), actor_kind: AdminAuditActorKind::System, diff --git a/backend/crates/gateway-store/tests/postgres/proxies.rs b/backend/crates/gateway-store/tests/postgres/proxies.rs index d6eed5c49..c52e43d60 100644 --- a/backend/crates/gateway-store/tests/postgres/proxies.rs +++ b/backend/crates/gateway-store/tests/postgres/proxies.rs @@ -43,6 +43,113 @@ fn update(account_id: &str, selection: AccountProxySelection) -> UpdateAccount { } } +#[tokio::test] +async fn import_reservation_blocks_proxy_mutations_until_rotated_credentials_are_committed() { + use gateway_store::postgres::{ + ImportProviderAccounts, ProviderAccountAdminRepository, ProviderAccountAdminScope, + }; + let Some(database) = TestDatabase::create("proxy_import_reservation").await else { + return; + }; + let store = PgProxyRepository::new(database.pool.clone()); + let other_process = PgProxyRepository::new(database.pool.clone()); + let context = context(); + let saved = store + .create( + NewProxy { + name: "导入出口".to_owned(), + proxy: OutboundProxy::parse("http://127.0.0.1:8080").unwrap(), + }, + &context, + ) + .await + .unwrap() + .record; + assert!(store.reserve_import(&saved.id).await.is_err()); + store + .record_test(&saved.id, saved.revision, success(), &context) + .await + .unwrap(); + let reservation = store.reserve_import(&saved.id).await.unwrap(); + let replacement = UpdateProxy { + id: saved.id.clone(), + revision: saved.revision, + name: saved.name, + proxy: Some(OutboundProxy::parse("http://127.0.0.1:9090").unwrap()), + }; + assert!( + other_process + .update(replacement.clone(), &context) + .await + .is_err() + ); + assert!( + other_process + .delete(&saved.id, saved.revision, &context) + .await + .is_err() + ); + assert!( + other_process + .record_test( + &saved.id, + saved.revision, + ProxyTestResult { + success: false, + ..success() + }, + &context + ) + .await + .is_err() + ); + + // 模拟上游已轮换的新凭据;持有保护时,另一个连接仍能完成导入事务。 + let mut candidate = account("acct_reserved_import", "reserved-import-user"); + candidate.outbound_proxy = Some(reservation.binding.proxy.clone()); + let repository = PgProviderAccountRepository::new(database.pool.clone()); + let audit = + super::provider_accounts::audit("audit_reserved_import", "import", "acct_reserved_import"); + tokio::time::timeout( + std::time::Duration::from_secs(5), + repository.import_provider_accounts(ImportProviderAccounts { + settings: None, + outbound_proxy: Some(reservation.binding.clone()), + scope: ProviderAccountAdminScope { + provider_kind: "openai".to_owned(), + }, + accounts: vec![candidate], + audit, + }), + ) + .await + .unwrap() + .unwrap(); + drop(reservation); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if other_process + .update(replacement.clone(), &context) + .await + .is_ok() + { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + assert!( + repository + .load_provider_account("acct_reserved_import") + .await + .unwrap() + .is_some() + ); + database.close().await; +} + #[tokio::test] async fn managed_proxies_persist_bind_update_all_accounts_and_protect_stale_tests() { let Some(database) = TestDatabase::create("managed_proxies").await else { diff --git a/backend/crates/providers/openai/src/lib.rs b/backend/crates/providers/openai/src/lib.rs index 9dd6b44d4..dbc98ec8d 100644 --- a/backend/crates/providers/openai/src/lib.rs +++ b/backend/crates/providers/openai/src/lib.rs @@ -36,7 +36,7 @@ pub use provider::{ pub mod credential; pub mod transport; -pub use transport::tls::ensure_rustls_provider; +pub use transport::tls::{build_reqwest_client_with_custom_ca, ensure_rustls_provider}; pub use transport::{ CodexCanonicalDecoder, CodexCanonicalError, CodexRequestEncodeError, OpenAiBillingUsage, encode_generate_request, openai_billing_breakdown, diff --git a/backend/crates/providers/openai/src/transport/client.rs b/backend/crates/providers/openai/src/transport/client.rs index 92ed9ec19..742528c2a 100644 --- a/backend/crates/providers/openai/src/transport/client.rs +++ b/backend/crates/providers/openai/src/transport/client.rs @@ -820,7 +820,7 @@ 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() diff --git a/docs/api.md b/docs/api.md index a1c228919..bd1f42a4b 100644 --- a/docs/api.md +++ b/docs/api.md @@ -236,14 +236,19 @@ Tests persist only when the requested revision still matches. Connectivity failu The test concurrency limit returns 429. 测试固定经代理访问 `https://api.ipify.org?format=json`,超时 15 秒,每进程最多同时测试 4 条。 +探测器复用 OpenAI 的证书信任配置:优先读取非空的 `CODEX_CA_CERTIFICATE`, +其次读取 `SSL_CERT_FILE`,并保留系统根证书;证书配置错误不会回退为不验证证书。 出口测试通过不表示 Provider 账号权限或额度可用;账号可用性使用账号连接测试。 导入请求可以携带顶层 `outboundProxyId`,在令牌交换前解析为默认出口;文件中显式的代理配置优先。 -OAuth 或导入期间若所选代理被删除、连接配置改变或测试失败,提交会拒绝过期配置。 +文件及 AT/RT 导入从凭据交换到落库期间保护所选代理;此时修改、删除或写入测试结果返回 409, +避免已轮换的凭据因代理状态变化而丢失。完成导入或请求取消后自动释放保护。 +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. +settings in the document take precedence. Credential imports reserve their selected proxy until commit; +concurrent proxy mutations return 409. OAuth commits still reject a deleted, changed or failed proxy. ### 账号连接测试 SSE diff --git a/docs/architecture.md b/docs/architecture.md index 4efeb1e27..b86cb208a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -191,6 +191,11 @@ Core 只理解 `Operation`、能力要求、Provider 候选、稳定错误和 ca 代理独立保存和测试,通过 `outboundProxyId` 绑定账号;账号保留解析后的 URL 供 Provider 使用。 连接配置改变时在同一事务内同步关联账号并清除旧测试结果,已绑定的代理不可删除。 +文件及 AT/RT 导入在上游凭据交换前取得 PostgreSQL 会话级共享咨询锁,直到凭据提交后释放。 +代理修改、删除和测试结果写入通过同一资源的事务级独占锁检查,导入期间直接返回冲突。 +锁连接不占用提交所需的连接池槽位,也不持有空闲事务;每进程最多增加四个导入保护连接。 +请求取消或进程退出时关闭会话并释放锁。探测器由组合根注入 OpenAI 的 HTTP 客户端构建函数, +复用其自定义 CA 加载与证书校验规则,Host 不依赖具体 Provider 包。 迁移 `0005_managed_outbound_proxies.sql` 将已有 URL 按完整连接信息合并为共享代理,保留认证及账号绑定。 导入器在认证交换前解析并校验账号出口。sub2api 的代理引用按完整 URL 登记为共享代理并绑定账号;