fix(proxy): improve robustness and prevent panics

- Add reqwest socks feature to support SOCKS proxy environments
- Fix UTF-8 safety in masked_key/masked_access_token (use chars() instead of byte slicing)
- Fix UTF-8 boundary check in usage_script HTTP response truncation
- Add defensive checks for JSON operations in proxy service
- Remove verbose debug logs that could trigger panic-prone code paths
This commit is contained in:
Jason
2026-01-09 12:42:18 +08:00
parent 01eee0ae22
commit ce9e5e25d3
5 changed files with 131 additions and 249 deletions
+1 -1
View File
@@ -37,7 +37,7 @@ tauri-plugin-deep-link = "2"
dirs = "5.0"
toml = "0.8"
toml_edit = "0.22"
reqwest = { version = "0.12", features = ["rustls-tls", "json", "stream"] }
reqwest = { version = "0.12", features = ["rustls-tls", "json", "stream", "socks"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time", "sync"] }
futures = "0.3"
async-stream = "0.3"
+19 -230
View File
@@ -15,7 +15,7 @@ use crate::{app_config::AppType, provider::Provider};
use reqwest::{Client, Response};
use serde_json::Value;
use std::sync::Arc;
use std::time::{Duration, Instant};
use std::time::Duration;
use tokio::sync::RwLock;
/// Headers 黑名单 - 不透传到上游的 Headers
@@ -127,29 +127,19 @@ impl RequestForwarder {
{
Ok(client) => (Some(client), None),
Err(e) => {
log::error!("Failed to create HTTP client (will fallback to no_proxy): {e}");
// 降级:忽略系统/环境代理,避免因代理配置问题导致整个应用崩溃
match Client::builder()
.timeout(Duration::from_secs(timeout_secs))
.no_proxy()
.build()
{
Ok(client) => {
log::warn!("HTTP client created with no_proxy() fallback");
(Some(client), Some(e.to_string()))
}
Err(fallback_err) => {
log::error!(
"Failed to create HTTP client even with no_proxy(): {fallback_err}"
);
(
None,
Some(format!(
"Failed to create HTTP client: {e}; no_proxy fallback failed: {fallback_err}"
)),
)
}
Ok(client) => (Some(client), Some(e.to_string())),
Err(fallback_err) => (
None,
Some(format!(
"Failed to create HTTP client: {e}; no_proxy fallback failed: {fallback_err}"
)),
),
}
}
};
@@ -193,12 +183,6 @@ impl RequestForwarder {
});
}
log::info!(
"[{}] 故障转移链: {} 个可用供应商",
app_type_str,
providers.len()
);
let mut last_error = None;
let mut last_provider = None;
let mut attempted_providers = 0usize;
@@ -221,25 +205,11 @@ impl RequestForwarder {
};
if !allowed {
log::debug!(
"[{}] Provider {} 熔断器拒绝本次请求,跳过",
app_type_str,
provider.name
);
continue;
}
attempted_providers += 1;
log::info!(
"[{}] 尝试 {}/{} - 使用Provider: {} (sort_index: {})",
app_type_str,
attempted_providers,
providers.len(),
provider.name,
provider.sort_index.unwrap_or(999999)
);
// 更新状态中的当前Provider信息
{
let mut status = self.status.write().await;
@@ -249,18 +219,14 @@ impl RequestForwarder {
status.last_request_at = Some(chrono::Utc::now().to_rfc3339());
}
let start = Instant::now();
// 转发请求(每个 Provider 只尝试一次,重试由客户端控制)
match self
.forward(provider, endpoint, &body, &headers, adapter.as_ref())
.await
{
Ok(response) => {
let latency = start.elapsed().as_millis() as u64;
// 成功:记录成功并更新熔断器
if let Err(e) = self
let _ = self
.router
.record_result(
&provider.id,
@@ -269,10 +235,7 @@ impl RequestForwarder {
true,
None,
)
.await
{
log::warn!("Failed to record success: {e}");
}
.await;
// 更新当前应用类型使用的 provider
{
@@ -292,12 +255,6 @@ impl RequestForwarder {
self.current_provider_id_at_start.as_str() != provider.id.as_str();
if should_switch {
status.failover_count += 1;
log::info!(
"[{}] 代理目标已切换到 Provider: {} (耗时: {}ms)",
app_type_str,
provider.name,
latency
);
// 异步触发供应商切换,更新 UI/托盘,并把“当前供应商”同步为实际使用的 provider
let fm = self.failover_manager.clone();
@@ -307,10 +264,7 @@ impl RequestForwarder {
let at = app_type_str.to_string();
tokio::spawn(async move {
if let Err(e) = fm.try_switch(ah.as_ref(), &at, &pid, &pname).await
{
log::error!("[Failover] 切换供应商失败: {e}");
}
let _ = fm.try_switch(ah.as_ref(), &at, &pid, &pname).await;
});
}
// 重新计算成功率
@@ -321,23 +275,14 @@ impl RequestForwarder {
}
}
log::info!(
"[{}] 请求成功 - Provider: {} - {}ms",
app_type_str,
provider.name,
latency
);
return Ok(ForwardResult {
response,
provider: provider.clone(),
});
}
Err(e) => {
let latency = start.elapsed().as_millis() as u64;
// 失败:记录失败并更新熔断器
if let Err(record_err) = self
let _ = self
.router
.record_result(
&provider.id,
@@ -346,10 +291,7 @@ impl RequestForwarder {
false,
Some(e.to_string()),
)
.await
{
log::warn!("Failed to record failure: {record_err}");
}
.await;
// 分类错误
let category = self.categorize_proxy_error(&e);
@@ -363,14 +305,6 @@ impl RequestForwarder {
Some(format!("Provider {} 失败: {}", provider.name, e));
}
log::warn!(
"[{}] Provider {} 失败(可重试): {} - {}ms",
app_type_str,
provider.name,
e,
latency
);
last_error = Some(e);
last_provider = Some(provider.clone());
// 继续尝试下一个供应商
@@ -388,12 +322,6 @@ impl RequestForwarder {
* 100.0;
}
}
log::error!(
"[{}] Provider {} 失败(不可重试): {}",
app_type_str,
provider.name,
e
);
return Err(ForwardError {
error: e,
provider: Some(provider.clone()),
@@ -432,12 +360,6 @@ impl RequestForwarder {
}
}
log::error!(
"[{}] 所有 {} 个供应商都失败了",
app_type_str,
providers.len()
);
Err(ForwardError {
error: last_error.unwrap_or(ProxyError::MaxRetriesExceeded),
provider: last_provider,
@@ -455,7 +377,6 @@ impl RequestForwarder {
) -> Result<Response, ProxyError> {
// 使用适配器提取 base_url
let base_url = adapter.extract_base_url(provider)?;
log::info!("[{}] base_url: {}", adapter.name(), base_url);
// 检查是否需要格式转换
let needs_transform = adapter.needs_transform(provider);
@@ -470,36 +391,13 @@ impl RequestForwarder {
// 使用适配器构建 URL
let url = adapter.build_url(&base_url, effective_endpoint);
// 记录原始请求 JSON
log::info!(
"[{}] ====== 请求开始 ======\n>>> 原始请求 JSON:\n{}",
adapter.name(),
serde_json::to_string_pretty(body).unwrap_or_else(|_| body.to_string())
);
// 应用模型映射(独立于格式转换)
let (mapped_body, _original_model, mapped_model) =
let (mapped_body, _original_model, _mapped_model) =
super::model_mapper::apply_model_mapping(body.clone(), provider);
if let Some(ref mapped) = mapped_model {
log::info!(
"[{}] >>> 模型映射后的请求 JSON:\n{}",
adapter.name(),
serde_json::to_string_pretty(&mapped_body).unwrap_or_default()
);
log::info!("[{}] 模型已映射到: {}", adapter.name(), mapped);
}
// 转换请求体(如果需要)
let request_body = if needs_transform {
log::info!("[{}] 转换请求格式 (Anthropic → OpenAI)", adapter.name());
let transformed = adapter.transform_request(mapped_body, provider)?;
log::info!(
"[{}] >>> 转换后的请求 JSON:\n{}",
adapter.name(),
serde_json::to_string_pretty(&transformed).unwrap_or_default()
);
transformed
adapter.transform_request(mapped_body, provider)?
} else {
mapped_body
};
@@ -508,38 +406,6 @@ impl RequestForwarder {
// 默认使用空白名单,过滤所有 _ 前缀字段
let filtered_body = filter_private_params_with_whitelist(request_body, &[]);
// ========== 请求体日志(截断显示) ==========
let body_str = serde_json::to_string_pretty(&filtered_body)
.unwrap_or_else(|_| filtered_body.to_string());
const BODY_PREVIEW_MAX_BYTES: usize = 2000;
let body_preview = if body_str.len() > BODY_PREVIEW_MAX_BYTES {
// Rust 的 String 是 UTF-8,不能用固定字节下标直接切片,否则可能切到多字节字符中间导致 panic。
// 这里向左寻找最近的字符边界,确保截断安全。
let mut safe_cut = BODY_PREVIEW_MAX_BYTES;
while !body_str.is_char_boundary(safe_cut) {
safe_cut = safe_cut.saturating_sub(1);
}
format!(
"{}...\n[截断,总长度: {} bytes]",
&body_str[..safe_cut],
body_str.len()
)
} else {
body_str
};
log::info!(
"[{}] ====== 最终请求体 ======\n{}",
adapter.name(),
body_preview
);
log::info!(
"[{}] 转发请求: {} -> {}",
adapter.name(),
provider.name,
url
);
// 构建请求
let client = self.client.as_ref().ok_or_else(|| {
ProxyError::ForwardFailed(
@@ -550,43 +416,17 @@ impl RequestForwarder {
})?;
let mut request = client.post(&url);
// ========== 详细 Headers 日志 ==========
log::info!("[{}] ====== 客户端原始 Headers ======", adapter.name());
for (key, value) in headers {
log::info!(
"[{}] {}: {:?}",
adapter.name(),
key.as_str(),
value.to_str().unwrap_or("<binary>")
);
}
// 过滤黑名单 Headers,保护隐私并避免冲突
let mut filtered_headers: Vec<String> = Vec::new();
let mut passed_headers: Vec<(String, String)> = Vec::new();
for (key, value) in headers {
let key_str = key.as_str().to_lowercase();
if HEADER_BLACKLIST.contains(&key_str.as_str()) {
filtered_headers.push(key_str);
if HEADER_BLACKLIST
.iter()
.any(|h| key.as_str().eq_ignore_ascii_case(h))
{
continue;
}
let value_str = value.to_str().unwrap_or("<binary>").to_string();
passed_headers.push((key.as_str().to_string(), value_str.clone()));
request = request.header(key, value);
}
if !filtered_headers.is_empty() {
log::info!(
"[{}] ====== 被过滤的 Headers ({}) ======",
adapter.name(),
filtered_headers.len()
);
for h in &filtered_headers {
log::info!("[{}] - {}", adapter.name(), h);
}
}
// 处理 anthropic-beta Header(仅 Claude
// 关键:确保包含 claude-code-20250219 标记,这是上游服务验证请求来源的依据
// 如果客户端发送的 beta 标记中没有包含 claude-code-20250219,需要补充
@@ -609,53 +449,27 @@ impl RequestForwarder {
CLAUDE_CODE_BETA.to_string()
};
request = request.header("anthropic-beta", &beta_value);
passed_headers.push(("anthropic-beta".to_string(), beta_value.clone()));
log::info!("[{}] 设置 anthropic-beta: {}", adapter.name(), beta_value);
}
// 客户端 IP 透传(默认开启)
if let Some(xff) = headers.get("x-forwarded-for") {
if let Ok(xff_str) = xff.to_str() {
request = request.header("x-forwarded-for", xff_str);
passed_headers.push(("x-forwarded-for".to_string(), xff_str.to_string()));
log::debug!("[{}] 透传 x-forwarded-for: {}", adapter.name(), xff_str);
}
}
if let Some(real_ip) = headers.get("x-real-ip") {
if let Ok(real_ip_str) = real_ip.to_str() {
request = request.header("x-real-ip", real_ip_str);
passed_headers.push(("x-real-ip".to_string(), real_ip_str.to_string()));
log::debug!("[{}] 透传 x-real-ip: {}", adapter.name(), real_ip_str);
}
}
// 禁用压缩,避免 gzip 流式响应解析错误
// 参考 CCH: undici 在连接提前关闭时会对不完整的 gzip 流抛出错误
request = request.header("accept-encoding", "identity");
passed_headers.push(("accept-encoding".to_string(), "identity".to_string()));
// 使用适配器添加认证头
if let Some(auth) = adapter.extract_auth(provider) {
log::debug!(
"[{}] 使用认证: {:?} (key: {})",
adapter.name(),
auth.strategy,
auth.masked_key()
);
request = adapter.add_auth_headers(request, &auth);
// 记录认证头(脱敏)
let key_preview: String = auth.api_key.chars().take(8).collect();
passed_headers.push((
"authorization".to_string(),
format!("Bearer {}...", &key_preview),
));
passed_headers.push(("x-api-key".to_string(), format!("{}...", &key_preview)));
} else {
log::error!(
"[{}] 未找到 API KeyProvider: {}",
adapter.name(),
provider.name
);
}
// anthropic-version 统一处理(仅 Claude):优先使用客户端的版本号,否则使用默认值
@@ -666,28 +480,10 @@ impl RequestForwarder {
.and_then(|v| v.to_str().ok())
.unwrap_or("2023-06-01");
request = request.header("anthropic-version", version_str);
passed_headers.push(("anthropic-version".to_string(), version_str.to_string()));
log::info!(
"[{}] 设置 anthropic-version: {}",
adapter.name(),
version_str
);
}
// ========== 最终发送的 Headers 日志 ==========
log::info!(
"[{}] ====== 最终发送的 Headers ({}) ======",
adapter.name(),
passed_headers.len()
);
for (k, v) in &passed_headers {
log::info!("[{}] {}: {}", adapter.name(), k, v);
}
// 发送请求
log::info!("[{}] 发送请求到: {}", adapter.name(), url);
let response = request.json(&filtered_body).send().await.map_err(|e| {
log::error!("[{}] 请求失败: {}", adapter.name(), e);
if e.is_timeout() {
ProxyError::Timeout(format!("请求超时: {e}"))
} else if e.is_connect() {
@@ -699,19 +495,12 @@ impl RequestForwarder {
// 检查响应状态
let status = response.status();
log::info!("[{}] 响应状态: {}", adapter.name(), status);
if status.is_success() {
Ok(response)
} else {
let status_code = status.as_u16();
let body_text = response.text().await.ok();
log::error!(
"[{}] 上游错误 ({}): {:?}",
adapter.name(),
status_code,
body_text
);
Err(ProxyError::UpstreamError {
status: status_code,
+39 -8
View File
@@ -38,13 +38,20 @@ impl AuthInfo {
///
/// 显示前4位和后4位,中间用 `...` 代替
/// 如果 key 长度不足8位,则返回 `***`
#[allow(dead_code)]
pub fn masked_key(&self) -> String {
if self.api_key.len() > 8 {
format!(
"{}...{}",
&self.api_key[..4],
&self.api_key[self.api_key.len() - 4..]
)
if self.api_key.chars().count() > 8 {
let prefix: String = self.api_key.chars().take(4).collect();
let suffix: String = self
.api_key
.chars()
.rev()
.take(4)
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect();
format!("{prefix}...{suffix}")
} else {
"***".to_string()
}
@@ -54,8 +61,17 @@ impl AuthInfo {
#[allow(dead_code)]
pub fn masked_access_token(&self) -> Option<String> {
self.access_token.as_ref().map(|token| {
if token.len() > 8 {
format!("{}...{}", &token[..4], &token[token.len() - 4..])
if token.chars().count() > 8 {
let prefix: String = token.chars().take(4).collect();
let suffix: String = token
.chars()
.rev()
.take(4)
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect();
format!("{prefix}...{suffix}")
} else {
"***".to_string()
}
@@ -126,6 +142,13 @@ mod tests {
assert_eq!(auth.masked_key(), "1234...6789");
}
#[test]
fn test_masked_key_utf8_safe() {
let auth = AuthInfo::new("测试⚠️1234567890".to_string(), AuthStrategy::Bearer);
let masked = auth.masked_key();
assert!(!masked.is_empty());
}
#[test]
fn test_auth_strategy_equality() {
assert_eq!(AuthStrategy::Anthropic, AuthStrategy::Anthropic);
@@ -160,6 +183,14 @@ mod tests {
assert_eq!(auth.masked_access_token(), Some("ya29...cdef".to_string()));
}
#[test]
fn test_masked_access_token_utf8_safe() {
let auth =
AuthInfo::with_access_token("refresh".to_string(), "令牌⚠️1234567890".to_string());
let masked = auth.masked_access_token().unwrap();
assert!(!masked.is_empty());
}
#[test]
fn test_masked_access_token_short() {
let auth = AuthInfo::with_access_token("refresh".to_string(), "short".to_string());
+67 -9
View File
@@ -441,8 +441,21 @@ impl ProxyService {
}
None => {
// 至少写入一份可用的 Token
provider.settings_config["env"] =
json!({ token_key: token });
if provider.settings_config.is_null() {
provider.settings_config = json!({});
}
if let Some(root) = provider.settings_config.as_object_mut()
{
root.insert(
"env".to_string(),
json!({ token_key: token }),
);
} else {
log::warn!(
"Claude provider settings_config 格式异常(非对象),跳过写入 Token (provider: {provider_id})"
);
}
}
}
@@ -485,9 +498,20 @@ impl ProxyService {
{
auth_obj.insert("OPENAI_API_KEY".to_string(), json!(token));
} else {
provider.settings_config["auth"] = json!({
"OPENAI_API_KEY": token
});
if provider.settings_config.is_null() {
provider.settings_config = json!({});
}
if let Some(root) = provider.settings_config.as_object_mut() {
root.insert(
"auth".to_string(),
json!({ "OPENAI_API_KEY": token }),
);
} else {
log::warn!(
"Codex provider settings_config 格式异常(非对象),跳过写入 Token (provider: {provider_id})"
);
}
}
if let Err(e) = self.db.update_provider_settings_config(
@@ -526,9 +550,20 @@ impl ProxyService {
{
env_obj.insert("GEMINI_API_KEY".to_string(), json!(token));
} else {
provider.settings_config["env"] = json!({
"GEMINI_API_KEY": token
});
if provider.settings_config.is_null() {
provider.settings_config = json!({});
}
if let Some(root) = provider.settings_config.as_object_mut() {
root.insert(
"env".to_string(),
json!({ "GEMINI_API_KEY": token }),
);
} else {
log::warn!(
"Gemini provider settings_config 格式异常(非对象),跳过写入 Token (provider: {provider_id})"
);
}
}
if let Err(e) = self.db.update_provider_settings_config(
@@ -1526,7 +1561,30 @@ impl ProxyService {
if !path.exists() {
return Err("Claude 配置文件不存在".to_string());
}
read_json_file(&path).map_err(|e| format!("读取 Claude 配置失败: {e}"))
let mut value: Value =
read_json_file(&path).map_err(|e| format!("读取 Claude 配置失败: {e}"))?;
if value.is_null() {
value = json!({});
}
if !value.is_object() {
let kind = match &value {
Value::Null => "null",
Value::Bool(_) => "boolean",
Value::Number(_) => "number",
Value::String(_) => "string",
Value::Array(_) => "array",
Value::Object(_) => "object",
};
return Err(format!(
"Claude 配置文件格式错误:根节点必须是 JSON 对象(当前为 {kind}),路径: {}",
path.display()
));
}
Ok(value)
}
fn write_claude_live(&self, config: &Value) -> Result<(), String> {
+5 -1
View File
@@ -269,7 +269,11 @@ async fn send_http_request(config: &RequestConfig, timeout_secs: u64) -> Result<
if !status.is_success() {
let preview = if text.len() > 200 {
format!("{}...", &text[..200])
let mut safe_cut = 200usize;
while !text.is_char_boundary(safe_cut) {
safe_cut = safe_cut.saturating_sub(1);
}
format!("{}...", &text[..safe_cut])
} else {
text.clone()
};