i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 09:20:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(aether-proxy): 网络层全面优化与代理耗时追踪

Browse Source 
aether-proxy:
- 新增 20+ 可配置参数：Aether API/delegate/CONNECT 各阶段超时、连接池、TCP keepalive/nodelay
- AetherClient 支持指数退避重试(可配置次数/延迟)和 HTTP/2
- 新增 DNS 缓存(TTL + 容量限制)避免重复解析
- 新增并发连接数限制(Semaphore，默认基于硬件估算)
- CONNECT 隧道增加建连超时和升级超时
- TLS 增加握手超时、会话缓存(session ticket + memory cache)、ALPN 协商
- 新增 ProxyMetrics 收集请求计数和延迟，通过心跳上报
- delegate 响应注入 X-Proxy-Timing 头(dns_ms/upstream_ms/total_ms)

后端:
- StreamContext 和 handler 提取 X-Proxy-Timing 写入 proxy_info 追踪数据

前端:
- 请求时间线展示代理分阶段耗时(DNS/上游)
- Endpoint 添加按钮改为文字按钮样式
This commit is contained in:
fawney19
2026-02-11 11:30:29 +08:00
parent 68f5e7e502
commit ed3f208dda
20 changed files with 832 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
aether-proxy/src/proxy/connect.rs
View File

