feat: delegate 协议改为 header 元数据 + gzip 压缩 body 直传

- delegate wire format 从 JSON body 改为 HTTP headers 传递元数据
  (X-Delegate-Method/Url/Headers),上游请求体 gzip 压缩后直传,
  减少跨国代理传输耗时
- Rust 侧新增 gzip 解压(含 decompression bomb 防护 50MB 上限)
- 升级模块从 GitHub API asset 下载改为公开 release URL 直链,
  GITHUB_TOKEN 变为可选
- 前端适配新 timing 字段,展示压缩率与 wire_size/body_size
- CI release notes 改用 generate_release_notes 自动生成
This commit is contained in:
fawney19
2026-02-11 16:34:18 +08:00
parent 2436ce45a2
commit 9647d95759
6 changed files with 215 additions and 157 deletions

View File

@@ -108,42 +108,7 @@ jobs:
uses: softprops/action-gh-release@v2 uses: softprops/action-gh-release@v2
with: with:
name: "aether-proxy ${{ github.ref_name }}" name: "aether-proxy ${{ github.ref_name }}"
body: | generate_release_notes: true
## aether-proxy ${{ github.ref_name }}
### 下载
| 平台 | 文件 |
|------|------|
| Linux x86_64 | `aether-proxy-linux-amd64.tar.gz` |
| Linux ARM64 | `aether-proxy-linux-arm64.tar.gz` |
| macOS x86_64 (Intel) | `aether-proxy-macos-amd64.tar.gz` |
| macOS ARM64 (Apple Silicon) | `aether-proxy-macos-arm64.tar.gz` |
| Windows x86_64 | `aether-proxy-windows-amd64.zip` |
### 使用
```bash
# Linux/macOS: 解压
tar xzf aether-proxy-<platform>.tar.gz
# Windows: 解压 zip 文件
# 配置环境变量 (或写入 .env 文件)
export AETHER_PROXY_AETHER_URL=https://your-aether.example.com
export AETHER_PROXY_MANAGEMENT_TOKEN=ae_xxx
export AETHER_PROXY_HMAC_KEY=your-hmac-key
# 运行
./aether-proxy
```
### 校验
下载 `SHA256SUMS.txt` 后可验证文件完整性:
```bash
sha256sum -c SHA256SUMS.txt
```
files: | files: |
artifacts/aether-proxy-* artifacts/aether-proxy-*
artifacts/SHA256SUMS.txt artifacts/SHA256SUMS.txt

View File

@@ -10,7 +10,7 @@ checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
[[package]] [[package]]
name = "aether-proxy" name = "aether-proxy"
version = "0.1.0" version = "0.1.2"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"base64", "base64",

View File

