mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-06 08:06:19 +00:00
fix(proxies): share CA trust and preserve credentials during import
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
Codex
Generated
+1
@@ -744,6 +744,7 @@ dependencies = [
|
||||
"tempfile",
|
||||
"thiserror",
|
||||
"tokio",
|
||||
"tokio-rustls",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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(),
|
||||
)
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
//! Reusable account egress configurations and redacted connectivity observations.
|
||||
//! 可复用的账号出口配置，以及脱敏后的连通性测试结果。
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
use gateway_core::account::OutboundProxy;
|
||||
|
||||
@@ -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<ProxyImportReservation>;
|
||||
async fn list(&self, query: ProxyListQuery) -> AdminStoreResult<ProxyPage>;
|
||||
async fn get(&self, id: &str) -> AdminStoreResult<ProxyRecord>;
|
||||
async fn create(
|
||||
@@ -39,6 +41,14 @@ pub trait ProxyStore: Send + Sync {
|
||||
) -> AdminStoreResult<ProxyRecord>;
|
||||
}
|
||||
|
||||
/// 离开作用域时释放保护，错误返回和请求取消也遵循相同规则。
|
||||
pub trait ProxyImportGuard: Send + Sync {}
|
||||
|
||||
pub struct ProxyImportReservation {
|
||||
pub binding: ImportProxyBinding,
|
||||
pub guard: Box<dyn ProxyImportGuard>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
pub trait ProxyProbe: Send + Sync {
|
||||
async fn test(&self, proxy: &OutboundProxy) -> ProxyTestResult;
|
||||
|
||||
@@ -113,19 +113,13 @@ async fn required_credential(
|
||||
async fn import_proxy_binding(
|
||||
proxies: &dyn crate::ports::proxy::ProxyStore,
|
||||
id: Option<&str>,
|
||||
) -> Result<Option<crate::model::proxies::ImportProxyBinding>, AdminError> {
|
||||
) -> Result<Option<crate::ports::proxy::ProxyImportReservation>, 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(
|
||||
|
||||
@@ -131,11 +131,14 @@ impl OpenAiService for DefaultOpenAiService {
|
||||
command: ImportCredentials,
|
||||
) -> Result<CredentialImportResult, AdminError> {
|
||||
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;
|
||||
|
||||
@@ -84,11 +84,14 @@ impl XaiService for DefaultXaiService {
|
||||
command: ImportCredentials,
|
||||
) -> Result<CredentialImportResult, AdminError> {
|
||||
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 {
|
||||
|
||||
@@ -83,6 +83,7 @@ pub(super) struct AdminHarness {
|
||||
default_password: String,
|
||||
session_ttl_minutes: u64,
|
||||
accounts: Arc<dyn AccountStore>,
|
||||
proxies: Arc<dyn gateway_admin::ports::proxy::ProxyStore>,
|
||||
account_runtime: Arc<dyn AccountRuntimeStore>,
|
||||
account_groups: Arc<dyn AccountGroupStore>,
|
||||
auth: Arc<dyn AuthStore>,
|
||||
@@ -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<dyn gateway_admin::ports::proxy::ProxyStore>,
|
||||
) -> Self {
|
||||
self.proxies = proxies;
|
||||
self
|
||||
}
|
||||
|
||||
pub(super) fn system(mut self, system: Arc<dyn SystemOperations>) -> 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,
|
||||
)
|
||||
|
||||
@@ -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<super::accounts::EventLog>,
|
||||
}
|
||||
|
||||
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<gateway_admin::ports::proxy::ProxyImportReservation> {
|
||||
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<ProxyPage> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -28,6 +28,12 @@ fn missing() -> AdminStoreError {
|
||||
|
||||
#[async_trait]
|
||||
impl ProxyStore for MemoryProxies {
|
||||
async fn reserve_import(
|
||||
&self,
|
||||
_: &str,
|
||||
) -> AdminStoreResult<gateway_admin::ports::proxy::ProxyImportReservation> {
|
||||
Err(missing())
|
||||
}
|
||||
async fn list(&self, query: ProxyListQuery) -> AdminStoreResult<ProxyPage> {
|
||||
let items: Vec<_> = self.0.lock().unwrap().iter().cloned().collect();
|
||||
Ok(ProxyPage {
|
||||
|
||||
@@ -83,8 +83,14 @@ impl HostBundle {
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn proxy_probe(&self) -> Arc<dyn gateway_admin::ports::proxy::ProxyProbe> {
|
||||
Arc::new(proxy_probe::HttpProxyProbe::default())
|
||||
pub fn proxy_probe<E>(
|
||||
&self,
|
||||
build_client: impl Fn(reqwest::ClientBuilder) -> Result<reqwest::Client, E>
|
||||
+ Send
|
||||
+ Sync
|
||||
+ 'static,
|
||||
) -> Arc<dyn gateway_admin::ports::proxy::ProxyProbe> {
|
||||
Arc::new(proxy_probe::HttpProxyProbe::default().with_client_builder(build_client))
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
|
||||
@@ -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<ProxyClientBuilder>,
|
||||
}
|
||||
|
||||
type ProxyClientBuilder =
|
||||
dyn Fn(reqwest::ClientBuilder) -> Result<reqwest::Client, &'static str> + 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<String>) -> Self {
|
||||
Self {
|
||||
endpoint: endpoint.into(),
|
||||
build_client: Arc::new(|builder| builder.build().map_err(|_| "无法创建代理连接")),
|
||||
}
|
||||
}
|
||||
|
||||
/// 由组合根注入与 Provider 请求一致的证书信任策略。
|
||||
#[must_use]
|
||||
pub fn with_client_builder<E>(
|
||||
mut self,
|
||||
build: impl Fn(reqwest::ClientBuilder) -> Result<reqwest::Client, E> + Send + Sync + 'static,
|
||||
) -> Self {
|
||||
self.build_client = Arc::new(move |builder| {
|
||||
build(builder).map_err(|_| "无法创建代理连接，请检查证书信任配置")
|
||||
});
|
||||
self
|
||||
}
|
||||
|
||||
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()
|
||||
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() {
|
||||
"代理连接超时"
|
||||
|
||||
@@ -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"));
|
||||
}
|
||||
|
||||
@@ -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<Semaphore>,
|
||||
}
|
||||
|
||||
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<ProxyImportReservation> {
|
||||
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<ProxyPage> {
|
||||
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()))?)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
+7
-2
@@ -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
|
||||
|
||||
|
||||
@@ -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 登记为共享代理并绑定账号；
|
||||
|
||||
Reference in New Issue
Block a user