@@ -1,14 +1,16 @@
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use hyper::body::Incoming;
use hyper::{Request, Response};
use tokio::net::TcpStream;
use tokio::time::timeout;
use tracing::{debug, warn};
use crate::auth;
use crate::config::Config;
use crate::proxy::target_filter;
use crate::proxy::target_filter::{self, DnsCache};
/// Handle HTTP CONNECT tunnel requests.
///
@@ -18,6 +20,7 @@ pub async fn handle_connect(
config: Arc<Config>,
allowed_ports: &HashSet<u16>,
timestamp_tolerance: u64,
dns_cache: &DnsCache,
) -> Response<http_body_util::Empty<bytes::Bytes>> {
// Extract Proxy-Authorization header
let proxy_auth = req
@@ -44,30 +47,43 @@ pub async fn handle_connect(
let port = authority.port_u16().unwrap_or(443);
// Target filter: private IP + port whitelist
let target_addr = match target_filter::validate_target(&host, port, allowed_ports).await {
Ok(addr) => addr,
Err(e) => {
warn!(host = %host, port, error = %e, "CONNECT target rejected");
return forbidden(&e.to_string());
}
};
let target_addr =
match target_filter::validate_target(&host, port, allowed_ports, dns_cache).await {
Ok(addr) => addr,
Err(e) => {
warn!(host = %host, port, error = %e, "CONNECT target rejected");
return forbidden(&e.to_string());
}
};
debug!(target = %target_addr, "CONNECT tunnel establishing");
// Connect to target
let target_stream = match TcpStream::connect(target_addr).await {
Ok(s) => s,
Err(e) => {
let connect_timeout = Duration::from_secs(config.connect_timeout_secs);
let target_stream = match timeout(connect_timeout, TcpStream::connect(target_addr)).await {
Ok(Ok(s)) => s,
Ok(Err(e)) => {
warn!(target = %target_addr, error = %e, "CONNECT target connection failed");
return bad_gateway(&e.to_string());
}
Err(_) => {
warn!(target = %target_addr, "CONNECT target connection timeout");
return gateway_timeout("connect timeout");
}
};
if let Err(e) = target_stream.set_nodelay(true) {
debug!(target = %target_addr, error = %e, "failed to set TCP_NODELAY");
}
// Respond 200 and upgrade connection to raw TCP tunnel
let target_display = target_addr.to_string();
// Reuse connect_timeout for upgrade: both are connection-phase operations
// and should complete within the same order of magnitude.
let upgrade_timeout = Duration::from_secs(config.connect_timeout_secs);
tokio::task::spawn(async move {
match hyper::upgrade::on(req).await {
Ok(upgraded) => {
match timeout(upgrade_timeout, hyper::upgrade::on(req)).await {
Ok(Ok(upgraded)) => {
let mut upgraded = hyper_util::rt::TokioIo::new(upgraded);
let mut target = target_stream;
@@ -85,9 +101,12 @@ pub async fn handle_connect(
}
}
}
Err(e) => {
Ok(Err(e)) => {
warn!(target = %target_display, error = %e, "CONNECT upgrade failed");
}
Err(_) => {
warn!(target = %target_display, "CONNECT upgrade timeout");
}
}
});
@@ -133,3 +152,12 @@ fn bad_gateway(msg: &str) -> Response<http_body_util::Empty<bytes::Bytes>> {
.body(http_body_util::Empty::new())
.unwrap()
}
fn gateway_timeout(msg: &str) -> Response<http_body_util::Empty<bytes::Bytes>> {
Response::builder()
.status(504)
.header("Content-Length", "0")
.header("X-Error", msg)
.body(http_body_util::Empty::new())
.unwrap()
}

44
aether-proxy/src/proxy/delegate.rs
View File

@@ -1,6 +1,7 @@
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Instant;
use futures_util::TryStreamExt;
use http_body_util::{BodyExt, Full, Limited, StreamBody};
@@ -13,7 +14,7 @@ use url::Url;
use super::BoxBody;
use crate::auth;
use crate::config::Config;
use crate::proxy::target_filter;
use crate::proxy::target_filter::{self, DnsCache};
/// Delegation request payload sent by Aether.
#[derive(Debug, Deserialize)]
@@ -36,8 +37,11 @@ pub async fn handle_delegate(
config: Arc<Config>,
allowed_ports: &HashSet<u16>,
timestamp_tolerance: u64,
dns_cache: &DnsCache,
http_client: &reqwest::Client,
) -> Response<BoxBody> {
let total_start = Instant::now();
// Authenticate via Authorization header (same HMAC scheme as Proxy-Authorization)
let auth_header = req
.headers()
@@ -86,10 +90,12 @@ pub async fn handle_delegate(
let port = parsed_url.port_or_known_default().unwrap_or(443);
if let Err(e) = target_filter::validate_target(&host, port, allowed_ports).await {
let dns_start = Instant::now();
if let Err(e) = target_filter::validate_target(&host, port, allowed_ports, dns_cache).await {
warn!(host = %host, port, error = %e, "delegate target rejected");
return error_response(403, "target_not_allowed", &e.to_string());
}
let dns_ms = dns_start.elapsed().as_millis() as u64;
debug!(
method = %delegate_req.method,
@@ -110,8 +116,8 @@ pub async fn handle_delegate(
// 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 30s connect_timeout for connection establishment, and Aether controls
// 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,
@@ -129,6 +135,7 @@ pub async fn handle_delegate(
}
// Send upstream request
let upstream_start = Instant::now();
let upstream_resp = match upstream_req.send().await {
Ok(resp) => resp,
Err(e) => {
@@ -142,12 +149,29 @@ pub async fn handle_delegate(
return error_response(502, "upstream_connection_failed", &safe_detail);
}
};
let upstream_ms = upstream_start.elapsed().as_millis() as u64;
// Build response: pass through upstream status + headers, stream body back
let status = upstream_resp.status().as_u16();
let upstream_headers = upstream_resp.headers().clone();
debug!(url = %delegate_req.url, status, "delegate upstream response");
let total_ms = total_start.elapsed().as_millis() as u64;
debug!(
url = %delegate_req.url,
status,
dns_ms,
upstream_ms,
total_ms,
"delegate upstream response"
);
// Inject proxy timing header for Aether to parse
let timing = serde_json::json!({
"dns_ms": dns_ms,
"upstream_ms": upstream_ms,
"total_ms": total_ms,
});
// Stream the response body
let body_stream = upstream_resp
@@ -161,10 +185,14 @@ pub async fn handle_delegate(
for (name, value) in upstream_headers.iter() {
builder = builder.header(name, value);
}
builder = builder.header("X-Proxy-Timing", timing.to_string());
builder
.body(stream_body)
.unwrap_or_else(|_| Response::builder().status(500).body(super::empty_box_body()).unwrap())
builder.body(stream_body).unwrap_or_else(|_| {
Response::builder()
.status(500)
.body(super::empty_box_body())
.unwrap()
})
}
// ── Sanitisation ─────────────────────────────────────────────────────────────

62
aether-proxy/src/proxy/server.rs
View File

@@ -1,6 +1,7 @@
use std::net::SocketAddr;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::{Duration, Instant};
use http_body_util::BodyExt;
use hyper::body::Incoming;
@@ -11,6 +12,7 @@ use hyper::{Method, Request, Response};
use hyper_util::rt::TokioIo;
use tokio::net::TcpListener;
use tokio::sync::watch;
use tokio::time::timeout;
use tracing::{debug, info, warn};
use crate::proxy::{connect, delegate, tls, BoxBody};
@@ -39,6 +41,8 @@ pub async fn run(
info!(addr = %addr, "proxy server listening (HTTP only)");
}
let handshake_timeout = Duration::from_secs(state.config.tls_handshake_timeout_secs);
loop {
tokio::select! {
result = listener.accept() => {
@@ -52,15 +56,38 @@ pub async fn run(
debug!(peer = %peer_addr, "new connection");
if let Err(e) = stream.set_nodelay(true) {
debug!(peer = %peer_addr, error = %e, "failed to set TCP_NODELAY");
}
let permit = match state.connection_semaphore.clone().try_acquire_owned() {
Ok(permit) => permit,
Err(_) => {
warn!(peer = %peer_addr, "connection rejected: limit reached");
continue;
}
};
let state = Arc::clone(state);
state.active_connections.fetch_add(1, Ordering::Relaxed);
tokio::task::spawn(async move {
let _permit = permit;
// Dual-stack: peek first byte to decide TLS vs plain HTTP
if let Some(ref acceptor) = state.tls_acceptor {
if tls::is_tls_client_hello(&stream).await {
match acceptor.clone().accept(stream).await {
Ok(tls_stream) => {
let is_tls = match timeout(handshake_timeout, tls::is_tls_client_hello(&stream)).await {
Ok(v) => v,
Err(_) => {
debug!(peer = %peer_addr, "TLS detection timeout");
state.active_connections.fetch_sub(1, Ordering::Relaxed);
return;
}
};
if is_tls {
match timeout(handshake_timeout, acceptor.clone().accept(stream)).await {
Ok(Ok(tls_stream)) => {
debug!(peer = %peer_addr, "TLS handshake ok");
serve_connection(
TokioIo::new(tls_stream),
@@ -69,9 +96,12 @@ pub async fn run(
)
.await;
}
Err(e) => {
Ok(Err(e)) => {
debug!(peer = %peer_addr, error = %e, "TLS handshake failed");
}
Err(_) => {
debug!(peer = %peer_addr, "TLS handshake timeout");
}
}
state.active_connections.fetch_sub(1, Ordering::Relaxed);
return;
@@ -107,13 +137,18 @@ where
let config = Arc::clone(&state.config);
let dynamic = Arc::clone(&state.dynamic);
let delegate_client = state.delegate_client.clone();
let dns_cache = Arc::clone(&state.dns_cache);
let metrics = Arc::clone(&state.metrics);
let service = service_fn(move |req: Request<Incoming>| {
let config = Arc::clone(&config);
let dynamic = Arc::clone(&dynamic);
let delegate_client = delegate_client.clone();
let dns_cache = Arc::clone(&dns_cache);
let metrics = Arc::clone(&metrics);
async move {
let start = Instant::now();
// Snapshot current dynamic values (may be updated by remote config)
let (allowed_ports, timestamp_tolerance) = {
let d = dynamic.read().unwrap();
@@ -121,13 +156,20 @@ where
};
if req.method() == Method::CONNECT {
let resp =
connect::handle_connect(req, config, &allowed_ports, timestamp_tolerance).await;
let resp = connect::handle_connect(
req,
config,
&allowed_ports,
timestamp_tolerance,
dns_cache.as_ref(),
)
.await;
let resp = resp.map(|_| -> BoxBody {
http_body_util::Empty::new()
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { match e {} })
.boxed()
});
metrics.record_request(start.elapsed());
Ok::<_, hyper::Error>(resp)
} else if req.uri().path() == "/_aether/delegate" && req.method() == hyper::Method::POST
{
@@ -136,19 +178,23 @@ where
config,
&allowed_ports,
timestamp_tolerance,
dns_cache.as_ref(),
&delegate_client,
)
.await;
metrics.record_request(start.elapsed());
Ok(resp)
} else {
// Only CONNECT tunnels and /_aether/delegate are supported;
// plain HTTP forward proxy was removed (all API traffic is HTTPS).
Ok(Response::builder()
let resp = Response::builder()
.status(405)
.header("Allow", "CONNECT")
.header("Content-Length", "0")
.body(crate::proxy::empty_box_body())
.unwrap())
.unwrap();
metrics.record_request(start.elapsed());
Ok(resp)
}
}
});

106
aether-proxy/src/proxy/target_filter.rs
View File

@@ -1,5 +1,8 @@
use std::collections::HashSet;
use std::collections::{HashMap, HashSet};
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
/// Check if an IP address belongs to a private/reserved network.
fn is_private_ip(ip: &IpAddr) -> bool {
@@ -80,6 +83,85 @@ impl std::fmt::Display for FilterError {
}
}
struct DnsCacheEntry {
addr: SocketAddr,
expires_at: Instant,
inserted_at: Instant,
}
/// Lightweight DNS cache with TTL + capacity bounds.
pub struct DnsCache {
ttl: Duration,
capacity: usize,
entries: RwLock<HashMap<String, DnsCacheEntry>>,
}
impl DnsCache {
pub fn new(ttl: Duration, capacity: usize) -> Self {
Self {
ttl,
capacity,
entries: RwLock::new(HashMap::new()),
}
}
pub async fn get(&self, host: &str, port: u16) -> Option<SocketAddr> {
if self.capacity == 0 || self.ttl.is_zero() {
return None;
}
let key = Self::key(host, port);
let now = Instant::now();
// Fast path: read lock for cache hit
{
let entries = self.entries.read().await;
match entries.get(&key) {
Some(entry) if entry.expires_at > now => return Some(entry.addr),
None => return None,
Some(_) => {} // expired, fall through to evict
}
}
// Slow path: write lock to remove expired entry
let mut entries = self.entries.write().await;
entries.remove(&key);
None
}
pub async fn insert(&self, host: &str, port: u16, addr: SocketAddr) {
if self.capacity == 0 || self.ttl.is_zero() {
return;
}
let key = Self::key(host, port);
let now = Instant::now();
let mut entries = self.entries.write().await;
entries.retain(|_, entry| entry.expires_at > now);
while entries.len() >= self.capacity {
let oldest_key = entries
.iter()
.min_by_key(|(_, entry)| entry.inserted_at)
.map(|(key, _)| key.clone());
if let Some(key) = oldest_key {
entries.remove(&key);
} else {
break;
}
}
entries.insert(
key,
DnsCacheEntry {
addr,
expires_at: now + self.ttl,
inserted_at: now,
},
);
}
fn key(host: &str, port: u16) -> String {
format!("{}:{}", host, port)
}
}
/// Validate that the target host:port is allowed.
///
/// Uses async DNS resolution (via `tokio::net::lookup_host`) to avoid
@@ -90,6 +172,7 @@ pub async fn validate_target(
host: &str,
port: u16,
allowed_ports: &HashSet<u16>,
dns_cache: &DnsCache,
) -> Result<SocketAddr, FilterError> {
// Port whitelist check
if !allowed_ports.contains(&port) {
@@ -104,6 +187,10 @@ pub async fn validate_target(
return Ok(SocketAddr::new(ip, port));
}
if let Some(addr) = dns_cache.get(host, port).await {
return Ok(addr);
}
// Async DNS resolution with private IP check (DNS rebinding protection)
let addr_str = format!("{}:{}", host, port);
let addrs: Vec<SocketAddr> = tokio::net::lookup_host(&addr_str)
@@ -123,7 +210,9 @@ pub async fn validate_target(
}
// Return the first valid address
Ok(addrs[0])
let selected = addrs[0];
dns_cache.insert(host, port, selected).await;
Ok(selected)
}
#[cfg(test)]
@@ -134,6 +223,10 @@ mod tests {
[80, 443, 8080, 8443].into_iter().collect()
}
fn cache() -> DnsCache {
DnsCache::new(Duration::from_secs(60), 128)
}
#[test]
fn test_private_ipv4() {
assert!(is_private_ip(&IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1))));
@@ -162,19 +255,22 @@ mod tests {
#[tokio::test]
async fn test_port_not_allowed() {
let result = validate_target("8.8.8.8", 22, &ports()).await;
let cache = cache();
let result = validate_target("8.8.8.8", 22, &ports(), &cache).await;
assert!(matches!(result, Err(FilterError::PortNotAllowed(22))));
}
#[tokio::test]
async fn test_private_ip_blocked() {
let result = validate_target("127.0.0.1", 80, &ports()).await;
let cache = cache();
let result = validate_target("127.0.0.1", 80, &ports(), &cache).await;
assert!(matches!(result, Err(FilterError::PrivateIp(_))));
}
#[tokio::test]
async fn test_public_ip_allowed() {
let result = validate_target("8.8.8.8", 443, &ports()).await;
let cache = cache();
let result = validate_target("8.8.8.8", 443, &ports(), &cache).await;
assert!(result.is_ok());
}
}

15
aether-proxy/src/proxy/tls.rs
View File

@@ -9,6 +9,8 @@ use sha2::{Digest, Sha256};
use tokio_rustls::TlsAcceptor;
use tracing::{info, warn};
const SESSION_CACHE_SIZE: usize = 1024;
/// Generate a self-signed certificate if the files do not already exist.
///
/// The certificate includes SANs: `localhost` and `aether-proxy`.
@@ -73,10 +75,21 @@ pub fn build_tls_acceptor(cert_path: &Path, key_path: &Path) -> anyhow::Result<T
rustls_pemfile::private_key(&mut BufReader::new(key_file))?
.ok_or_else(|| anyhow::anyhow!("no private key found in {}", key_path.display()))?;
let config = rustls::ServerConfig::builder()
let mut config = rustls::ServerConfig::builder()
.with_no_client_auth()
.with_single_cert(certs, key)?;
config.alpn_protocols = vec![b"http/1.1".to_vec()];
config.session_storage = rustls::server::ServerSessionMemoryCache::new(SESSION_CACHE_SIZE);
match rustls::crypto::ring::Ticketer::new() {
Ok(ticketer) => {
config.ticketer = ticketer;
}
Err(e) => {
warn!(error = %e, "failed to init TLS ticketer; tickets disabled");
}
}
Ok(TlsAcceptor::from(Arc::new(config)))
}