@@ -1,5 +1,6 @@
use std::collections::HashMap; use std::collections::HashMap;
use std::collections::HashSet; use std::collections::HashSet;
use std::io::Read;
use std::sync::Arc; use std::sync::Arc;
use std::time::Instant; use std::time::Instant;
@@ -7,7 +8,6 @@ use futures_util::TryStreamExt;
use http_body_util::{BodyExt, Full, Limited, StreamBody}; use http_body_util::{BodyExt, Full, Limited, StreamBody};
use hyper::body::{Frame, Incoming}; use hyper::body::{Frame, Incoming};
use hyper::{Request, Response}; use hyper::{Request, Response};
use serde::Deserialize;
use tracing::{debug, warn}; use tracing::{debug, warn};
use url::Url; use url::Url;
@@ -16,22 +16,20 @@ use crate::auth;
use crate::config::Config; use crate::config::Config;
use crate::proxy::target_filter::{self, DnsCache}; use crate::proxy::target_filter::{self, DnsCache};
/// Delegation request payload sent by Aether.
#[derive(Debug, Deserialize)]
struct DelegateRequest {
method: String,
url: String,
headers: HashMap<String, String>,
body: Option<String>,
/// Accepted but not used on the proxy side — Aether controls timeouts.
#[allow(dead_code)]
timeout: Option<u64>,
}
/// Handle delegation requests: Aether sends a full request description, /// Handle delegation requests: Aether sends a full request description,
/// and the proxy issues the actual upstream HTTP call using its own TLS stack. /// and the proxy issues the actual upstream HTTP call using its own TLS stack.
/// ///
/// Endpoint: POST /_aether/delegate /// Endpoint: POST /_aether/delegate
///
/// Wire format: metadata in HTTP headers, upstream body sent directly
/// as HTTP body (optionally gzip-compressed via `Content-Encoding: gzip`).
///
/// Headers:
/// X-Delegate-Method: POST
/// X-Delegate-Url: https://api.anthropic.com/v1/messages
/// X-Delegate-Headers: base64-encoded JSON {"Authorization": "Bearer ...", ...}
/// X-Delegate-Timeout: 30 (accepted but not used — Aether controls timeouts)
/// Content-Encoding: gzip (optional, indicates body is gzip-compressed)
pub async fn handle_delegate( pub async fn handle_delegate(
req: Request<Incoming>, req: Request<Incoming>,
config: Arc<Config>, config: Arc<Config>,
@@ -42,7 +40,7 @@ pub async fn handle_delegate(
) -> Response<BoxBody> { ) -> Response<BoxBody> {
let total_start = Instant::now(); let total_start = Instant::now();
// Authenticate via Authorization header (same HMAC scheme as Proxy-Authorization) // ── Auth ──
let auth_header = req let auth_header = req
.headers() .headers()
.get("authorization") .get("authorization")
@@ -54,34 +52,74 @@ pub async fn handle_delegate(
} }
let auth_ms = total_start.elapsed().as_millis() as u64; let auth_ms = total_start.elapsed().as_millis() as u64;
// Read and parse request body (limit to 10 MB to prevent OOM) // ── Parse metadata from headers ──
const MAX_BODY: usize = 10 * 1024 * 1024; let meta_start = Instant::now();
let body_read_start = Instant::now();
let body_bytes = match Limited::new(req.into_body(), MAX_BODY).collect().await { let method_str = match req
Ok(collected) => collected.to_bytes(), .headers()
Err(e) => { .get("x-delegate-method")
warn!(error = %e, "failed to read delegate request body"); .and_then(|v| v.to_str().ok())
return error_response(413, "payload_too_large", "request body exceeds 10MB limit"); {
Some(m) => m.to_string(),
None => {
warn!("delegate missing X-Delegate-Method");
return error_response(400, "bad_request", "missing X-Delegate-Method header");
} }
}; };
let body_read_ms = body_read_start.elapsed().as_millis() as u64;
let body_size = body_bytes.len() as u64;
let body_parse_start = Instant::now(); let target_url = match req
let delegate_req: DelegateRequest = match serde_json::from_slice(&body_bytes) { .headers()
Ok(r) => r, .get("x-delegate-url")
Err(e) => { .and_then(|v| v.to_str().ok())
warn!(error = %e, "invalid delegate request JSON"); {
return error_response(400, "bad_request", &format!("invalid JSON: {}", e)); Some(u) => u.to_string(),
None => {
warn!("delegate missing X-Delegate-Url");
return error_response(400, "bad_request", "missing X-Delegate-Url header");
} }
}; };
let body_parse_ms = body_parse_start.elapsed().as_millis() as u64;
// Target filter: validate the upstream URL against allowed ports and private IP checks let upstream_headers: HashMap<String, String> = match req
let parsed_url = match Url::parse(&delegate_req.url) { .headers()
.get("x-delegate-headers")
.and_then(|v| v.to_str().ok())
{
Some(b64) => {
match base64::Engine::decode(&base64::engine::general_purpose::STANDARD, b64) {
Ok(decoded) => match serde_json::from_slice(&decoded) {
Ok(h) => h,
Err(e) => {
warn!(error = %e, "delegate invalid X-Delegate-Headers JSON");
return error_response(
400,
"bad_request",
"invalid X-Delegate-Headers JSON",
);
}
},
Err(e) => {
warn!(error = %e, "delegate invalid X-Delegate-Headers base64");
return error_response(400, "bad_request", "invalid X-Delegate-Headers base64");
}
}
}
None => HashMap::new(),
};
let is_gzip = req
.headers()
.get("content-encoding")
.and_then(|v| v.to_str().ok())
.map(|v| v.eq_ignore_ascii_case("gzip"))
.unwrap_or(false);
let meta_ms = meta_start.elapsed().as_millis() as u64;
// ── Target validation ──
let parsed_url = match Url::parse(&target_url) {
Ok(u) => u, Ok(u) => u,
Err(e) => { Err(e) => {
warn!(url = %delegate_req.url, error = %e, "invalid delegate target URL"); warn!(url = %target_url, error = %e, "delegate invalid target URL");
return error_response(400, "bad_request", &format!("invalid URL: {}", e)); return error_response(400, "bad_request", &format!("invalid URL: {}", e));
} }
}; };
@@ -89,7 +127,7 @@ pub async fn handle_delegate(
let host = match parsed_url.host_str() { let host = match parsed_url.host_str() {
Some(h) => h.to_string(), Some(h) => h.to_string(),
None => { None => {
warn!(url = %delegate_req.url, "delegate target URL missing host"); warn!(url = %target_url, "delegate target URL missing host");
return error_response(400, "bad_request", "URL missing host"); return error_response(400, "bad_request", "URL missing host");
} }
}; };
@@ -103,51 +141,91 @@ pub async fn handle_delegate(
} }
let dns_ms = dns_start.elapsed().as_millis() as u64; let dns_ms = dns_start.elapsed().as_millis() as u64;
debug!( debug!(method = %method_str, url = %target_url, is_gzip, "delegate request");
method = %delegate_req.method,
url = %delegate_req.url,
"delegate request"
);
// Build upstream request // ── Read body + decompress if gzip ──
let method = match delegate_req.method.parse::<reqwest::Method>() { let body_read_start = Instant::now();
const MAX_BODY: usize = 10 * 1024 * 1024;
let body_bytes = match Limited::new(req.into_body(), MAX_BODY).collect().await {
Ok(collected) => collected.to_bytes(),
Err(e) => {
warn!(error = %e, "delegate failed to read request body");
return error_response(413, "payload_too_large", "request body exceeds 10MB limit");
}
};
let body_read_ms = body_read_start.elapsed().as_millis() as u64;
let wire_size = body_bytes.len() as u64;
let decompress_start = Instant::now();
const MAX_DECOMPRESSED: usize = 50 * 1024 * 1024; // 50 MB
let upstream_body: Option<Vec<u8>> = if body_bytes.is_empty() {
None
} else if is_gzip {
let decoder = flate2::read::GzDecoder::new(&body_bytes[..]);
let mut decompressed = Vec::with_capacity((body_bytes.len() * 4).min(MAX_DECOMPRESSED));
match decoder
.take(MAX_DECOMPRESSED as u64 + 1)
.read_to_end(&mut decompressed)
{
Ok(n) if n > MAX_DECOMPRESSED => {
warn!(
wire_size = body_bytes.len(),
decompressed_size = n,
"delegate decompressed body exceeds limit"
);
return error_response(
413,
"payload_too_large",
"decompressed body exceeds 50MB limit",
);
}
Ok(_) => Some(decompressed),
Err(e) => {
warn!(error = %e, "delegate gzip decompression failed");
return error_response(400, "bad_request", "gzip decompression failed");
}
}
} else {
Some(body_bytes.to_vec())
};
let decompress_ms = decompress_start.elapsed().as_millis() as u64;
let body_size = upstream_body.as_ref().map(|b| b.len() as u64).unwrap_or(0);
// ── Build upstream request ──
let method = match method_str.parse::<reqwest::Method>() {
Ok(m) => m, Ok(m) => m,
Err(e) => { Err(e) => {
warn!(error = %e, method = %delegate_req.method, "invalid HTTP method"); warn!(error = %e, method = %method_str, "delegate invalid HTTP method");
return error_response(400, "bad_request", &format!("invalid method: {}", e)); return error_response(400, "bad_request", &format!("invalid method: {}", e));
} }
}; };
let mut upstream_req = http_client.request(method, &delegate_req.url); let mut upstream_req = http_client.request(method, &target_url);
// NOTE: We intentionally do NOT set a per-request timeout here.
// reqwest's `.timeout()` caps the *entire* request including body streaming,
// which would truncate long-lived SSE streams. The delegate_client already
// has a configured connect_timeout for connection establishment, and Aether controls
// first-byte / idle timeouts on its own side via asyncio.
// Set headers (skip `host` — reqwest sets it from the URL automatically, // Set headers (skip `host` — reqwest sets it from the URL automatically,
// and a duplicate Host header can confuse certain upstreams) // and a duplicate Host header can confuse certain upstreams)
for (name, value) in &delegate_req.headers { for (name, value) in &upstream_headers {
if name.eq_ignore_ascii_case("host") { if name.eq_ignore_ascii_case("host") {
continue; continue;
} }
upstream_req = upstream_req.header(name.as_str(), value.as_str()); upstream_req = upstream_req.header(name.as_str(), value.as_str());
} }
// Set body if let Some(body) = upstream_body {
if let Some(body) = delegate_req.body {
upstream_req = upstream_req.body(body); upstream_req = upstream_req.body(body);
} }
// Send upstream request // ── Send upstream request ──
// NOTE: We intentionally do NOT set a per-request timeout here.
// reqwest's `.timeout()` caps the *entire* request including body streaming,
// which would truncate long-lived SSE streams. The delegate_client already
// has a configured connect_timeout for connection establishment, and Aether controls
// first-byte / idle timeouts on its own side via asyncio.
let upstream_start = Instant::now(); let upstream_start = Instant::now();
let upstream_resp = match upstream_req.send().await { let upstream_resp = match upstream_req.send().await {
Ok(resp) => resp, Ok(resp) => resp,
Err(e) => { Err(e) => {
warn!(url = %delegate_req.url, error = %e, "delegate upstream request failed"); warn!(url = %target_url, error = %e, "delegate upstream request failed");
// Sanitize: strip URL details from error message to avoid leaking
// API keys or paths that may appear in query strings / paths.
let safe_detail = sanitize_upstream_error(&e.to_string()); let safe_detail = sanitize_upstream_error(&e.to_string());
if e.is_timeout() { if e.is_timeout() {
return error_response(504, "upstream_timeout", &safe_detail); return error_response(504, "upstream_timeout", &safe_detail);
@@ -157,33 +235,34 @@ pub async fn handle_delegate(
}; };
let upstream_ms = upstream_start.elapsed().as_millis() as u64; let upstream_ms = upstream_start.elapsed().as_millis() as u64;
// Build response: pass through upstream status + headers, stream body back // ── Build response ──
let status = upstream_resp.status().as_u16(); let status = upstream_resp.status().as_u16();
let upstream_headers = upstream_resp.headers().clone(); let resp_headers = upstream_resp.headers().clone();
let total_ms = total_start.elapsed().as_millis() as u64; let total_ms = total_start.elapsed().as_millis() as u64;
debug!( debug!(
url = %delegate_req.url, url = %target_url,
status, status,
dns_ms, dns_ms,
upstream_ms, upstream_ms,
total_ms, total_ms,
wire_size,
body_size,
"delegate upstream response" "delegate upstream response"
); );
// Inject proxy timing header for Aether to parse
let timing = serde_json::json!({ let timing = serde_json::json!({
"auth_ms": auth_ms, "auth_ms": auth_ms,
"meta_ms": meta_ms,
"body_read_ms": body_read_ms, "body_read_ms": body_read_ms,
"body_parse_ms": body_parse_ms, "decompress_ms": decompress_ms,
"wire_size": wire_size,
"body_size": body_size, "body_size": body_size,
"dns_ms": dns_ms, "dns_ms": dns_ms,
"upstream_ms": upstream_ms, "upstream_ms": upstream_ms,
"total_ms": total_ms, "total_ms": total_ms,
}); });
// Stream the response body
let body_stream = upstream_resp let body_stream = upstream_resp
.bytes_stream() .bytes_stream()
.map_ok(Frame::data) .map_ok(Frame::data)
@@ -192,7 +271,7 @@ pub async fn handle_delegate(
let stream_body: BoxBody = StreamBody::new(body_stream).boxed(); let stream_body: BoxBody = StreamBody::new(body_stream).boxed();
let mut builder = Response::builder().status(status); let mut builder = Response::builder().status(status);
for (name, value) in upstream_headers.iter() { for (name, value) in resp_headers.iter() {
builder = builder.header(name, value); builder = builder.header(name, value);
} }
builder = builder.header("X-Proxy-Timing", timing.to_string()); builder = builder.header("X-Proxy-Timing", timing.to_string());

View File

@@ -17,13 +17,6 @@ const CURRENT_VERSION: &str = env!("CARGO_PKG_VERSION");
struct GithubRelease { struct GithubRelease {
tag_name: String, tag_name: String,
name: String, name: String,
assets: Vec<GithubAsset>,
}
#[derive(serde::Deserialize)]
struct GithubAsset {
name: String,
id: u64,
} }
// ── Platform detection ─────────────────────────────────────────────────────── // ── Platform detection ───────────────────────────────────────────────────────
@@ -49,17 +42,15 @@ fn detect_platform() -> &'static str {
// ── GitHub HTTP client ─────────────────────────────────────────────────────── // ── GitHub HTTP client ───────────────────────────────────────────────────────
fn build_github_client() -> anyhow::Result<reqwest::Client> { fn build_github_client() -> anyhow::Result<reqwest::Client> {
let token = std::env::var("GITHUB_TOKEN").map_err(|_| {
anyhow::anyhow!(
"GITHUB_TOKEN is required (private repo).\n Set it via: export GITHUB_TOKEN=ghp_xxx"
)
})?;
let mut headers = reqwest::header::HeaderMap::new(); let mut headers = reqwest::header::HeaderMap::new();
headers.insert(
reqwest::header::AUTHORIZATION, if let Ok(token) = std::env::var("GITHUB_TOKEN") {
reqwest::header::HeaderValue::from_str(&format!("Bearer {}", token))?, headers.insert(
); reqwest::header::AUTHORIZATION,
reqwest::header::HeaderValue::from_str(&format!("Bearer {}", token))?,
);
}
headers.insert( headers.insert(
reqwest::header::ACCEPT, reqwest::header::ACCEPT,
reqwest::header::HeaderValue::from_static("application/vnd.github+json"), reqwest::header::HeaderValue::from_static("application/vnd.github+json"),
@@ -119,16 +110,18 @@ async fn fetch_release(
} }
} }
// ── Download & verify ──────────────────────────────────────────────────────── // ── Download via GitHub release direct links ─────────────────────────────────
async fn download_asset_bytes( /// Download a release asset via the public direct download URL:
/// `https://github.com/{repo}/releases/download/{tag}/{filename}`
async fn download_release_file(
client: &reqwest::Client, client: &reqwest::Client,
asset: &GithubAsset, tag: &str,
filename: &str,
) -> anyhow::Result<Vec<u8>> { ) -> anyhow::Result<Vec<u8>> {
// Use GitHub API asset endpoint for reliable private repo downloads
let url = format!( let url = format!(
"{}/repos/{}/releases/assets/{}", "https://github.com/{}/releases/download/{}/{}",
GITHUB_API_BASE, GITHUB_REPO, asset.id GITHUB_REPO, tag, filename
); );
let resp = client let resp = client
.get(&url) .get(&url)
@@ -138,7 +131,7 @@ async fn download_asset_bytes(
if !resp.status().is_success() { if !resp.status().is_success() {
anyhow::bail!( anyhow::bail!(
"download failed for '{}' (HTTP {})", "download failed for '{}' (HTTP {})",
asset.name, filename,
resp.status(), resp.status(),
); );
} }
@@ -161,28 +154,16 @@ fn parse_checksum(sums_text: &str, filename: &str) -> anyhow::Result<String> {
async fn download_and_verify( async fn download_and_verify(
client: &reqwest::Client, client: &reqwest::Client,
release: &GithubRelease, tag: &str,
platform: &str, platform: &str,
dest: &Path, dest: &Path,
) -> anyhow::Result<()> { ) -> anyhow::Result<()> {
let archive_name = format!("aether-proxy-{}.tar.gz", platform); let archive_name = format!("aether-proxy-{}.tar.gz", platform);
let archive_asset = release
.assets
.iter()
.find(|a| a.name == archive_name)
.ok_or_else(|| anyhow::anyhow!("asset '{}' not found in release", archive_name))?;
let checksum_asset = release
.assets
.iter()
.find(|a| a.name == "SHA256SUMS.txt")
.ok_or_else(|| anyhow::anyhow!("SHA256SUMS.txt not found in release"))?;
eprintln!(" Downloading {}...", archive_name); eprintln!(" Downloading {}...", archive_name);
let (archive_bytes, checksum_bytes) = tokio::try_join!( let (archive_bytes, checksum_bytes) = tokio::try_join!(
download_asset_bytes(client, archive_asset), download_release_file(client, tag, &archive_name),
download_asset_bytes(client, checksum_asset), download_release_file(client, tag, "SHA256SUMS.txt"),
)?; )?;
let checksum_text = String::from_utf8(checksum_bytes)?; let checksum_text = String::from_utf8(checksum_bytes)?;
@@ -348,7 +329,7 @@ pub async fn cmd_upgrade(version: Option<String>) -> anyhow::Result<()> {
eprintln!(" Upgrading: {} -> {}", CURRENT_VERSION, target_semver); eprintln!(" Upgrading: {} -> {}", CURRENT_VERSION, target_semver);
eprintln!(); eprintln!();
if let Err(e) = download_and_verify(&client, &release, platform, &temp_path).await { if let Err(e) = download_and_verify(&client, target_tag, platform, &temp_path).await {
let _ = std::fs::remove_file(&temp_path); let _ = std::fs::remove_file(&temp_path);
return Err(e); return Err(e);
} }

View File

@@ -452,6 +452,13 @@ const formatLatency = (ms: number | undefined | null): string => {
return `${ms}ms` return `${ms}ms`
} }
// 格式化字节大小
const formatSize = (bytes: number): string => {
if (bytes >= 1048576) return `${(bytes / 1048576).toFixed(1)}MB`
if (bytes >= 1024) return `${(bytes / 1024).toFixed(1)}KB`
return `${bytes}B`
}
// 代理 timing 分阶段展示 // 代理 timing 分阶段展示
const proxyTimingBreakdown = (proxy: Record<string, any>): string => { const proxyTimingBreakdown = (proxy: Record<string, any>): string => {
const t = proxy.timing const t = proxy.timing
@@ -459,9 +466,16 @@ const proxyTimingBreakdown = (proxy: Record<string, any>): string => {
const parts: string[] = [] const parts: string[] = []
// 新版细分字段aether-proxy 新版本上报 // 读取 + 解压合并显示(含压缩率
if (t.body_read_ms != null && t.body_read_ms > 0) { const readDecompress = (t.body_read_ms || 0) + (t.decompress_ms || 0)
parts.push(`读取 ${formatLatency(t.body_read_ms)}`) if (readDecompress > 0) {
let label = `读取 ${formatLatency(readDecompress)}`
if (t.decompress_ms != null && t.decompress_ms > 0 && t.wire_size != null && t.body_size != null && t.body_size > 0) {
const ratio = Math.round((1 - t.wire_size / t.body_size) * 100)
label += ` ${formatSize(t.wire_size)}${formatSize(t.body_size)}`
if (ratio > 0) label += ` -${ratio}%`
}
parts.push(label)
} }
parts.push(`DNS ${formatLatency(t.dns_ms)}`) parts.push(`DNS ${formatLatency(t.dns_ms)}`)
@@ -471,7 +485,6 @@ const proxyTimingBreakdown = (proxy: Record<string, any>): string => {
if (proxy.ttfb_ms != null && t.total_ms != null) { if (proxy.ttfb_ms != null && t.total_ms != null) {
const gap = proxy.ttfb_ms - t.total_ms const gap = proxy.ttfb_ms - t.total_ms
if (gap > 500) { if (gap > 500) {
// 超过 500ms 的差值才显示,排除正常网络 RTT
parts.push(`传输 ${formatLatency(Math.round(gap))}`) parts.push(`传输 ${formatLatency(Math.round(gap))}`)
} }
} }

View File

@@ -8,8 +8,10 @@
from __future__ import annotations from __future__ import annotations
import base64 import base64
import gzip as _gzip
import hashlib import hashlib
import hmac as _hmac import hmac as _hmac
import json as _json
import time import time
from typing import Any from typing import Any
from urllib.parse import quote, urlparse from urllib.parse import quote, urlparse
@@ -613,6 +615,9 @@ def _build_delegate_kwargs_core(
""" """
构建代发请求的核心参数post/stream 共用) 构建代发请求的核心参数post/stream 共用)
元数据通过 HTTP headers 传递X-Delegate-Method/Url/Headers
上游请求体 gzip 压缩后直接作为 HTTP body 发送,大幅减少跨国传输耗时。
Args: Args:
delegate_cfg: resolve_delegate_config 返回的配置 delegate_cfg: resolve_delegate_config 返回的配置
url: 上游实际 URL url: 上游实际 URL
@@ -621,27 +626,42 @@ def _build_delegate_kwargs_core(
timeout: 上游超时秒数 timeout: 上游超时秒数
refresh_auth: 为 True 时重新生成 HMAC 签名(用于 retry refresh_auth: 为 True 时重新生成 HMAC 签名(用于 retry
""" """
import json as _json
auth = ( auth = (
delegate_cfg["fresh_auth_header"]() delegate_cfg["fresh_auth_header"]()
if refresh_auth if refresh_auth
else delegate_cfg.get("auth_header") or delegate_cfg["fresh_auth_header"]() else delegate_cfg.get("auth_header") or delegate_cfg["fresh_auth_header"]()
) )
return { # 上游 headers base64 编码
headers_b64 = base64.b64encode(_json.dumps(headers, ensure_ascii=False).encode("utf-8")).decode(
"ascii"
)
# 构建代发请求 headers元数据
delegate_headers: dict[str, str] = {
"Authorization": auth,
"X-Delegate-Method": "POST",
"X-Delegate-Url": url,
"X-Delegate-Headers": headers_b64,
"X-Delegate-Timeout": str(int(timeout)),
}
kwargs: dict[str, Any] = {
"url": delegate_cfg["delegate_url"], "url": delegate_cfg["delegate_url"],
"json": { "headers": delegate_headers,
"method": "POST",
"url": url,
"headers": headers,
"body": _json.dumps(payload, ensure_ascii=False) if payload is not None else None,
"timeout": int(timeout),
},
"headers": {"Authorization": auth, "Content-Type": _JSON_CT},
"timeout": httpx.Timeout(timeout + 10), "timeout": httpx.Timeout(timeout + 10),
} }
# body gzip 压缩后直接作为 HTTP content
if payload is not None:
body_bytes = _json.dumps(payload, ensure_ascii=False).encode("utf-8")
compressed = _gzip.compress(body_bytes)
kwargs["content"] = compressed
kwargs["headers"]["Content-Encoding"] = "gzip"
kwargs["headers"]["Content-Type"] = _JSON_CT
return kwargs
def build_delegate_post_kwargs( def build_delegate_post_kwargs(
delegate_cfg: dict[str, Any], delegate_cfg: dict[str, Any],