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

refactor(proxy): 将 aether-proxy 从 HMAC 正向代理迁移到 WebSocket 隧道模式

Browse Source 
移除 HMAC 认证、TLS 自签名证书、HTTP CONNECT 代理和代发(delegate)模式，
改为 aether-proxy 主动通过 WebSocket 连接 Aether 服务端建立隧道。

Aether 服务端新增:
- WebSocket 隧道端点 (proxy_tunnel.py)
- TunnelManager 管理隧道连接和请求分发
- TunnelTransport 作为 httpx 自定义 transport 层
- 基于二进制帧的隧道协议 (tunnel_protocol.py)

aether-proxy (Rust) 重构:
- 新增 tunnel 模块 (client/dispatcher/stream_handler/protocol)
- 支持多 Aether 服务端连接 ([[servers]] 配置)
- 移除 proxy/auth/delegate 模块和 hyper 依赖
- 改用 tokio-tungstenite 实现 WebSocket 客户端

同时:
- 添加浏览器指纹 Headers 绕过 Cloudflare 防护
- 删除节点时自动清理 Provider/Endpoint 的代理引用
- 数据库迁移: 新增 tunnel_mode/tunnel_connected/tunnel_connected_at 字段
This commit is contained in:
fawney19
2026-02-25 21:59:29 +08:00
parent 39b036abd5
commit fd9040b9aa
53 changed files with 2938 additions and 2728 deletions
Download Patch File Download Diff File Expand all files Collapse all files
aether-proxy/.env.example
-6
View File
@@ -4,11 +4,5 @@ AETHER_PROXY_AETHER_URL=https://aether.example.com
# Management Token (ae_xxx, must belong to an ADMIN user)
AETHER_PROXY_MANAGEMENT_TOKEN=ae_xxxxx
# HMAC key (must match Aether's PROXY_HMAC_KEY)
AETHER_PROXY_HMAC_KEY=
# Proxy listen port
AETHER_PROXY_LISTEN_PORT=18080
# Node identification
AETHER_PROXY_NODE_NAME=proxy-01
aether-proxy/Cargo.lock
Generated
+94 -73
View File
@@ -10,7 +10,7 @@ checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
[[package]]
name = "aether-proxy"
version = "0.1.4"
version = "0.1.6"
dependencies = [
"anyhow",
"base64",
@@ -20,32 +20,22 @@ dependencies = [
"flate2",
"futures-util",
"hex",
"hmac",
"http-body-util",
"hyper",
"hyper-util",
"libc",
"ratatui",
"rcgen",
"reqwest",
"rustls",
"rustls-pemfile",
"rustls-pki-types",
"serde",
"serde_json",
"sha2",
"subtle",
"sysinfo",
"tar",
"thiserror 2.0.18",
"tokio",
"tokio-rustls",
"tokio-tungstenite",
"toml",
"tower-service",
"tracing",
"tracing-subscriber",
"url",
"webpki-roots",
]
[[package]]
@@ -216,6 +206,12 @@ version = "1.25.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c8efb64bd706a16a1bdde310ae86b351e4d21550d98d056f22f8a7f7a2183fec"
[[package]]
name = "byteorder"
version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
[[package]]
name = "bytes"
version = "1.11.1"
@@ -479,6 +475,12 @@ dependencies = [
"syn 2.0.114",
]
[[package]]
name = "data-encoding"
version = "2.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d7a1e2f27636f116493b8b860f5546edb47c8d8f8ea73e1d2a20be88e28d1fea"
[[package]]
name = "deltae"
version = "0.3.2"
@@ -524,7 +526,6 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
dependencies = [
"block-buffer",
"crypto-common",
"subtle",
]
[[package]]
@@ -811,15 +812,6 @@ version = "0.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"
[[package]]
name = "hmac"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e"
dependencies = [
"digest",
]
[[package]]
name = "http"
version = "1.4.0"
@@ -859,12 +851,6 @@ version = "1.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
[[package]]
name = "httpdate"
version = "1.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
[[package]]
name = "hyper"
version = "1.8.1"
@@ -879,7 +865,6 @@ dependencies = [
"http",
"http-body",
"httparse",
"httpdate",
"itoa",
"pin-project-lite",
"pin-utils",
@@ -902,7 +887,7 @@ dependencies = [
"tokio",
"tokio-rustls",
"tower-service",
"webpki-roots",
"webpki-roots 1.0.6",
]
[[package]]
@@ -1416,16 +1401,6 @@ dependencies = [
"windows-link",
]
[[package]]
name = "pem"
version = "3.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be"
dependencies = [
"base64",
"serde_core",
]
[[package]]
name = "percent-encoding"
version = "2.3.2"
@@ -1654,6 +1629,8 @@ version = "0.8.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404"
dependencies = [
"libc",
"rand_chacha 0.3.1",
"rand_core 0.6.4",
]
@@ -1663,10 +1640,20 @@ version = "0.9.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1"
dependencies = [
"rand_chacha",
"rand_chacha 0.9.0",
"rand_core 0.9.5",
]
[[package]]
name = "rand_chacha"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88"
dependencies = [
"ppv-lite86",
"rand_core 0.6.4",
]
[[package]]
name = "rand_chacha"
version = "0.9.0"
@@ -1682,6 +1669,9 @@ name = "rand_core"
version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c"
dependencies = [
"getrandom 0.2.17",
]
[[package]]
name = "rand_core"
@@ -1797,19 +1787,6 @@ dependencies = [
"crossbeam-utils",
]
[[package]]
name = "rcgen"
version = "0.13.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "75e669e5202259b5314d1ea5397316ad400819437857b90861765f24c4cf80a2"
dependencies = [
"pem",
"ring",
"rustls-pki-types",
"time",
"yasna",
]
[[package]]
name = "redox_syscall"
version = "0.5.18"
@@ -1896,7 +1873,7 @@ dependencies = [
"wasm-bindgen-futures",
"wasm-streams",
"web-sys",
"webpki-roots",
"webpki-roots 1.0.6",
]
[[package]]
@@ -1970,15 +1947,6 @@ dependencies = [
"zeroize",
]
[[package]]
name = "rustls-pemfile"
version = "2.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50"
dependencies = [
"rustls-pki-types",
]
[[package]]
name = "rustls-pki-types"
version = "1.14.0"
@@ -2089,6 +2057,17 @@ dependencies = [
"serde",
]
[[package]]
name = "sha1"
version = "0.10.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba"
dependencies = [
"cfg-if",
"cpufeatures",
"digest",
]
[[package]]
name = "sha2"
version = "0.10.9"
@@ -2488,6 +2467,22 @@ dependencies = [
"tokio",
]
[[package]]
name = "tokio-tungstenite"
version = "0.24.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9"
dependencies = [
"futures-util",
"log",
"rustls",
"rustls-pki-types",
"tokio",
"tokio-rustls",
"tungstenite",
"webpki-roots 0.26.11",
]
[[package]]
name = "tokio-util"
version = "0.7.18"
@@ -2667,6 +2662,26 @@ version = "0.2.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
[[package]]
name = "tungstenite"
version = "0.24.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a"
dependencies = [
"byteorder",
"bytes",
"data-encoding",
"http",
"httparse",
"log",
"rand 0.8.5",
"rustls",
"rustls-pki-types",
"sha1",
"thiserror 1.0.69",
"utf-8",
]
[[package]]
name = "typenum"
version = "1.19.0"
@@ -2726,6 +2741,12 @@ dependencies = [
"serde",
]
[[package]]
name = "utf-8"
version = "0.7.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
[[package]]
name = "utf8_iter"
version = "1.0.4"
@@ -2887,6 +2908,15 @@ dependencies = [
"wasm-bindgen",
]
[[package]]
name = "webpki-roots"
version = "0.26.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9"
dependencies = [
"webpki-roots 1.0.6",
]
[[package]]
name = "webpki-roots"
version = "1.0.6"
@@ -3236,15 +3266,6 @@ dependencies = [
"rustix 1.1.3",
]
[[package]]
name = "yasna"
version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e17bb3549cc1321ae1296b9cdc2698e2b6cb1992adfa19a8c72e5b7a738f44cd"
dependencies = [
"time",
]
[[package]]
name = "yoke"
version = "0.8.1"
aether-proxy/Cargo.toml
+3 -13
View File
@@ -2,19 +2,13 @@
name = "aether-proxy"
version = "0.1.6"
edition = "2021"
description = "Forward proxy for Aether with HMAC authentication"
description = "Tunnel proxy for Aether"
[dependencies]
tokio = { version = "1", features = ["full"] }
hyper = { version = "1", features = ["http1", "server"] }
hyper-util = { version = "0.1", features = ["tokio", "http1", "http2", "server", "client-legacy"] }
tower-service = "0.3"
http-body-util = "0.1"
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "stream", "http2"] }
tokio-tungstenite = { version = "0.24", features = ["rustls-tls-webpki-roots"] }
futures-util = "0.3"
hmac = "0.12"
sha2 = "0.10"
subtle = "2"
base64 = "0.22"
clap = { version = "4", features = ["derive", "env"] }
tracing = "0.1"
@@ -23,15 +17,11 @@ serde = { version = "1", features = ["derive"] }
serde_json = "1"
thiserror = "2"
bytes = "1"
sha2 = "0.10"
hex = "0.4"
anyhow = "1"
toml = "0.8"
tokio-rustls = "0.26"
webpki-roots = "1"
rustls = { version = "0.23", features = ["ring"] }
rustls-pki-types = "1"
rustls-pemfile = "2"
rcgen = "0.13"
ratatui = "0.30"
crossterm = "0.28"
url = "2"
aether-proxy/src/app.rs
+119 -104
View File
@@ -1,40 +1,38 @@
//! Application lifecycle: initialization, task orchestration, and shutdown.
//!
//! Extracted from `main.rs` to keep the entry point minimal and consolidate
//! the startup sequence, tracing init, and graceful shutdown logic.
use std::sync::atomic::AtomicU64;
use std::sync::atomic::{AtomicU32, AtomicU64};
use std::sync::{Arc, RwLock};
use std::time::Duration;
use tokio::signal;
use tokio::sync::{watch, Semaphore};
use tracing::{error, info};
use tokio::sync::watch;
use tracing::{error, info, warn};
use crate::config::Config;
use crate::config::{Config, ServerEntry};
use crate::net;
use crate::registration::client::AetherClient;
use crate::runtime::{self, DynamicConfig};
use crate::state::{AppState, ProxyMetrics};
use crate::{hardware, proxy};
use crate::state::{AppState, ProxyMetrics, ServerContext};
use crate::{hardware, target_filter, tunnel};
/// Run the full application lifecycle after config has been parsed.
pub async fn run(mut config: Config) -> anyhow::Result<()> {
pub async fn run(mut config: Config, servers: Vec<ServerEntry>) -> anyhow::Result<()> {
init_tracing(&config);
info!(
version = env!("CARGO_PKG_VERSION"),
port = config.listen_port,
node_name = %config.node_name,
"aether-proxy starting"
server_count = servers.len(),
"aether-proxy starting (tunnel mode)"
);
// Resolve public IP
// Resolve public IP (best-effort for region info)
let public_ip = match &config.public_ip {
Some(ip) => ip.clone(),
None => net::detect_public_ip().await?,
None => net::detect_public_ip()
.await
.unwrap_or_else(|_| "0.0.0.0".to_string()),
};
info!(public_ip = %public_ip, "using public IP");
// Auto-detect region if not configured
if config.node_region.is_none() {
@@ -43,118 +41,136 @@ pub async fn run(mut config: Config) -> anyhow::Result<()> {
}
}
// Initialize TLS if enabled
let (tls_acceptor, tls_fingerprint) = if config.enable_tls {
let cert_path = std::path::PathBuf::from(&config.tls_cert);
let key_path = std::path::PathBuf::from(&config.tls_key);
proxy::tls::ensure_self_signed_cert(&cert_path, &key_path)?;
let acceptor = proxy::tls::build_tls_acceptor(&cert_path, &key_path)?;
let fingerprint = proxy::tls::cert_sha256_fingerprint(&cert_path)?;
info!(fingerprint = %fingerprint, "TLS enabled");
(Some(acceptor), Some(fingerprint))
} else {
info!("TLS disabled");
(None, None)
};
// Collect hardware info (once at startup)
// Collect hardware info (once at startup, sent during registration)
let hw_info = hardware::collect();
let max_connections_raw = config
.max_concurrent_connections
.unwrap_or(hw_info.estimated_max_concurrency)
.max(1);
let max_connections = usize::try_from(max_connections_raw).unwrap_or(usize::MAX);
// Auto-detect tunnel_max_streams from hardware if not explicitly set
if config.tunnel_max_streams.is_none() {
let auto = (hw_info.estimated_max_concurrency / 10).clamp(64, 1024) as u32;
config.tunnel_max_streams = Some(auto);
info!(
tunnel_max_streams = auto,
"auto-detected tunnel_max_streams from hardware"
);
}
info!(
max_connections = max_connections_raw,
"connection limit configured"
max_concurrency = hw_info.estimated_max_concurrency,
"hardware info collected"
);
let connection_semaphore = Arc::new(Semaphore::new(max_connections));
let metrics = Arc::new(ProxyMetrics::new());
let dns_cache = Arc::new(proxy::target_filter::DnsCache::new(
let dns_cache = Arc::new(target_filter::DnsCache::new(
Duration::from_secs(config.dns_cache_ttl_secs),
config.dns_cache_capacity,
));
// Register with Aether
let aether_client = Arc::new(AetherClient::new(&config));
let node_id = aether_client
.register(
// Build reqwest client for tunnel upstream requests (shared).
let reqwest_client = reqwest::Client::builder()
.pool_max_idle_per_host(config.upstream_pool_max_idle_per_host)
.pool_idle_timeout(Duration::from_secs(config.upstream_pool_idle_timeout_secs))
.connect_timeout(Duration::from_secs(config.upstream_connect_timeout_secs))
.tcp_nodelay(config.upstream_tcp_nodelay)
.build()
.expect("failed to build reqwest client");
// Register with each Aether server and build per-server contexts
let mut server_contexts: Vec<Arc<ServerContext>> = Vec::new();
for (i, entry) in servers.iter().enumerate() {
let label = if servers.len() == 1 {
"server".to_string()
} else {
format!("server-{}", i)
};
let node_name = entry
.node_name
.clone()
.unwrap_or_else(|| config.node_name.clone());
let client = Arc::new(AetherClient::new(
&config,
&public_ip,
config.enable_tls,
tls_fingerprint.as_deref(),
Some(&hw_info),
)
.await?;
&entry.aether_url,
&entry.management_token,
));
match client
.register(&config, &node_name, &public_ip, Some(&hw_info))
.await
{
Ok(node_id) => {
info!(server = %label, node_id = %node_id, url = %entry.aether_url, node_name = %node_name, "registered");
server_contexts.push(Arc::new(ServerContext {
server_label: label,
aether_url: entry.aether_url.clone(),
management_token: entry.management_token.clone(),
node_name,
node_id: Arc::new(RwLock::new(node_id)),
aether_client: client,
dynamic: Arc::new(RwLock::new(DynamicConfig::from_config(&config))),
active_connections: Arc::new(AtomicU64::new(0)),
metrics: Arc::new(ProxyMetrics::new()),
reconnect_attempts: AtomicU32::new(0),
}));
}
Err(e) => {
warn!(
server = %label,
url = %entry.aether_url,
error = %e,
"registration failed, skipping server"
);
}
}
}
info!(node_id = %node_id, "node registered");
// Build DynamicConfig before moving config into Arc
let dynamic = Arc::new(RwLock::new(DynamicConfig::from_config(&config)));
// Build delegate HTTP client (for proxy-initiated upstream requests).
let delegate_client = proxy::delegate_client::build_delegate_client(&config);
if server_contexts.is_empty() {
anyhow::bail!("no servers registered successfully");
}
// Build shared application state
let state = Arc::new(AppState {
config: Arc::new(config),
node_id: Arc::new(RwLock::new(node_id)),
dynamic,
aether_client,
hardware_info: Arc::new(hw_info),
public_ip,
tls_fingerprint,
tls_acceptor,
delegate_client,
active_connections: Arc::new(AtomicU64::new(0)),
connection_semaphore,
dns_cache,
metrics,
reqwest_client,
});
// Shutdown signal channel
let (shutdown_tx, shutdown_rx) = watch::channel(false);
// Start heartbeat task
let heartbeat_handle = {
let state = Arc::clone(&state);
info!(
active_servers = server_contexts.len(),
"running in tunnel mode"
);
// Spawn one tunnel task per server
let mut tunnel_handles = Vec::new();
for server in &server_contexts {
let s = Arc::clone(&state);
let srv = Arc::clone(server);
let rx = shutdown_rx.clone();
tokio::spawn(async move {
crate::registration::heartbeat::run(&state, rx).await;
})
};
// Start proxy server
let server_handle = {
let state = Arc::clone(&state);
let rx = shutdown_rx.clone();
tokio::spawn(async move {
if let Err(e) = proxy::server::run(&state, rx).await {
error!(error = %e, "proxy server error");
}
})
};
// Wait for shutdown signal (SIGTERM or SIGINT)
wait_for_shutdown().await;
info!("shutdown signal received, cleaning up...");
// Signal all tasks to stop
let _ = shutdown_tx.send(true);
// Graceful unregister (best-effort)
let current_node_id = state.node_id.read().unwrap().clone();
if let Err(e) = state.aether_client.unregister(&current_node_id).await {
error!(error = %e, "unregister failed during shutdown");
tunnel_handles.push(tokio::spawn(async move {
tunnel::run(&s, &srv, rx).await;
}));
}
// Wait for tasks to finish
let _ = tokio::join!(heartbeat_handle, server_handle);
// Wait for shutdown signal
wait_for_shutdown().await;
info!("shutdown signal received, cleaning up...");
let _ = shutdown_tx.send(true);
// Graceful unregister from all servers
for server in &server_contexts {
let node_id = server.node_id.read().unwrap().clone();
if let Err(e) = server.aether_client.unregister(&node_id).await {
error!(
server = %server.server_label,
error = %e,
"unregister failed during shutdown"
);
}
}
// Wait for all tunnel tasks
for h in tunnel_handles {
let _ = h.await;
}
info!("aether-proxy stopped");
Ok(())
@@ -168,7 +184,6 @@ fn init_tracing(config: &Config) {
let (filter_layer, reload_handle) = reload::Layer::new(filter);
// Register log-level hot-reloader
runtime::set_log_reloader(Box::new(move |level: &str| {
if let Ok(new_filter) = EnvFilter::try_new(level) {
let _ = reload_handle.modify(|f| *f = new_filter);
aether-proxy/src/auth/hmac.rs
-196
View File
@@ -1,196 +0,0 @@
use base64::Engine;
use hmac::{Hmac, Mac};
use sha2::Sha256;
use subtle::ConstantTimeEq;
use crate::config::Config;
type HmacSha256 = Hmac<Sha256>;
#[derive(Debug)]
pub enum AuthError {
MissingHeader,
InvalidBasicAuth,
InvalidUsername,
InvalidPasswordFormat,
TimestampParseError,
TimestampExpired,
SignatureMismatch,
}
impl std::fmt::Display for AuthError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::MissingHeader => write!(f, "missing Proxy-Authorization header"),
Self::InvalidBasicAuth => write!(f, "invalid Basic auth encoding"),
Self::InvalidUsername => write!(f, "username must be 'hmac'"),
Self::InvalidPasswordFormat => {
write!(f, "password format must be 'timestamp.signature'")
}
Self::TimestampParseError => write!(f, "invalid timestamp"),
Self::TimestampExpired => write!(f, "timestamp outside tolerance window"),
Self::SignatureMismatch => write!(f, "HMAC signature mismatch"),
}
}
}
/// Validate Proxy-Authorization header.
///
/// Expected format: `Basic base64(hmac:{timestamp}.{signature})`
/// where signature = hex(HMAC-SHA256(hmac_key, "{timestamp}"))
///
/// The signature no longer includes `node_id`, eliminating race conditions
/// during re-registration where the Aether server's cached `node_id` could
/// differ from the proxy's freshly assigned `node_id`.
///
/// `timestamp_tolerance` is accepted separately so the caller can supply
/// the value from [`DynamicConfig`](crate::runtime::DynamicConfig) (which
/// may be updated remotely).
pub fn validate_proxy_auth(
proxy_auth_header: Option<&str>,
config: &Config,
timestamp_tolerance: u64,
) -> Result<(), AuthError> {
let header = proxy_auth_header.ok_or(AuthError::MissingHeader)?;
let encoded = header
.strip_prefix("Basic ")
.or_else(|| header.strip_prefix("basic "))
.ok_or(AuthError::InvalidBasicAuth)?;
let decoded_bytes = base64::engine::general_purpose::STANDARD
.decode(encoded.trim())
.map_err(|_| AuthError::InvalidBasicAuth)?;
let decoded = String::from_utf8(decoded_bytes).map_err(|_| AuthError::InvalidBasicAuth)?;
// format: hmac:{timestamp}.{signature}
let (username, password) = decoded.split_once(':').ok_or(AuthError::InvalidBasicAuth)?;
if username != "hmac" {
return Err(AuthError::InvalidUsername);
}
let (timestamp_str, signature_hex) = password
.split_once('.')
.ok_or(AuthError::InvalidPasswordFormat)?;
// Validate timestamp window
let timestamp: u64 = timestamp_str
.parse()
.map_err(|_| AuthError::TimestampParseError)?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.expect("system clock before epoch")
.as_secs();
let diff = now.abs_diff(timestamp);
if diff > timestamp_tolerance {
return Err(AuthError::TimestampExpired);
}
// Recompute signature: HMAC-SHA256(key, timestamp)
let mut mac =
HmacSha256::new_from_slice(config.hmac_key.as_bytes()).expect("HMAC accepts any key size");
mac.update(timestamp_str.as_bytes());
let expected = mac.finalize().into_bytes();
let expected_hex = hex::encode(expected);
// Constant-time comparison
let sig_bytes = signature_hex.as_bytes();
let exp_bytes = expected_hex.as_bytes();
if sig_bytes.len() != exp_bytes.len() || sig_bytes.ct_eq(exp_bytes).unwrap_u8() != 1 {
return Err(AuthError::SignatureMismatch);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn make_config() -> Config {
Config {
aether_url: String::new(),
management_token: String::new(),
hmac_key: "test-hmac-key".to_string(),
listen_port: 18080,
public_ip: None,
node_name: "test".to_string(),
node_region: None,
heartbeat_interval: 30,
allowed_ports: vec![80, 443],
timestamp_tolerance: 300,
aether_request_timeout_secs: 10,
aether_connect_timeout_secs: 10,
aether_pool_max_idle_per_host: 8,
aether_pool_idle_timeout_secs: 90,
aether_tcp_keepalive_secs: 60,
aether_tcp_nodelay: true,
aether_http2: true,
aether_retry_max_attempts: 3,
aether_retry_base_delay_ms: 200,
aether_retry_max_delay_ms: 2000,
max_concurrent_connections: None,
connect_timeout_secs: 30,
tls_handshake_timeout_secs: 10,
dns_cache_ttl_secs: 60,
dns_cache_capacity: 1024,
delegate_connect_timeout_secs: 30,
delegate_pool_max_idle_per_host: 64,
delegate_pool_idle_timeout_secs: 300,
delegate_tcp_keepalive_secs: 60,
delegate_tcp_nodelay: true,
log_level: "info".to_string(),
log_json: false,
enable_tls: false,
tls_cert: String::new(),
tls_key: String::new(),
}
}
fn make_valid_auth(config: &Config) -> String {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs();
let mut mac = HmacSha256::new_from_slice(config.hmac_key.as_bytes()).unwrap();
mac.update(now.to_string().as_bytes());
let sig = hex::encode(mac.finalize().into_bytes());
let cred = format!("hmac:{}.{}", now, sig);
let encoded = base64::engine::general_purpose::STANDARD.encode(cred);
format!("Basic {}", encoded)
}
#[test]
fn test_valid_auth() {
let config = make_config();
let header = make_valid_auth(&config);
assert!(validate_proxy_auth(Some(&header), &config, config.timestamp_tolerance).is_ok());
}
#[test]
fn test_missing_header() {
let config = make_config();
assert!(matches!(
validate_proxy_auth(None, &config, config.timestamp_tolerance),
Err(AuthError::MissingHeader)
));
}
#[test]
fn test_wrong_username() {
let cred = "user:12345.abc";
let encoded = base64::engine::general_purpose::STANDARD.encode(cred);
let header = format!("Basic {}", encoded);
let config = make_config();
assert!(matches!(
validate_proxy_auth(Some(&header), &config, config.timestamp_tolerance),
Err(AuthError::InvalidUsername)
));
}
}
aether-proxy/src/auth/mod.rs
-3
View File
@@ -1,3 +0,0 @@
pub mod hmac;
pub use self::hmac::validate_proxy_auth;
aether-proxy/src/config.rs
+122 -93
View File
@@ -3,11 +3,11 @@ use std::path::Path;
use clap::Parser;
use serde::{Deserialize, Serialize};
/// Aether forward proxy with HMAC authentication.
/// Aether tunnel proxy.
///
/// Deployed on overseas VPS to relay API traffic for Aether instances
/// behind the GFW. Registers with Aether, sends heartbeats, and validates
/// incoming proxy requests via HMAC-SHA256 signatures in Basic Auth.
/// behind the GFW. Connects to Aether via WebSocket tunnel, registers
/// with Aether, and relays upstream requests.
#[derive(Parser, Debug, Clone)]
#[command(version, about)]
pub struct Config {
@@ -19,14 +19,6 @@ pub struct Config {
#[arg(long, env = "AETHER_PROXY_MANAGEMENT_TOKEN")]
pub management_token: String,
/// HMAC-SHA256 key for proxy authentication
#[arg(long, env = "AETHER_PROXY_HMAC_KEY")]
pub hmac_key: String,
/// Port to listen on for proxy connections
#[arg(long, env = "AETHER_PROXY_LISTEN_PORT", default_value_t = 18080)]
pub listen_port: u16,
/// Public IP address of this node (auto-detected if omitted)
#[arg(long, env = "AETHER_PROXY_PUBLIC_IP")]
pub public_ip: Option<String>,
@@ -52,10 +44,6 @@ pub struct Config {
)]
pub allowed_ports: Vec<u16>,
/// Timestamp tolerance window in seconds for HMAC validation
#[arg(long, env = "AETHER_PROXY_TIMESTAMP_TOLERANCE", default_value_t = 300)]
pub timestamp_tolerance: u64,
/// Aether API request timeout in seconds
#[arg(
long,
@@ -128,14 +116,6 @@ pub struct Config {
#[arg(long, env = "AETHER_PROXY_MAX_CONCURRENT_CONNECTIONS")]
pub max_concurrent_connections: Option<u64>,
/// Upstream TCP connect timeout in seconds for CONNECT tunnels
#[arg(long, env = "AETHER_PROXY_CONNECT_TIMEOUT", default_value_t = 30)]
pub connect_timeout_secs: u64,
/// TLS handshake timeout in seconds for incoming TLS connections
#[arg(long, env = "AETHER_PROXY_TLS_HANDSHAKE_TIMEOUT", default_value_t = 10)]
pub tls_handshake_timeout_secs: u64,
/// DNS cache TTL in seconds
#[arg(long, env = "AETHER_PROXY_DNS_CACHE_TTL", default_value_t = 60)]
pub dns_cache_ttl_secs: u64,
@@ -144,45 +124,45 @@ pub struct Config {
#[arg(long, env = "AETHER_PROXY_DNS_CACHE_CAPACITY", default_value_t = 1024)]
pub dns_cache_capacity: usize,
/// Delegate HTTP client connect timeout in seconds
/// Upstream HTTP client connect timeout in seconds
#[arg(
long,
env = "AETHER_PROXY_DELEGATE_CONNECT_TIMEOUT",
env = "AETHER_PROXY_UPSTREAM_CONNECT_TIMEOUT",
default_value_t = 30
)]
pub delegate_connect_timeout_secs: u64,
pub upstream_connect_timeout_secs: u64,
/// Delegate HTTP client max idle connections per host
/// Upstream HTTP client max idle connections per host
#[arg(
long,
env = "AETHER_PROXY_DELEGATE_POOL_MAX_IDLE_PER_HOST",
env = "AETHER_PROXY_UPSTREAM_POOL_MAX_IDLE_PER_HOST",
default_value_t = 64
)]
pub delegate_pool_max_idle_per_host: usize,
pub upstream_pool_max_idle_per_host: usize,
/// Delegate HTTP client idle timeout in seconds
/// Upstream HTTP client idle timeout in seconds
#[arg(
long,
env = "AETHER_PROXY_DELEGATE_POOL_IDLE_TIMEOUT",
env = "AETHER_PROXY_UPSTREAM_POOL_IDLE_TIMEOUT",
default_value_t = 300
)]
pub delegate_pool_idle_timeout_secs: u64,
pub upstream_pool_idle_timeout_secs: u64,
/// Delegate TCP keepalive in seconds (0 disables)
/// Upstream TCP keepalive in seconds (0 disables)
#[arg(
long,
env = "AETHER_PROXY_DELEGATE_TCP_KEEPALIVE",
env = "AETHER_PROXY_UPSTREAM_TCP_KEEPALIVE",
default_value_t = 60
)]
pub delegate_tcp_keepalive_secs: u64,
pub upstream_tcp_keepalive_secs: u64,
/// Delegate TCP_NODELAY
/// Upstream TCP_NODELAY
#[arg(
long,
env = "AETHER_PROXY_DELEGATE_TCP_NODELAY",
env = "AETHER_PROXY_UPSTREAM_TCP_NODELAY",
default_value_t = true
)]
pub delegate_tcp_nodelay: bool,
pub upstream_tcp_nodelay: bool,
/// Log level (trace, debug, info, warn, error)
#[arg(long, env = "AETHER_PROXY_LOG_LEVEL", default_value = "info")]
@@ -192,25 +172,38 @@ pub struct Config {
#[arg(long, env = "AETHER_PROXY_LOG_JSON", default_value_t = false)]
pub log_json: bool,
/// Enable TLS encryption (dual-stack: accepts both HTTP and TLS on same port)
#[arg(long, env = "AETHER_PROXY_ENABLE_TLS", default_value_t = true)]
pub enable_tls: bool,
/// Path to TLS certificate PEM file
/// WebSocket reconnect base delay in milliseconds
#[arg(
long,
env = "AETHER_PROXY_TLS_CERT",
default_value = "aether-proxy-cert.pem"
env = "AETHER_PROXY_TUNNEL_RECONNECT_BASE_MS",
default_value_t = 1000
)]
pub tls_cert: String,
pub tunnel_reconnect_base_ms: u64,
/// Path to TLS private key PEM file
/// WebSocket reconnect max delay in milliseconds
#[arg(
long,
env = "AETHER_PROXY_TLS_KEY",
default_value = "aether-proxy-key.pem"
env = "AETHER_PROXY_TUNNEL_RECONNECT_MAX_MS",
default_value_t = 30000
)]
pub tls_key: String,
pub tunnel_reconnect_max_ms: u64,
/// WebSocket tunnel ping interval in seconds
#[arg(long, env = "AETHER_PROXY_TUNNEL_PING_INTERVAL", default_value_t = 15)]
pub tunnel_ping_interval_secs: u64,
/// Maximum concurrent streams over tunnel (auto-detected from hardware if omitted)
#[arg(long, env = "AETHER_PROXY_TUNNEL_MAX_STREAMS")]
pub tunnel_max_streams: Option<u32>,
}
/// Per-server connection config (used in multi-server TOML `[[servers]]`).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ServerEntry {
pub aether_url: String,
pub management_token: String,
/// Per-server node name override. Falls back to the global `node_name`.
pub node_name: Option<String>,
}
// ---------------------------------------------------------------------------
@@ -218,7 +211,7 @@ pub struct Config {
// ---------------------------------------------------------------------------
/// Serializable config for TOML file persistence.
/// All fields are optional — only populated values are written.
/// All fields are optional -- only populated values are written.
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct ConfigFile {
#[serde(skip_serializing_if = "Option::is_none")]
@@ -226,10 +219,6 @@ pub struct ConfigFile {
#[serde(skip_serializing_if = "Option::is_none")]
pub management_token: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub hmac_key: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub listen_port: Option<u16>,
#[serde(skip_serializing_if = "Option::is_none")]
pub public_ip: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub node_name: Option<String>,
@@ -240,8 +229,6 @@ pub struct ConfigFile {
#[serde(skip_serializing_if = "Option::is_none")]
pub allowed_ports: Option<Vec<u16>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub timestamp_tolerance: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub aether_request_timeout_secs: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub aether_connect_timeout_secs: Option<u64>,
@@ -264,33 +251,37 @@ pub struct ConfigFile {
#[serde(skip_serializing_if = "Option::is_none")]
pub max_concurrent_connections: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub connect_timeout_secs: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tls_handshake_timeout_secs: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub dns_cache_ttl_secs: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub dns_cache_capacity: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
pub delegate_connect_timeout_secs: Option<u64>,
pub upstream_connect_timeout_secs: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub delegate_pool_max_idle_per_host: Option<usize>,
pub upstream_pool_max_idle_per_host: Option<usize>,
#[serde(skip_serializing_if = "Option::is_none")]
pub delegate_pool_idle_timeout_secs: Option<u64>,
pub upstream_pool_idle_timeout_secs: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub delegate_tcp_keepalive_secs: Option<u64>,
pub upstream_tcp_keepalive_secs: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub delegate_tcp_nodelay: Option<bool>,
pub upstream_tcp_nodelay: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub log_level: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub log_json: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub enable_tls: Option<bool>,
pub tunnel_reconnect_base_ms: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tls_cert: Option<String>,
pub tunnel_reconnect_max_ms: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tls_key: Option<String>,
pub tunnel_ping_interval_secs: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tunnel_max_streams: Option<u32>,
/// Multi-server config: each entry connects to a separate Aether instance.
/// When present, top-level aether_url/management_token are ignored for
/// tunnel connections (but still injected as env for clap compatibility).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub servers: Vec<ServerEntry>,
}
impl ConfigFile {
@@ -307,6 +298,24 @@ impl ConfigFile {
Ok(())
}
/// Resolve the effective server list.
///
/// If `[[servers]]` is present, use it. Otherwise fall back to the
/// top-level `aether_url` + `management_token` as a single server.
pub fn effective_servers(&self) -> Vec<ServerEntry> {
if !self.servers.is_empty() {
return self.servers.clone();
}
match (&self.aether_url, &self.management_token) {
(Some(url), Some(token)) => vec![ServerEntry {
aether_url: url.clone(),
management_token: token.clone(),
node_name: None,
}],
_ => vec![],
}
}
/// Inject values as environment variables so clap picks them up.
///
/// Only sets variables that are **not** already present in the
@@ -332,15 +341,30 @@ impl ConfigFile {
}
};
}
set!("AETHER_PROXY_AETHER_URL", self.aether_url);
set!("AETHER_PROXY_MANAGEMENT_TOKEN", self.management_token);
set!("AETHER_PROXY_HMAC_KEY", self.hmac_key);
set!("AETHER_PROXY_LISTEN_PORT", self.listen_port);
// When top-level fields are absent, fall back to the first [[servers]]
// entry so that clap's required `aether_url` / `management_token` are
// satisfied even with the new config format.
let first_server = self.servers.first();
let aether_url = self
.aether_url
.clone()
.or_else(|| first_server.map(|s| s.aether_url.clone()));
let management_token = self
.management_token
.clone()
.or_else(|| first_server.map(|s| s.management_token.clone()));
let node_name = self
.node_name
.clone()
.or_else(|| first_server.and_then(|s| s.node_name.clone()));
set!("AETHER_PROXY_AETHER_URL", aether_url);
set!("AETHER_PROXY_MANAGEMENT_TOKEN", management_token);
set!("AETHER_PROXY_PUBLIC_IP", self.public_ip);
set!("AETHER_PROXY_NODE_NAME", self.node_name);
set!("AETHER_PROXY_NODE_NAME", node_name);
set!("AETHER_PROXY_NODE_REGION", self.node_region);
set!("AETHER_PROXY_HEARTBEAT_INTERVAL", self.heartbeat_interval);
set!("AETHER_PROXY_TIMESTAMP_TOLERANCE", self.timestamp_tolerance);
set!(
"AETHER_PROXY_AETHER_REQUEST_TIMEOUT",
self.aether_request_timeout_secs
@@ -379,38 +403,43 @@ impl ConfigFile {
"AETHER_PROXY_MAX_CONCURRENT_CONNECTIONS",
self.max_concurrent_connections
);
set!("AETHER_PROXY_CONNECT_TIMEOUT", self.connect_timeout_secs);
set!(
"AETHER_PROXY_TLS_HANDSHAKE_TIMEOUT",
self.tls_handshake_timeout_secs
);
set!("AETHER_PROXY_DNS_CACHE_TTL", self.dns_cache_ttl_secs);
set!("AETHER_PROXY_DNS_CACHE_CAPACITY", self.dns_cache_capacity);
set!(
"AETHER_PROXY_DELEGATE_CONNECT_TIMEOUT",
self.delegate_connect_timeout_secs
"AETHER_PROXY_UPSTREAM_CONNECT_TIMEOUT",
self.upstream_connect_timeout_secs
);
set!(
"AETHER_PROXY_DELEGATE_POOL_MAX_IDLE_PER_HOST",
self.delegate_pool_max_idle_per_host
"AETHER_PROXY_UPSTREAM_POOL_MAX_IDLE_PER_HOST",
self.upstream_pool_max_idle_per_host
);
set!(
"AETHER_PROXY_DELEGATE_POOL_IDLE_TIMEOUT",
self.delegate_pool_idle_timeout_secs
"AETHER_PROXY_UPSTREAM_POOL_IDLE_TIMEOUT",
self.upstream_pool_idle_timeout_secs
);
set!(
"AETHER_PROXY_DELEGATE_TCP_KEEPALIVE",
self.delegate_tcp_keepalive_secs
"AETHER_PROXY_UPSTREAM_TCP_KEEPALIVE",
self.upstream_tcp_keepalive_secs
);
set!(
"AETHER_PROXY_DELEGATE_TCP_NODELAY",
self.delegate_tcp_nodelay
"AETHER_PROXY_UPSTREAM_TCP_NODELAY",
self.upstream_tcp_nodelay
);
set!("AETHER_PROXY_LOG_LEVEL", self.log_level);
set!("AETHER_PROXY_LOG_JSON", self.log_json);
set!("AETHER_PROXY_ENABLE_TLS", self.enable_tls);
set!("AETHER_PROXY_TLS_CERT", self.tls_cert);
set!("AETHER_PROXY_TLS_KEY", self.tls_key);
set!(
"AETHER_PROXY_TUNNEL_RECONNECT_BASE_MS",
self.tunnel_reconnect_base_ms
);
set!(
"AETHER_PROXY_TUNNEL_RECONNECT_MAX_MS",
self.tunnel_reconnect_max_ms
);
set!(
"AETHER_PROXY_TUNNEL_PING_INTERVAL",
self.tunnel_ping_interval_secs
);
set!("AETHER_PROXY_TUNNEL_MAX_STREAMS", self.tunnel_max_streams);
// allowed_ports needs special handling (comma-separated)
if let Some(ref ports) = self.allowed_ports {
aether-proxy/src/main.rs
+26 -3
View File
@@ -1,13 +1,13 @@
mod app;
mod auth;
mod config;
mod hardware;
mod net;
mod proxy;
mod registration;
mod runtime;
mod setup;
mod state;
mod target_filter;
mod tunnel;
use std::path::PathBuf;
@@ -135,5 +135,28 @@ async fn run_proxy(config: Config) -> anyhow::Result<()> {
std::process::exit(1);
}
app::run(config).await
// Resolve server list: prefer [[servers]] from TOML, fall back to CLI/env single server.
let config_path =
std::env::var("AETHER_PROXY_CONFIG").unwrap_or_else(|_| DEFAULT_CONFIG.to_string());
let servers = if std::path::Path::new(&config_path).exists() {
config::ConfigFile::load(std::path::Path::new(&config_path))
.ok()
.map(|f| f.effective_servers())
.filter(|s| !s.is_empty())
.unwrap_or_else(|| {
vec![config::ServerEntry {
aether_url: config.aether_url.clone(),
management_token: config.management_token.clone(),
node_name: None,
}]
})
} else {
vec![config::ServerEntry {
aether_url: config.aether_url.clone(),
management_token: config.management_token.clone(),
node_name: None,
}]
};
app::run(config, servers).await
}
aether-proxy/src/proxy/connect.rs
-163
View File
@@ -1,163 +0,0 @@
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::{self, DnsCache};
/// Handle HTTP CONNECT tunnel requests.
///
/// Flow: validate auth -> check target filter -> TCP connect -> 200 -> bidirectional copy
pub async fn handle_connect(
req: Request<Incoming>,
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
.headers()
.get("proxy-authorization")
.and_then(|v| v.to_str().ok());
// HMAC authentication
if let Err(e) = auth::validate_proxy_auth(proxy_auth, &config, timestamp_tolerance) {
warn!(error = %e, "CONNECT auth failed");
return proxy_auth_required(&e.to_string());
}
// Parse target host:port from CONNECT URI
let authority = match req.uri().authority() {
Some(auth) => auth.clone(),
None => {
warn!("CONNECT request missing authority");
return bad_request("missing target authority");
}
};
let host = authority.host().to_string();
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, 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 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 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;
match tokio::io::copy_bidirectional(&mut upgraded, &mut target).await {
Ok((from_client, from_target)) => {
debug!(
target = %target_display,
from_client,
from_target,
"CONNECT tunnel closed"
);
}
Err(e) => {
debug!(target = %target_display, error = %e, "CONNECT tunnel error");
}
}
}
Ok(Err(e)) => {
warn!(target = %target_display, error = %e, "CONNECT upgrade failed");
}
Err(_) => {
warn!(target = %target_display, "CONNECT upgrade timeout");
}
}
});
Response::builder()
.status(200)
.body(http_body_util::Empty::new())
.unwrap()
}
fn proxy_auth_required(msg: &str) -> Response<http_body_util::Empty<bytes::Bytes>> {
Response::builder()
.status(407)
.header("Proxy-Authenticate", "HMAC-SHA256")
.header("Content-Length", "0")
.header("X-Error", msg)
.body(http_body_util::Empty::new())
.unwrap()
}
fn forbidden(msg: &str) -> Response<http_body_util::Empty<bytes::Bytes>> {
Response::builder()
.status(403)
.header("Content-Length", "0")
.header("X-Error", msg)
.body(http_body_util::Empty::new())
.unwrap()
}
fn bad_request(msg: &str) -> Response<http_body_util::Empty<bytes::Bytes>> {
Response::builder()
.status(400)
.header("Content-Length", "0")
.header("X-Error", msg)
.body(http_body_util::Empty::new())
.unwrap()
}
fn bad_gateway(msg: &str) -> Response<http_body_util::Empty<bytes::Bytes>> {
Response::builder()
.status(502)
.header("Content-Length", "0")
.header("X-Error", msg)
.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()
}
aether-proxy/src/proxy/delegate.rs
-389
View File
@@ -1,389 +0,0 @@
use std::collections::HashMap;
use std::collections::HashSet;
use std::error::Error as StdError;
use std::sync::Arc;
use std::time::Instant;
use futures_util::StreamExt;
use http_body_util::{BodyExt, Full, Limited, StreamBody};
use hyper::body::{Frame, Incoming};
use hyper::header::{HeaderName, HeaderValue};
use hyper::{Method, Request, Response, Uri};
use tracing::{debug, warn};
use url::Url;
use super::BoxBody;
use crate::auth;
use crate::config::Config;
use crate::proxy::delegate_client::{ConnectTiming, DelegateClient};
use crate::proxy::target_filter::{self, DnsCache};
/// Handle delegation requests: Aether sends a full request description,
/// and the proxy issues the actual upstream HTTP call using its own TLS stack.
///
/// 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(
req: Request<Incoming>,
config: Arc<Config>,
allowed_ports: &HashSet<u16>,
timestamp_tolerance: u64,
dns_cache: &DnsCache,
http_client: &DelegateClient,
) -> Response<BoxBody> {
let total_start = Instant::now();
// ── Auth ──
let auth_header = req
.headers()
.get("authorization")
.and_then(|v| v.to_str().ok());
if let Err(e) = auth::validate_proxy_auth(auth_header, &config, timestamp_tolerance) {
warn!(error = %e, "delegate auth failed");
return error_response(401, "authentication_failed", &e.to_string());
}
let auth_ms = total_start.elapsed().as_millis() as u64;
// ── Parse metadata from headers ──
let meta_start = Instant::now();
let method_str = match req
.headers()
.get("x-delegate-method")
.and_then(|v| v.to_str().ok())
{
Some(m) => m.to_string(),
None => {
warn!("delegate missing X-Delegate-Method");
return error_response(400, "bad_request", "missing X-Delegate-Method header");
}
};
let target_url = match req
.headers()
.get("x-delegate-url")
.and_then(|v| v.to_str().ok())
{
Some(u) => u.to_string(),
None => {
warn!("delegate missing X-Delegate-Url");
return error_response(400, "bad_request", "missing X-Delegate-Url header");
}
};
let upstream_headers: HashMap<String, String> = match req
.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 req_content_length: u64 = req
.headers()
.get("content-length")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.parse().ok())
.unwrap_or(0);
let meta_ms = meta_start.elapsed().as_millis() as u64;
// ── Target validation ──
let parsed_url = match Url::parse(&target_url) {
Ok(u) => u,
Err(e) => {
warn!(url = %target_url, error = %e, "delegate invalid target URL");
return error_response(400, "bad_request", &format!("invalid URL: {}", e));
}
};
let host = match parsed_url.host_str() {
Some(h) => h.to_string(),
None => {
warn!(url = %target_url, "delegate target URL missing host");
return error_response(400, "bad_request", "URL missing host");
}
};
let port = parsed_url.port_or_known_default().unwrap_or(443);
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 = %method_str, url = %target_url, is_gzip, "delegate request");
// ── Build upstream request ──
let method = match method_str.parse::<Method>() {
Ok(m) => m,
Err(e) => {
warn!(error = %e, method = %method_str, "delegate invalid HTTP method");
return error_response(400, "bad_request", &format!("invalid method: {}", e));
}
};
let uri = match target_url.parse::<Uri>() {
Ok(u) => u,
Err(e) => {
warn!(error = %e, url = %target_url, "delegate invalid target URI");
return error_response(400, "bad_request", &format!("invalid URL: {}", e));
}
};
// ── Stream body passthrough ──
// When body is gzip-compressed, forward it directly to upstream with
// Content-Encoding: gzip header — no collect/decompress needed.
// All major AI API providers (Anthropic, OpenAI, Google) accept gzip request bodies.
let wire_size: u64;
let upstream_body: BoxBody;
if is_gzip {
let body_stream =
http_body_util::BodyStream::new(req.into_body()).filter_map(|result| async {
match result {
Ok(frame) => frame.into_data().ok().map(|data| {
Ok::<_, Box<dyn std::error::Error + Send + Sync>>(Frame::data(data))
}),
Err(e) => Some(Err(Box::new(e) as Box<dyn std::error::Error + Send + Sync>)),
}
});
let stream_body = StreamBody::new(body_stream);
upstream_body = BodyExt::boxed(stream_body);
// wire_size will be reported from Content-Length if available, otherwise 0
wire_size = req_content_length;
} else {
// Non-gzip: read body into memory (legacy path)
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");
}
};
wire_size = body_bytes.len() as u64;
let body = Full::new(body_bytes)
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { match e {} })
.boxed();
upstream_body = body;
}
let mut upstream_req = Request::new(upstream_body);
*upstream_req.method_mut() = method;
*upstream_req.uri_mut() = uri;
{
let headers = upstream_req.headers_mut();
// Set headers (skip `host` — hyper sets it from the URI automatically,
// and a duplicate Host header can confuse certain upstreams)
for (name, value) in &upstream_headers {
if name.eq_ignore_ascii_case("host") {
continue;
}
let header_name = match HeaderName::from_bytes(name.as_bytes()) {
Ok(n) => n,
Err(_) => {
warn!(header = %name, "delegate invalid header name");
return error_response(400, "bad_request", "invalid header name");
}
};
let header_value = match HeaderValue::from_str(value) {
Ok(v) => v,
Err(_) => {
warn!(header = %name, "delegate invalid header value");
return error_response(400, "bad_request", "invalid header value");
}
};
headers.insert(header_name, header_value);
}
if is_gzip {
headers.insert(
hyper::header::CONTENT_ENCODING,
HeaderValue::from_static("gzip"),
);
}
}
// ── Send upstream request ──
// NOTE: We intentionally do NOT set a per-request timeout here.
// Connect timeout limits connection establishment; Aether controls
// first-byte / idle timeouts on its own side via asyncio.
let upstream_start = Instant::now();
let upstream_resp = match http_client.request(upstream_req).await {
Ok(resp) => resp,
Err(e) => {
warn!(url = %target_url, error = %e, "delegate upstream request failed");
let safe_detail = sanitize_upstream_error(&root_error_message(&e));
if is_timeout_error(&e) {
return error_response(504, "upstream_timeout", &safe_detail);
}
return error_response(502, "upstream_connection_failed", &safe_detail);
}
};
let ttfb_ms = upstream_start.elapsed().as_millis() as u64;
// ── Build response ──
let status = upstream_resp.status().as_u16();
let resp_headers = upstream_resp.headers().clone();
let (connect_ms, tls_ms) = upstream_resp
.extensions()
.get::<ConnectTiming>()
.map(|t| (t.connect_ms, t.tls_ms))
.unwrap_or((0, 0));
let upstream_processing_ms = ttfb_ms.saturating_sub(connect_ms.saturating_add(tls_ms));
let total_ms = total_start.elapsed().as_millis() as u64;
debug!(
url = %target_url,
status,
dns_ms,
connect_ms,
tls_ms,
ttfb_ms,
upstream_processing_ms,
total_ms,
wire_size,
is_gzip,
"delegate upstream response"
);
let timing = serde_json::json!({
"auth_ms": auth_ms,
"meta_ms": meta_ms,
"wire_size": wire_size,
"passthrough": is_gzip,
"dns_ms": dns_ms,
"connect_ms": connect_ms,
"tls_ms": tls_ms,
"ttfb_ms": ttfb_ms,
"upstream_ms": ttfb_ms,
"upstream_processing_ms": upstream_processing_ms,
"total_ms": total_ms,
});
let stream_body: BoxBody = upstream_resp
.into_body()
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { Box::new(e) })
.boxed();
let mut builder = Response::builder().status(status);
for (name, value) in resp_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()
})
}
fn root_error_message(err: &dyn StdError) -> String {
let mut current = err;
while let Some(source) = current.source() {
current = source;
}
current.to_string()
}
fn is_timeout_error(err: &(dyn StdError + 'static)) -> bool {
if err.is::<tokio::time::error::Elapsed>() {
return true;
}
if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
if io_err.kind() == std::io::ErrorKind::TimedOut {
return true;
}
}
if let Some(source) = err.source() {
// source() returns &(dyn Error + 'static), so this is safe
return is_timeout_error(source);
}
false
}
// ── Sanitisation ─────────────────────────────────────────────────────────────
/// Strip full URLs from error messages to prevent leaking upstream API keys,
/// paths, or query parameters in the delegate error response.
///
/// Replaces `https://api.example.com/v1/chat?key=xxx` with `api.example.com`.
fn sanitize_upstream_error(msg: &str) -> String {
// Simple regex-free approach: find "https://..." or "http://..." spans and
// replace them with just the host portion.
let mut result = msg.to_string();
for scheme in &["https://", "http://"] {
while let Some(start) = result.find(scheme) {
let after_scheme = start + scheme.len();
// Host ends at '/', '?', '#', ' ', or end of string
let host_end = result[after_scheme..]
.find(['/', '?', '#', ' '])
.map(|i| after_scheme + i)
.unwrap_or(result.len());
let host = &result[after_scheme..host_end];
result = format!("{}{}{}", &result[..start], host, &result[host_end..]);
}
}
result
}
// ── Error response helpers ───────────────────────────────────────────────────
fn error_response(status: u16, error: &str, detail: &str) -> Response<BoxBody> {
let body = serde_json::json!({
"error": error,
"detail": detail,
});
let body_bytes = bytes::Bytes::from(body.to_string());
Response::builder()
.status(status)
.header("Content-Type", "application/json")
.header("X-Delegate-Error", "true")
.body(
Full::new(body_bytes)
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { match e {} })
.boxed(),
)
.unwrap()
}
aether-proxy/src/proxy/delegate_client.rs
-282
View File
@@ -1,282 +0,0 @@
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use hyper::rt;
use hyper::Uri;
use hyper_util::client::legacy::connect::{Connected, Connection, HttpConnector};
use hyper_util::client::legacy::Client;
use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
use rustls::ClientConfig;
use rustls_pki_types::ServerName;
use tokio_rustls::TlsConnector;
use tower_service::Service;
use crate::config::Config;
use crate::proxy::BoxBody;
type BoxError = Box<dyn std::error::Error + Send + Sync>;
type DelegateStream = MaybeHttpsStream<TokioIo<tokio::net::TcpStream>>;
type DelegateConn = TimedConn<DelegateStream>;
pub(crate) type DelegateClient = Client<InstrumentedConnector, BoxBody>;
#[derive(Clone, Copy, Debug, Default)]
pub(crate) struct ConnectTiming {
pub connect_ms: u64,
pub tls_ms: u64,
}
pub(crate) fn build_delegate_client(config: &Config) -> DelegateClient {
let mut http = HttpConnector::new();
http.enforce_http(false);
http.set_connect_timeout(Some(Duration::from_secs(
config.delegate_connect_timeout_secs,
)));
http.set_nodelay(config.delegate_tcp_nodelay);
if config.delegate_tcp_keepalive_secs > 0 {
http.set_keepalive(Some(Duration::from_secs(
config.delegate_tcp_keepalive_secs,
)));
} else {
http.set_keepalive(None);
}
let connector = InstrumentedConnector {
http,
tls_config: build_tls_config(),
};
let mut builder = Client::builder(TokioExecutor::new());
builder.pool_max_idle_per_host(config.delegate_pool_max_idle_per_host);
builder.pool_idle_timeout(Duration::from_secs(config.delegate_pool_idle_timeout_secs));
builder.pool_timer(TokioTimer::new());
builder.build::<_, BoxBody>(connector)
}
fn build_tls_config() -> Arc<ClientConfig> {
let root_store =
rustls::RootCertStore::from_iter(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
let mut config = ClientConfig::builder()
.with_root_certificates(root_store)
.with_no_client_auth();
config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
Arc::new(config)
}
#[derive(Clone)]
pub(crate) struct InstrumentedConnector {
http: HttpConnector,
tls_config: Arc<ClientConfig>,
}
impl Service<Uri> for InstrumentedConnector {
type Response = DelegateConn;
type Error = BoxError;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, BoxError>> + Send>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.http.poll_ready(cx).map_err(Into::into)
}
fn call(&mut self, dst: Uri) -> Self::Future {
let scheme = dst.scheme_str().map(|s| s.to_ascii_lowercase());
let tls_config = self.tls_config.clone();
let connecting = self.http.call(dst.clone());
let connect_start = Instant::now();
Box::pin(async move {
match scheme.as_deref() {
Some("http") => {
let tcp = connecting.await.map_err(|e| Box::new(e) as BoxError)?;
let connect_ms = connect_start.elapsed().as_millis() as u64;
Ok(TimedConn::new(
MaybeHttpsStream::Http(tcp),
ConnectTiming {
connect_ms,
tls_ms: 0,
},
))
}
Some("https") => {
let server_name = resolve_server_name(&dst)?;
let tcp = connecting.await.map_err(|e| Box::new(e) as BoxError)?;
let connect_ms = connect_start.elapsed().as_millis() as u64;
let tls_start = Instant::now();
let tls_stream = TlsConnector::from(tls_config)
.connect(server_name, TokioIo::new(tcp))
.await
.map_err(std::io::Error::other)?;
let tls_ms = tls_start.elapsed().as_millis() as u64;
Ok(TimedConn::new(
MaybeHttpsStream::Https(TokioIo::new(tls_stream)),
ConnectTiming { connect_ms, tls_ms },
))
}
Some(other) => {
Err(std::io::Error::other(format!("unsupported scheme {other}")).into())
}
None => Err(std::io::Error::other("missing scheme").into()),
}
})
}
}
fn resolve_server_name(uri: &Uri) -> Result<ServerName<'static>, BoxError> {
let host = uri.host().ok_or("missing host")?;
let host = host.trim_start_matches('[').trim_end_matches(']');
Ok(ServerName::try_from(host.to_string())?)
}
pub(crate) struct TimedConn<T> {
inner: T,
timing: ConnectTiming,
}
impl<T> TimedConn<T> {
fn new(inner: T, timing: ConnectTiming) -> Self {
Self { inner, timing }
}
}
impl<T: Connection> Connection for TimedConn<T> {
fn connected(&self) -> Connected {
self.inner.connected().extra(self.timing)
}
}
impl<T: rt::Read + Unpin> rt::Read for TimedConn<T> {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: rt::ReadBufCursor<'_>,
) -> Poll<Result<(), std::io::Error>> {
Pin::new(&mut self.inner).poll_read(cx, buf)
}
}
impl<T: rt::Write + Unpin> rt::Write for TimedConn<T> {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<Result<usize, std::io::Error>> {
Pin::new(&mut self.inner).poll_write(cx, buf)
}
fn poll_flush(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), std::io::Error>> {
Pin::new(&mut self.inner).poll_flush(cx)
}
fn poll_shutdown(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), std::io::Error>> {
Pin::new(&mut self.inner).poll_shutdown(cx)
}
fn is_write_vectored(&self) -> bool {
self.inner.is_write_vectored()
}
fn poll_write_vectored(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
bufs: &[std::io::IoSlice<'_>],
) -> Poll<Result<usize, std::io::Error>> {
Pin::new(&mut self.inner).poll_write_vectored(cx, bufs)
}
}
#[allow(clippy::large_enum_variant)]
pub(crate) enum MaybeHttpsStream<T> {
Http(T),
Https(TokioIo<tokio_rustls::client::TlsStream<TokioIo<T>>>),
}
impl<T: rt::Read + rt::Write + Connection + Unpin> Connection for MaybeHttpsStream<T> {
fn connected(&self) -> Connected {
match self {
Self::Http(stream) => stream.connected(),
Self::Https(stream) => {
let (tcp, tls) = stream.inner().get_ref();
if tls.alpn_protocol() == Some(b"h2") {
tcp.inner().connected().negotiated_h2()
} else {
tcp.inner().connected()
}
}
}
}
}
impl<T: rt::Read + rt::Write + Unpin> rt::Read for MaybeHttpsStream<T> {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: rt::ReadBufCursor<'_>,
) -> Poll<Result<(), std::io::Error>> {
match Pin::get_mut(self) {
Self::Http(stream) => Pin::new(stream).poll_read(cx, buf),
Self::Https(stream) => Pin::new(stream).poll_read(cx, buf),
}
}
}
impl<T: rt::Write + rt::Read + Unpin> rt::Write for MaybeHttpsStream<T> {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<Result<usize, std::io::Error>> {
match Pin::get_mut(self) {
Self::Http(stream) => Pin::new(stream).poll_write(cx, buf),
Self::Https(stream) => Pin::new(stream).poll_write(cx, buf),
}
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
match Pin::get_mut(self) {
Self::Http(stream) => Pin::new(stream).poll_flush(cx),
Self::Https(stream) => Pin::new(stream).poll_flush(cx),
}
}
fn poll_shutdown(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), std::io::Error>> {
match Pin::get_mut(self) {
Self::Http(stream) => Pin::new(stream).poll_shutdown(cx),
Self::Https(stream) => Pin::new(stream).poll_shutdown(cx),
}
}
fn is_write_vectored(&self) -> bool {
match self {
Self::Http(stream) => stream.is_write_vectored(),
Self::Https(stream) => stream.is_write_vectored(),
}
}
fn poll_write_vectored(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
bufs: &[std::io::IoSlice<'_>],
) -> Poll<Result<usize, std::io::Error>> {
match Pin::get_mut(self) {
Self::Http(stream) => Pin::new(stream).poll_write_vectored(cx, bufs),
Self::Https(stream) => Pin::new(stream).poll_write_vectored(cx, bufs),
}
}
}
aether-proxy/src/proxy/mod.rs
-19
View File
@@ -1,19 +0,0 @@
pub mod connect;
pub mod delegate;
pub mod delegate_client;
pub mod server;
pub mod target_filter;
pub mod tls;
use http_body_util::BodyExt;
/// Boxed body type used across proxy handlers.
pub type BoxBody =
http_body_util::combinators::BoxBody<bytes::Bytes, Box<dyn std::error::Error + Send + Sync>>;
/// Create an empty [`BoxBody`] (for error responses, 405, etc.).
pub fn empty_box_body() -> BoxBody {
http_body_util::Full::new(bytes::Bytes::new())
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { match e {} })
.boxed()
}
aether-proxy/src/proxy/server.rs
-213
View File
@@ -1,213 +0,0 @@
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;
use hyper::rt::{Read, Write};
use hyper::server::conn::http1;
use hyper::service::service_fn;
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};
use crate::state::AppState;
/// Start the proxy server.
///
/// Listens for incoming TCP connections and dispatches:
/// - CONNECT requests -> tunnel handler
/// - POST /_aether/delegate -> delegate handler
/// - Other requests -> 405 Method Not Allowed
///
/// When TLS is configured, the server operates in dual-stack mode:
/// it peeks at the first byte of each connection to distinguish TLS ClientHello
/// (0x16) from plain HTTP, and handles both on the same port.
pub async fn run(
state: &Arc<AppState>,
mut shutdown_rx: watch::Receiver<bool>,
) -> anyhow::Result<()> {
let addr = SocketAddr::from(([0, 0, 0, 0], state.config.listen_port));
let listener = TcpListener::bind(addr).await?;
if state.tls_acceptor.is_some() {
info!(addr = %addr, "proxy server listening (HTTP+TLS dual-stack)");
} else {
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() => {
let (stream, peer_addr) = match result {
Ok(v) => v,
Err(e) => {
warn!(error = %e, "failed to accept connection");
continue;
}
};
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 {
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),
peer_addr,
&state,
)
.await;
}
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;
}
}
// Plain HTTP
serve_connection(
TokioIo::new(stream),
peer_addr,
&state,
)
.await;
state.active_connections.fetch_sub(1, Ordering::Relaxed);
});
}
_ = shutdown_rx.changed() => {
info!("proxy server shutting down");
break;
}
}
}
Ok(())
}
/// Serve a single HTTP/1.1 connection (works over both plain TCP and TLS).
async fn serve_connection<I>(io: I, peer_addr: SocketAddr, state: &Arc<AppState>)
where
I: Read + Write + Unpin + Send + 'static,
{
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();
(d.allowed_ports.clone(), d.timestamp_tolerance)
};
if req.method() == Method::CONNECT {
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
{
let resp = delegate::handle_delegate(
req,
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).
let resp = Response::builder()
.status(405)
.header("Allow", "CONNECT")
.header("Content-Length", "0")
.body(crate::proxy::empty_box_body())
.unwrap();
metrics.record_request(start.elapsed());
Ok(resp)
}
}
});
if let Err(e) = http1::Builder::new()
.preserve_header_case(true)
.title_case_headers(false)
.serve_connection(io, service)
.with_upgrades()
.await
{
if !e.to_string().contains("connection closed") {
debug!(peer = %peer_addr, error = %e, "connection error");
}
}
}
aether-proxy/src/proxy/tls.rs
-125
View File
@@ -1,125 +0,0 @@
use std::fs;
use std::io::BufReader;
use std::path::Path;
use std::sync::Arc;
use rcgen::{CertificateParams, KeyPair};
use rustls_pki_types::{CertificateDer, PrivateKeyDer};
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`.
/// The private key file is set to mode 0600 on unix.
pub fn ensure_self_signed_cert(cert_path: &Path, key_path: &Path) -> anyhow::Result<()> {
if cert_path.exists() && key_path.exists() {
info!(
cert = %cert_path.display(),
key = %key_path.display(),
"using existing TLS certificate"
);
return Ok(());
}
info!("generating self-signed TLS certificate");
let mut params = CertificateParams::new(vec!["localhost".into(), "aether-proxy".into()])?;
params.distinguished_name = rcgen::DistinguishedName::new();
params
.distinguished_name
.push(rcgen::DnType::CommonName, "aether-proxy");
let key_pair = KeyPair::generate()?;
let cert = params.self_signed(&key_pair)?;
let cert_pem = cert.pem();
let key_pem = key_pair.serialize_pem();
fs::write(cert_path, &cert_pem)?;
fs::write(key_path, &key_pem)?;
// Set key file permissions to 0600 on unix
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let perms = fs::Permissions::from_mode(0o600);
fs::set_permissions(key_path, perms)?;
}
info!(
cert = %cert_path.display(),
key = %key_path.display(),
"self-signed TLS certificate generated"
);
Ok(())
}
/// Build a `TlsAcceptor` from PEM certificate and key files.
pub fn build_tls_acceptor(cert_path: &Path, key_path: &Path) -> anyhow::Result<TlsAcceptor> {
let cert_file = fs::File::open(cert_path)?;
let key_file = fs::File::open(key_path)?;
let certs: Vec<CertificateDer<'static>> =
rustls_pemfile::certs(&mut BufReader::new(cert_file)).collect::<Result<Vec<_>, _>>()?;
if certs.is_empty() {
anyhow::bail!("no certificates found in {}", cert_path.display());
}
let key: PrivateKeyDer<'static> =
rustls_pemfile::private_key(&mut BufReader::new(key_file))?
.ok_or_else(|| anyhow::anyhow!("no private key found in {}", key_path.display()))?;
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)))
}
/// Compute the SHA-256 fingerprint of the first certificate in a PEM file.
///
/// Returns the hex-encoded fingerprint (lowercase, no separators).
pub fn cert_sha256_fingerprint(cert_path: &Path) -> anyhow::Result<String> {
let cert_file = fs::File::open(cert_path)?;
let certs: Vec<CertificateDer<'static>> =
rustls_pemfile::certs(&mut BufReader::new(cert_file)).collect::<Result<Vec<_>, _>>()?;
let cert = certs
.first()
.ok_or_else(|| anyhow::anyhow!("no certificates found in {}", cert_path.display()))?;
let digest = Sha256::digest(cert.as_ref());
Ok(hex::encode(digest))
}
/// Peek at the first byte of a TCP stream to determine if it is a TLS ClientHello.
///
/// Returns `true` if the first byte is 0x16 (TLS record type: Handshake).
pub async fn is_tls_client_hello(stream: &tokio::net::TcpStream) -> bool {
let mut buf = [0u8; 1];
match stream.peek(&mut buf).await {
Ok(1) => buf[0] == 0x16,
Ok(_) => false,
Err(e) => {
warn!(error = %e, "failed to peek first byte");
false
}
}
}
aether-proxy/src/registration/client.rs
+9 -142
View File
@@ -3,30 +3,11 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
use reqwest::{Client, StatusCode};
use serde::{Deserialize, Serialize};
use tokio::time::sleep;
use tracing::{debug, error, info, warn};
use tracing::{debug, error, info};
use crate::config::Config;
use crate::hardware::HardwareInfo;
/// Heartbeat-specific error that distinguishes "node not found" (needs
/// re-registration) from transient / other failures.
#[derive(Debug)]
pub enum HeartbeatError {
/// HTTP 404 – the node_id is no longer known to Aether.
NodeNotFound(String),
/// Any other failure (network, 5xx, etc.).
Other(anyhow::Error),
}
impl std::fmt::Display for HeartbeatError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::NodeNotFound(msg) => write!(f, "node not found: {}", msg),
Self::Other(e) => write!(f, "{}", e),
}
}
}
#[derive(Debug, Serialize)]
struct RegisterRequest {
name: String,
@@ -35,14 +16,11 @@ struct RegisterRequest {
#[serde(skip_serializing_if = "Option::is_none")]
region: Option<String>,
heartbeat_interval: u64,
#[serde(skip_serializing_if = "std::ops::Not::not")]
tls_enabled: bool,
#[serde(skip_serializing_if = "Option::is_none")]
tls_cert_fingerprint: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
hardware_info: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
estimated_max_concurrency: Option<u64>,
tunnel_mode: bool,
}
#[derive(Debug, Deserialize)]
@@ -50,17 +28,6 @@ pub struct RegisterResponse {
pub node_id: String,
}
#[derive(Debug, Serialize)]
struct HeartbeatRequest {
node_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
active_connections: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
total_requests: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
avg_latency_ms: Option<f64>,
}
/// Remote configuration pushed by the Aether management backend.
#[derive(Debug, Clone, Deserialize)]
pub struct RemoteConfig {
@@ -68,29 +35,6 @@ pub struct RemoteConfig {
pub allowed_ports: Option<Vec<u16>>,
pub log_level: Option<String>,
pub heartbeat_interval: Option<u64>,
pub timestamp_tolerance: Option<u64>,
}
/// Parsed heartbeat response from Aether.
#[derive(Debug, Deserialize)]
struct HeartbeatResponseBody {
#[serde(default)]
node: Option<HeartbeatNodeInfo>,
}
#[derive(Debug, Deserialize)]
struct HeartbeatNodeInfo {
#[serde(default)]
remote_config: Option<RemoteConfig>,
#[serde(default)]
config_version: Option<u64>,
}
/// Heartbeat result returned to the caller.
#[derive(Debug)]
pub struct HeartbeatResult {
pub remote_config: Option<RemoteConfig>,
pub config_version: u64,
}
#[derive(Debug, Serialize)]
@@ -109,7 +53,7 @@ pub struct AetherClient {
}
impl AetherClient {
pub fn new(config: &Config) -> Self {
pub fn new(config: &Config, aether_url: &str, management_token: &str) -> Self {
let mut builder = Client::builder()
.timeout(Duration::from_secs(config.aether_request_timeout_secs))
.connect_timeout(Duration::from_secs(config.aether_connect_timeout_secs))
@@ -136,8 +80,8 @@ impl AetherClient {
Self {
http,
base_url: config.aether_url.trim_end_matches('/').to_string(),
token: config.management_token.clone(),
base_url: aether_url.trim_end_matches('/').to_string(),
token: management_token.to_string(),
retry_max_attempts: config.aether_retry_max_attempts.max(1),
retry_base_delay,
retry_max_delay,
@@ -150,29 +94,26 @@ impl AetherClient {
pub async fn register(
&self,
config: &Config,
node_name: &str,
public_ip: &str,
tls_enabled: bool,
tls_cert_fingerprint: Option<&str>,
hw: Option<&HardwareInfo>,
) -> anyhow::Result<String> {
let url = format!("{}/api/admin/proxy-nodes/register", self.base_url);
let body = RegisterRequest {
name: config.node_name.clone(),
name: node_name.to_string(),
ip: public_ip.to_string(),
port: config.listen_port,
port: 0,
region: config.node_region.clone(),
heartbeat_interval: config.heartbeat_interval,
tls_enabled,
tls_cert_fingerprint: tls_cert_fingerprint.map(|s| s.to_string()),
hardware_info: hw.and_then(|h| serde_json::to_value(h).ok()),
estimated_max_concurrency: hw.map(|h| h.estimated_max_concurrency),
tunnel_mode: true,
};
info!(
url = %url,
name = %body.name,
ip = %body.ip,
port = body.port,
"registering with Aether"
);
@@ -199,80 +140,6 @@ impl AetherClient {
Ok(data.node_id)
}
/// Send heartbeat to Aether.
///
/// On success, returns any remote config included in the response.
/// Returns [`HeartbeatError::NodeNotFound`] on HTTP 404 so the caller
/// can trigger re-registration.
pub async fn heartbeat(
&self,
node_id: &str,
active_connections: Option<i64>,
total_requests: Option<i64>,
avg_latency_ms: Option<f64>,
) -> Result<HeartbeatResult, HeartbeatError> {
let url = format!("{}/api/admin/proxy-nodes/heartbeat", self.base_url);
let body = HeartbeatRequest {
node_id: node_id.to_string(),
active_connections,
total_requests,
avg_latency_ms,
};
debug!(node_id = %node_id, "sending heartbeat");
let resp = self
.send_with_retry(
|| {
self.http
.post(&url)
.header("Authorization", format!("Bearer {}", self.token))
.json(&body)
},
"heartbeat",
)
.await
.map_err(|e| HeartbeatError::Other(e.into()))?;
let status = resp.status();
if !status.is_success() {
let text = resp.text().await.unwrap_or_default();
warn!(status = %status, body = %text, "heartbeat failed");
if status == StatusCode::NOT_FOUND {
return Err(HeartbeatError::NodeNotFound(text));
}
return Err(HeartbeatError::Other(anyhow::anyhow!(
"heartbeat failed (HTTP {}): {}",
status,
text
)));
}
// Parse remote config from response (best-effort)
let result = match resp.json::<HeartbeatResponseBody>().await {
Ok(body) => {
let (remote_config, config_version) = match body.node {
Some(node) => (node.remote_config, node.config_version.unwrap_or(0)),
None => (None, 0),
};
HeartbeatResult {
remote_config,
config_version,
}
}
Err(e) => {
debug!(error = %e, "failed to parse heartbeat response body");
HeartbeatResult {
remote_config: None,
config_version: 0,
}
}
};
debug!(node_id = %node_id, config_version = result.config_version, "heartbeat ok");
Ok(result)
}
/// Unregister this node from Aether (graceful shutdown).
pub async fn unregister(&self, node_id: &str) -> anyhow::Result<()> {
let url = format!("{}/api/admin/proxy-nodes/unregister", self.base_url);
aether-proxy/src/registration/heartbeat.rs
-127
View File
@@ -1,127 +0,0 @@
use std::sync::atomic::Ordering;
use std::sync::Arc;
use tokio::sync::watch;
use tracing::{debug, error, info, warn};
use crate::registration::client::HeartbeatError;
use crate::runtime;
use crate::state::AppState;
/// Run periodic heartbeat task until shutdown signal.
///
/// When Aether responds with 404 (node not found), this task automatically
/// re-registers the node and updates the shared `node_id` so the proxy
/// server and future heartbeats use the new identity.
///
/// When the heartbeat response includes a `remote_config`, it is applied
/// to the [`DynamicConfig`](crate::runtime::DynamicConfig) so the proxy
/// picks up changes without a restart.
pub async fn run(state: &Arc<AppState>, mut shutdown_rx: watch::Receiver<bool>) {
let mut consecutive_failures: u32 = 0;
// Skip the first tick (registration already acts as initial heartbeat)
let initial_interval = state.dynamic.read().unwrap().heartbeat_interval;
tokio::select! {
_ = tokio::time::sleep(std::time::Duration::from_secs(initial_interval)) => {}
_ = shutdown_rx.changed() => {
debug!("heartbeat task stopping (during initial wait)");
return;
}
}
loop {
let current_node_id = state.node_id.read().unwrap().clone();
let active_conns = state.active_connections.load(Ordering::Relaxed) as i64;
// Swap-and-reset: report incremental metrics since last heartbeat
let interval_requests = state.metrics.total_requests.swap(0, Ordering::Relaxed);
let interval_latency_ns = state.metrics.total_latency_ns.swap(0, Ordering::Relaxed);
let interval_requests_i64 = i64::try_from(interval_requests).unwrap_or(i64::MAX);
let avg_latency_ms = if interval_requests > 0 {
Some(interval_latency_ns as f64 / interval_requests as f64 / 1_000_000.0)
} else {
None
};
match state
.aether_client
.heartbeat(
&current_node_id,
Some(active_conns),
Some(interval_requests_i64),
avg_latency_ms,
)
.await
{
Ok(result) => {
if consecutive_failures > 0 {
info!(
previous_failures = consecutive_failures,
"heartbeat recovered"
);
}
consecutive_failures = 0;
// Apply remote config if present and version changed
if let Some(ref remote) = result.remote_config {
runtime::apply_remote_config(&state.dynamic, remote, result.config_version);
}
}
Err(HeartbeatError::NodeNotFound(_)) => {
warn!(
old_node_id = %current_node_id,
"node not found, re-registering"
);
match state
.aether_client
.register(
&state.config,
&state.public_ip,
state.config.enable_tls,
state.tls_fingerprint.as_deref(),
Some(&state.hardware_info),
)
.await
{
Ok(new_id) => {
info!(
old_node_id = %current_node_id,
new_node_id = %new_id,
"re-registered successfully"
);
*state.node_id.write().unwrap() = new_id;
consecutive_failures = 0;
}
Err(e) => {
consecutive_failures += 1;
error!(
error = %e,
consecutive_failures,
"re-registration failed"
);
}
}
}
Err(HeartbeatError::Other(e)) => {
consecutive_failures += 1;
warn!(
error = %e,
consecutive_failures,
"heartbeat failed"
);
}
}
// Read interval from dynamic config (may have been updated remotely)
let interval_secs = state.dynamic.read().unwrap().heartbeat_interval;
tokio::select! {
_ = tokio::time::sleep(std::time::Duration::from_secs(interval_secs)) => {}
_ = shutdown_rx.changed() => {
debug!("heartbeat task stopping");
break;
}
}
}
}
aether-proxy/src/registration/mod.rs
-1
View File
@@ -1,2 +1 @@
pub mod client;
pub mod heartbeat;
aether-proxy/src/runtime.rs
+1 -10
View File
@@ -16,7 +16,6 @@ use crate::config::Config;
pub struct DynamicConfig {
pub node_name: String,
pub allowed_ports: HashSet<u16>,
pub timestamp_tolerance: u64,
pub log_level: String,
pub heartbeat_interval: u64,
/// Monotonically increasing version from the backend.
@@ -30,7 +29,6 @@ impl DynamicConfig {
Self {
node_name: config.node_name.clone(),
allowed_ports: config.allowed_ports.iter().copied().collect(),
timestamp_tolerance: config.timestamp_tolerance,
log_level: config.log_level.clone(),
heartbeat_interval: config.heartbeat_interval,
config_version: 0,
@@ -79,18 +77,11 @@ pub fn apply_remote_config(
if let Some(ref ports) = remote.allowed_ports {
let new_set: HashSet<u16> = ports.iter().copied().collect();
if new_set != cfg.allowed_ports {
changed.push(format!("allowed_ports → {:?}", ports));
changed.push(format!("allowed_ports -> {:?}", ports));
cfg.allowed_ports = new_set;
}
}
if let Some(tol) = remote.timestamp_tolerance {
if tol != cfg.timestamp_tolerance {
changed.push(format!("timestamp_tolerance → {}", tol));
cfg.timestamp_tolerance = tol;
}
}
if let Some(interval) = remote.heartbeat_interval {
if interval != cfg.heartbeat_interval {
changed.push(format!("heartbeat_interval → {}s", interval));
aether-proxy/src/setup/tui.rs
+366 -200
View File
@@ -2,7 +2,8 @@
//!
//! Launched via `aether-proxy setup [path]`. Presents a full-screen form
//! backed by ratatui where the user can navigate fields, edit values, and
//! save to a TOML config file.
//! save to a TOML config file. Supports multi-server configuration via
//! a tabbed interface.
use std::io;
use std::path::PathBuf;
@@ -19,13 +20,13 @@ use ratatui::widgets::{Block, Borders, Paragraph};
use ratatui::Frame;
use ratatui::Terminal;
use crate::config::ConfigFile;
use crate::config::{ConfigFile, ServerEntry};
/// Outcome of the setup wizard, returned to the caller.
pub enum SetupOutcome {
/// Config saved; systemd service installed and started.
ServiceInstalled,
/// Config saved; no service — caller should start the proxy directly.
/// Config saved; no service -- caller should start the proxy directly.
ReadyToRun(PathBuf),
/// User quit without saving.
Cancelled,
@@ -34,13 +35,12 @@ pub enum SetupOutcome {
/// Column width reserved for the field label (chars).
const LABEL_WIDTH: usize = 22;
// ── Field types ──────────────────────────────────────────────────────────────
// -- Field types --------------------------------------------------------------
#[derive(Clone, Copy, PartialEq)]
enum FieldKind {
Text,
Secret,
Number,
Bool,
LogLevel,
}
@@ -53,31 +53,15 @@ struct Field {
required: bool,
help: &'static str,
}
// -- Server tab ---------------------------------------------------------------
// ── App state ────────────────────────────────────────────────────────────────
#[derive(PartialEq)]
enum Mode {
Normal,
Editing,
}
struct App {
/// A single server tab's editable fields.
struct ServerTab {
fields: Vec<Field>,
selected: usize,
mode: Mode,
edit_buffer: String,
edit_cursor: usize, // char index
config_path: PathBuf,
modified: bool,
message: Option<(String, Instant, bool)>, // (text, when, is_error)
scroll_offset: usize,
saved_once: bool,
pending_quit: bool, // true after first q/Esc with unsaved changes
}
impl App {
fn new(config_path: PathBuf) -> Self {
impl ServerTab {
fn new() -> Self {
Self {
fields: vec![
Field {
@@ -86,7 +70,7 @@ impl App {
value: String::new(),
kind: FieldKind::Text,
required: true,
help: "Aether 服务器 URL (如 https://aether.example.com)",
help: "Aether URL (e.g. https://aether.example.com)",
},
Field {
label: "Management Token",
@@ -94,23 +78,7 @@ impl App {
value: String::new(),
kind: FieldKind::Secret,
required: true,
help: "Aether 管理 API Token (ae_xxx)",
},
Field {
label: "HMAC Key",
key: "hmac_key",
value: String::new(),
kind: FieldKind::Secret,
required: true,
help: "HMAC-SHA256 签名密钥，用于代理请求认证",
},
Field {
label: "Listen Port",
key: "listen_port",
value: "18080".into(),
kind: FieldKind::Number,
required: true,
help: "代理服务监听端口",
help: "Aether Management Token (ae_xxx)",
},
Field {
label: "Node Name",
@@ -118,15 +86,60 @@ impl App {
value: "proxy-01".into(),
kind: FieldKind::Text,
required: true,
help: "节点名称，用于在 Aether 后台识别",
help: "Node name for identification in Aether dashboard",
},
],
}
}
fn from_entry(entry: &ServerEntry) -> Self {
let mut tab = Self::new();
tab.fields[0].value = entry.aether_url.clone();
tab.fields[1].value = entry.management_token.clone();
if let Some(ref name) = entry.node_name {
tab.fields[2].value = name.clone();
}
tab
}
}
// -- App state ----------------------------------------------------------------
#[derive(PartialEq)]
enum Mode {
Normal,
Editing,
}
struct App {
server_tabs: Vec<ServerTab>,
active_tab: usize,
global_fields: Vec<Field>,
selected: usize,
mode: Mode,
edit_buffer: String,
edit_cursor: usize,
config_path: PathBuf,
modified: bool,
message: Option<(String, Instant, bool)>,
scroll_offset: usize,
saved_once: bool,
pending_quit: bool,
confirm_delete: bool,
}
impl App {
fn new(config_path: PathBuf) -> Self {
Self {
server_tabs: vec![ServerTab::new()],
active_tab: 0,
global_fields: vec![
Field {
label: "Log Level",
key: "log_level",
value: "info".into(),
kind: FieldKind::LogLevel,
required: true,
help: "日志级别 -- Enter 切换: trace / debug / info / warn / error",
help: "Log level -- Enter to cycle: trace / debug / info / warn / error",
},
Field {
label: "Log JSON",
@@ -134,7 +147,7 @@ impl App {
value: "false".into(),
kind: FieldKind::Bool,
required: true,
help: "是否以 JSON 格式输出日志 -- Enter 切换",
help: "Output logs as JSON -- Enter to toggle",
},
Field {
label: "Install Service",
@@ -147,7 +160,7 @@ impl App {
.into(),
kind: FieldKind::Bool,
required: true,
help: "注册为 systemd 开机启动服务 (需要 root 权限) -- Enter 切换",
help: "Install as systemd service (requires root) -- Enter to toggle",
},
],
selected: 0,
@@ -160,10 +173,47 @@ impl App {
scroll_offset: 0,
saved_once: false,
pending_quit: false,
confirm_delete: false,
}
}
// ── Config ↔ fields ──────────────────────────────────────────────────
// -- Field accessors (unified index across server + global) ---------------
fn server_field_count(&self) -> usize {
self.server_tabs[self.active_tab].fields.len()
}
fn total_field_count(&self) -> usize {
self.server_field_count() + self.global_fields.len()
}
fn selected_field(&self) -> &Field {
let sc = self.server_field_count();
if self.selected < sc {
&self.server_tabs[self.active_tab].fields[self.selected]
} else {
&self.global_fields[self.selected - sc]
}
}
fn selected_field_mut(&mut self) -> &mut Field {
let sc = self.server_field_count();
if self.selected < sc {
&mut self.server_tabs[self.active_tab].fields[self.selected]
} else {
&mut self.global_fields[self.selected - sc]
}
}
fn clamp_selection(&mut self) {
let max = self.total_field_count();
if self.selected >= max {
self.selected = max.saturating_sub(1);
}
self.scroll_offset = 0;
self.confirm_delete = false;
}
// -- Config <-> fields -----------------------------------------------------
fn load_from_file(&mut self) {
if let Ok(cfg) = ConfigFile::load(&self.config_path) {
@@ -172,13 +222,9 @@ impl App {
}
fn apply_config(&mut self, cfg: &ConfigFile) {
for field in &mut self.fields {
// Global fields
for field in &mut self.global_fields {
let val: Option<String> = match field.key {
"aether_url" => cfg.aether_url.clone(),
"management_token" => cfg.management_token.clone(),
"hmac_key" => cfg.hmac_key.clone(),
"listen_port" => cfg.listen_port.map(|v| v.to_string()),
"node_name" => cfg.node_name.clone(),
"log_level" => cfg.log_level.clone(),
"log_json" => cfg.log_json.map(|v| v.to_string()),
_ => None,
@@ -187,54 +233,64 @@ impl App {
field.value = v;
}
}
// Server tabs
let servers = cfg.effective_servers();
if servers.is_empty() {
let mut tab = ServerTab::new();
// Single-server fallback: use top-level node_name
if let Some(ref name) = cfg.node_name {
tab.fields[2].value = name.clone();
}
self.server_tabs = vec![tab];
} else {
self.server_tabs = servers.iter().map(ServerTab::from_entry).collect();
// For single-server mode, node_name might be in top-level only
if self.server_tabs.len() == 1 && self.server_tabs[0].fields[2].value.is_empty() {
if let Some(ref name) = cfg.node_name {
self.server_tabs[0].fields[2].value = name.clone();
}
}
}
self.active_tab = 0;
self.selected = 0;
self.scroll_offset = 0;
}
fn to_config(&self) -> ConfigFile {
let get = |key: &str| -> Option<String> {
self.fields
let get_global = |key: &str| -> Option<String> {
self.global_fields
.iter()
.find(|f| f.key == key)
.map(|f| f.value.clone())
.filter(|v| !v.is_empty())
};
ConfigFile {
aether_url: get("aether_url"),
management_token: get("management_token"),
hmac_key: get("hmac_key"),
listen_port: get("listen_port").and_then(|v| v.parse().ok()),
public_ip: None,
node_name: get("node_name"),
node_region: None,
heartbeat_interval: None,
allowed_ports: None,
timestamp_tolerance: None,
aether_request_timeout_secs: None,
aether_connect_timeout_secs: None,
aether_pool_max_idle_per_host: None,
aether_pool_idle_timeout_secs: None,
aether_tcp_keepalive_secs: None,
aether_tcp_nodelay: None,
aether_http2: None,
aether_retry_max_attempts: None,
aether_retry_base_delay_ms: None,
aether_retry_max_delay_ms: None,
max_concurrent_connections: None,
connect_timeout_secs: None,
tls_handshake_timeout_secs: None,
dns_cache_ttl_secs: None,
dns_cache_capacity: None,
delegate_connect_timeout_secs: None,
delegate_pool_max_idle_per_host: None,
delegate_pool_idle_timeout_secs: None,
delegate_tcp_keepalive_secs: None,
delegate_tcp_nodelay: None,
log_level: get("log_level"),
log_json: get("log_json").and_then(|v| v.parse().ok()),
enable_tls: None,
tls_cert: None,
tls_key: None,
}
let get_tab = |tab: &ServerTab, key: &str| -> Option<String> {
tab.fields
.iter()
.find(|f| f.key == key)
.map(|f| f.value.clone())
.filter(|v| !v.is_empty())
};
let mut cfg = ConfigFile {
log_level: get_global("log_level"),
log_json: get_global("log_json").and_then(|v| v.parse().ok()),
..ConfigFile::default()
};
// Always write [[servers]] format; old top-level fields are read-only compat
cfg.servers = self
.server_tabs
.iter()
.map(|tab| ServerEntry {
aether_url: get_tab(tab, "aether_url").unwrap_or_default(),
management_token: get_tab(tab, "management_token").unwrap_or_default(),
node_name: get_tab(tab, "node_name"),
})
.collect();
cfg
}
fn save(&mut self) -> anyhow::Result<()> {
@@ -249,27 +305,33 @@ impl App {
));
Ok(())
}
// ── Scrolling ────────────────────────────────────────────────────────
// -- Scrolling ---------------------------------------------------------------
fn ensure_visible(&mut self, visible_rows: usize) {
if visible_rows == 0 {
return;
}
if self.selected < self.scroll_offset {
self.scroll_offset = self.selected;
} else if self.selected >= self.scroll_offset + visible_rows {
self.scroll_offset = self.selected - visible_rows + 1;
// Account for separator line between server and global fields
let display_row = if self.selected >= self.server_field_count() {
self.selected + 1
} else {
self.selected
};
if display_row < self.scroll_offset {
self.scroll_offset = display_row;
} else if display_row >= self.scroll_offset + visible_rows {
self.scroll_offset = display_row - visible_rows + 1;
}
}
// ── Key handling ─────────────────────────────────────────────────────
// -- Key handling -------------------------------------------------------------
/// Returns `true` when the app should exit.
fn handle_key(&mut self, key: KeyEvent) -> bool {
// Expire old messages (but keep quit-confirmation messages alive)
if let Some((_, when, _)) = &self.message {
if !self.pending_quit && when.elapsed() > Duration::from_secs(4) {
if !self.pending_quit && !self.confirm_delete && when.elapsed() > Duration::from_secs(4)
{
self.message = None;
}
}
@@ -284,7 +346,7 @@ impl App {
}
fn handle_normal(&mut self, key: KeyEvent) -> bool {
// ── Quit handling (with unsaved-changes confirmation) ─────────
// -- Quit handling (with unsaved-changes confirmation) -----------------
let is_quit_key = matches!(key.code, KeyCode::Char('q') | KeyCode::Esc);
if is_quit_key {
@@ -292,6 +354,7 @@ impl App {
return true;
}
self.pending_quit = true;
self.confirm_delete = false;
self.message = Some((
"unsaved changes! q again to discard, ^S to save".into(),
Instant::now(),
@@ -300,14 +363,21 @@ impl App {
return false;
}
// Any other key cancels the pending quit
// Any other key cancels pending quit / pending delete
if self.pending_quit {
self.pending_quit = false;
self.message = None;
}
if self.confirm_delete && !matches!(key.code, KeyCode::Delete | KeyCode::Char('x')) {
self.confirm_delete = false;
self.message = None;
}
match key.code {
KeyCode::Char('s') if key.modifiers.contains(KeyModifiers::CONTROL) => {
KeyCode::Char('s')
if key.modifiers.contains(KeyModifiers::CONTROL)
|| key.modifiers.contains(KeyModifiers::SUPER) =>
{
if let Err(e) = self.save() {
self.message = Some((format!("error: {}", e), Instant::now(), true));
}
@@ -316,23 +386,20 @@ impl App {
self.selected = self.selected.saturating_sub(1);
}
KeyCode::Down | KeyCode::Char('j') => {
if self.selected + 1 < self.fields.len() {
if self.selected + 1 < self.total_field_count() {
self.selected += 1;
}
}
KeyCode::Home => self.selected = 0,
KeyCode::End => self.selected = self.fields.len() - 1,
KeyCode::End => self.selected = self.total_field_count() - 1,
KeyCode::Enter | KeyCode::Char(' ') => {
let field = &self.fields[self.selected];
match field.kind {
let kind = self.selected_field().kind;
let key_str = self.selected_field().key;
let value = self.selected_field().value.clone();
match kind {
FieldKind::Bool => {
let toggled = if field.value == "true" {
"false"
} else {
"true"
};
// Block enabling service install without root/systemd
if field.key == "install_service"
let toggled = if value == "true" { "false" } else { "true" };
if key_str == "install_service"
&& toggled == "true"
&& !super::service::is_available()
{
@@ -342,27 +409,79 @@ impl App {
true,
));
} else {
self.fields[self.selected].value = toggled.into();
self.selected_field_mut().value = toggled.into();
self.modified = true;
}
}
FieldKind::LogLevel => {
const LEVELS: &[&str] = &["trace", "debug", "info", "warn", "error"];
let idx = LEVELS.iter().position(|l| *l == field.value).unwrap_or(2);
self.fields[self.selected].value = LEVELS[(idx + 1) % LEVELS.len()].into();
let idx = LEVELS.iter().position(|l| *l == value).unwrap_or(2);
self.selected_field_mut().value = LEVELS[(idx + 1) % LEVELS.len()].into();
self.modified = true;
}
_ => {
self.edit_buffer = field.value.clone();
self.edit_buffer = value;
self.edit_cursor = self.edit_buffer.chars().count();
self.mode = Mode::Editing;
}
}
}
// -- Tab navigation --
KeyCode::Tab => {
// Quick save shortcut
if let Err(e) = self.save() {
self.message = Some((format!("error: {}", e), Instant::now(), true));
if self.server_tabs.len() > 1 {
self.active_tab = (self.active_tab + 1) % self.server_tabs.len();
self.clamp_selection();
}
}
KeyCode::BackTab => {
if self.server_tabs.len() > 1 {
self.active_tab = if self.active_tab == 0 {
self.server_tabs.len() - 1
} else {
self.active_tab - 1
};
self.clamp_selection();
}
}
KeyCode::Char(c @ '1'..='9') if !key.modifiers.contains(KeyModifiers::CONTROL) => {
let idx = (c as usize) - ('1' as usize);
if idx < self.server_tabs.len() && idx != self.active_tab {
self.active_tab = idx;
self.clamp_selection();
}
}
// -- Add / remove server --
KeyCode::Char('+') | KeyCode::Char('a') => {
self.server_tabs.push(ServerTab::new());
self.active_tab = self.server_tabs.len() - 1;
self.selected = 0;
self.scroll_offset = 0;
self.modified = true;
self.message = Some((
format!("added server {}", self.server_tabs.len()),
Instant::now(),
false,
));
}
KeyCode::Delete | KeyCode::Char('x') => {
if self.server_tabs.len() <= 1 {
self.message =
Some(("cannot remove the last server".into(), Instant::now(), true));
} else if self.confirm_delete {
let removed = self.active_tab + 1;
self.server_tabs.remove(self.active_tab);
self.active_tab = self.active_tab.min(self.server_tabs.len() - 1);
self.clamp_selection();
self.modified = true;
self.message =
Some((format!("server {} removed", removed), Instant::now(), false));
} else {
self.confirm_delete = true;
self.message = Some((
"press Delete/x again to remove this server".into(),
Instant::now(),
true,
));
}
}
_ => {}
@@ -373,12 +492,11 @@ impl App {
fn handle_edit(&mut self, key: KeyEvent) {
match key.code {
KeyCode::Esc => {
// Cancel -- discard changes to this field
self.mode = Mode::Normal;
}
KeyCode::Enter => {
if self.validate_edit() {
self.fields[self.selected].value = self.edit_buffer.clone();
self.selected_field_mut().value = self.edit_buffer.clone();
self.modified = true;
self.mode = Mode::Normal;
} else {
@@ -419,12 +537,7 @@ impl App {
}
fn validate_edit(&self) -> bool {
let kind = self.fields[self.selected].kind;
let buf = &self.edit_buffer;
match kind {
FieldKind::Number => buf.is_empty() || buf.parse::<u64>().is_ok(),
_ => true,
}
true
}
/// Byte offset of the char at `char_idx`.
@@ -436,13 +549,11 @@ impl App {
.unwrap_or(self.edit_buffer.len())
}
}
// ── Rendering ────────────────────────────────────────────────────────────────
// -- Rendering ----------------------------------------------------------------
fn ui(f: &mut Frame, app: &mut App) {
let area = f.area();
// Outer block
let title = if app.modified {
" Aether Proxy Setup [*] "
} else {
@@ -458,53 +569,52 @@ fn ui(f: &mut Frame, app: &mut App) {
let inner = outer.inner(area);
f.render_widget(outer, area);
// Split: fields | footer
let chunks = Layout::vertical([Constraint::Min(1), Constraint::Length(4)]).split(inner);
// Split: fields | tab bar | footer
let chunks = Layout::vertical([
Constraint::Min(1),
Constraint::Length(1),
Constraint::Length(4),
])
.split(inner);
let fields_area = chunks[0];
let footer_area = chunks[1];
render_fields(f, app, fields_area);
render_footer(f, app, footer_area);
render_fields(f, app, chunks[0]);
render_tab_bar(f, app, chunks[1]);
render_footer(f, app, chunks[2]);
}
fn render_fields(f: &mut Frame, app: &mut App, area: Rect) {
let visible = area.height as usize;
app.ensure_visible(visible);
let server_count = app.server_field_count();
let mut lines: Vec<Line> = Vec::new();
// display_row tracks the actual row index (including separator)
let mut display_row: usize = 0;
for (i, field) in app.fields.iter().enumerate() {
if i < app.scroll_offset || i >= app.scroll_offset + visible {
continue;
// Server fields
for i in 0..server_count {
if display_row >= app.scroll_offset && display_row < app.scroll_offset + visible {
lines.push(build_field_line(app, i, display_row));
}
display_row += 1;
}
let selected = i == app.selected;
let indicator = if selected { " > " } else { " " };
// Separator line
if display_row >= app.scroll_offset && display_row < app.scroll_offset + visible {
lines.push(Line::from(Span::styled(
" ----------------------------------------",
Style::default().fg(Color::DarkGray),
)));
}
display_row += 1;
let label_style = if selected {
Style::default()
.fg(Color::Cyan)
.add_modifier(Modifier::BOLD)
} else {
Style::default().fg(Color::DarkGray)
};
let padded_label = format!("{:<width$}", field.label, width = LABEL_WIDTH);
// Value display
let (value_text, value_style) = if app.mode == Mode::Editing && selected {
(app.edit_buffer.clone(), Style::default().fg(Color::Yellow))
} else {
field_display(field)
};
lines.push(Line::from(vec![
Span::styled(indicator, label_style),
Span::styled(padded_label, label_style),
Span::raw(" "),
Span::styled(value_text, value_style),
]));
// Global fields
for i in 0..app.global_fields.len() {
let field_idx = server_count + i;
if display_row >= app.scroll_offset && display_row < app.scroll_offset + visible {
lines.push(build_field_line(app, field_idx, display_row));
}
display_row += 1;
}
let paragraph = Paragraph::new(lines);
@@ -512,8 +622,12 @@ fn render_fields(f: &mut Frame, app: &mut App, area: Rect) {
// Cursor position while editing
if app.mode == Mode::Editing {
let row_in_view = app.selected - app.scroll_offset;
// prefix: 3 (indicator) + LABEL_WIDTH + 2 (gap) = 27
let sel_display_row = if app.selected >= server_count {
app.selected + 1
} else {
app.selected
};
let row_in_view = sel_display_row.saturating_sub(app.scroll_offset);
let prefix: u16 = 3 + LABEL_WIDTH as u16 + 2;
let cx = area.x + prefix + app.edit_cursor as u16;
let cy = area.y + row_in_view as u16;
@@ -522,6 +636,40 @@ fn render_fields(f: &mut Frame, app: &mut App, area: Rect) {
}
}
}
fn build_field_line(app: &App, field_idx: usize, _display_row: usize) -> Line<'static> {
let sc = app.server_field_count();
let field = if field_idx < sc {
&app.server_tabs[app.active_tab].fields[field_idx]
} else {
&app.global_fields[field_idx - sc]
};
let selected = field_idx == app.selected;
let indicator = if selected { " > " } else { " " };
let label_style = if selected {
Style::default()
.fg(Color::Cyan)
.add_modifier(Modifier::BOLD)
} else {
Style::default().fg(Color::DarkGray)
};
let padded_label = format!("{:<width$}", field.label, width = LABEL_WIDTH);
let (value_text, value_style) = if app.mode == Mode::Editing && selected {
(app.edit_buffer.clone(), Style::default().fg(Color::Yellow))
} else {
field_display(field)
};
Line::from(vec![
Span::styled(indicator.to_string(), label_style),
Span::styled(padded_label, label_style),
Span::raw(" "),
Span::styled(value_text, value_style),
])
}
/// Returns (display_text, style) for a field in normal mode.
fn field_display(field: &Field) -> (String, Style) {
@@ -565,18 +713,54 @@ fn field_display(field: &Field) -> (String, Style) {
_ => (field.value.clone(), Style::default().fg(Color::White)),
}
}
fn render_tab_bar(f: &mut Frame, app: &App, area: Rect) {
let mut spans: Vec<Span> = Vec::new();
spans.push(Span::raw(" "));
for (i, tab) in app.server_tabs.iter().enumerate() {
let num = i + 1;
let name = tab
.fields
.iter()
.find(|f| f.key == "node_name")
.filter(|f| !f.value.is_empty())
.map(|f| f.value.clone())
.unwrap_or_else(|| format!("Server {}", num));
let label = format!(" {} {} ", num, name);
if i == app.active_tab {
spans.push(Span::styled(
label,
Style::default()
.fg(Color::Black)
.bg(Color::Cyan)
.add_modifier(Modifier::BOLD),
));
} else {
spans.push(Span::styled(label, Style::default().fg(Color::DarkGray)));
}
spans.push(Span::raw(" "));
}
spans.push(Span::styled(" + Add ", Style::default().fg(Color::Green)));
f.render_widget(Paragraph::new(Line::from(spans)), area);
}
fn render_footer(f: &mut Frame, app: &App, area: Rect) {
let help = app.fields[app.selected].help;
let help = app.selected_field().help;
let keybindings = if app.mode == Mode::Editing {
"Enter confirm Esc cancel"
} else if app.server_tabs.len() > 1 {
"j/k select Enter edit Tab switch + add x remove ^S save q quit"
} else {
"Up/Down select Enter edit ^S save q quit"
"j/k select Enter edit + add server ^S save q quit"
};
let mut status_spans: Vec<Span> = vec![Span::styled(
keybindings,
format!(" {}", keybindings),
Style::default().fg(Color::DarkGray),
)];
@@ -592,18 +776,7 @@ fn render_footer(f: &mut Frame, app: &App, area: Rect) {
format!(" {}", help),
Style::default().fg(Color::DarkGray),
)),
Line::from(
status_spans
.into_iter()
.map(|mut s| {
// add left padding to first span
if s.content.as_ref() == keybindings {
s.content = format!(" {}", s.content).into();
}
s
})
.collect::<Vec<_>>(),
),
Line::from(status_spans),
];
let footer = Paragraph::new(footer_text).block(
@@ -614,11 +787,9 @@ fn render_footer(f: &mut Frame, app: &App, area: Rect) {
f.render_widget(footer, area);
}
// ── Entry point ──────────────────────────────────────────────────────────────
// -- Entry point --------------------------------------------------------------
pub fn run(config_path: PathBuf) -> anyhow::Result<SetupOutcome> {
// Setup terminal
terminal::enable_raw_mode()?;
let mut stdout = io::stdout();
execute!(stdout, EnterAlternateScreen)?;
@@ -630,14 +801,13 @@ pub fn run(config_path: PathBuf) -> anyhow::Result<SetupOutcome> {
let result = event_loop(&mut terminal, &mut app);
// Restore terminal
terminal::disable_raw_mode()?;
execute!(terminal.backend_mut(), LeaveAlternateScreen)?;
terminal.show_cursor()?;
result?;
// ── Post-TUI: decide outcome ─────────────────────────────────────
// -- Post-TUI: decide outcome ---------------------------------------------
if !app.saved_once {
return Ok(SetupOutcome::Cancelled);
@@ -648,7 +818,7 @@ pub fn run(config_path: PathBuf) -> anyhow::Result<SetupOutcome> {
eprintln!();
let wants_service = app
.fields
.global_fields
.iter()
.find(|f| f.key == "install_service")
.map(|f| f.value == "true")
@@ -662,13 +832,10 @@ pub fn run(config_path: PathBuf) -> anyhow::Result<SetupOutcome> {
eprintln!(" Starting proxy directly instead.\n");
}
}
} else {
// Uninstall service if it was previously installed but toggled off
if super::service::is_installed() {
if let Err(e) = super::service::uninstall_service() {
eprintln!(" Service uninstall failed: {}", e);
eprintln!();
}
} else if super::service::is_installed() {
if let Err(e) = super::service::uninstall_service() {
eprintln!(" Service uninstall failed: {}", e);
eprintln!();
}
}
@@ -684,7 +851,6 @@ fn event_loop(
if event::poll(Duration::from_millis(200))? {
if let Event::Key(key) = event::read()? {
// Only handle Press events (ignore Release on Windows)
if key.kind == KeyEventKind::Press && app.handle_key(key) {
break;
}
aether-proxy/src/state.rs
+29 -26
View File
@@ -1,42 +1,45 @@
//! Shared application state passed to all subsystems.
//!
//! Consolidates the multiple `Arc<...>` parameters that were previously
//! threaded individually through proxy server, heartbeat, and handlers.
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio_rustls::TlsAcceptor;
use crate::config::Config;
use crate::hardware::HardwareInfo;
use crate::proxy::delegate_client::DelegateClient;
use crate::proxy::target_filter::DnsCache;
use crate::registration::client::AetherClient;
use crate::runtime::SharedDynamicConfig;
use crate::target_filter::DnsCache;
/// Central application state shared across all tasks.
/// Central application state shared across all servers/tunnels.
pub struct AppState {
pub config: Arc<Config>,
pub node_id: Arc<RwLock<String>>,
pub dynamic: SharedDynamicConfig,
pub aether_client: Arc<AetherClient>,
pub hardware_info: Arc<HardwareInfo>,
pub public_ip: String,
pub tls_fingerprint: Option<String>,
pub tls_acceptor: Option<TlsAcceptor>,
/// Shared delegate client for proxy-initiated upstream requests.
pub delegate_client: DelegateClient,
/// Active connection count for metrics reporting.
pub active_connections: Arc<AtomicU64>,
/// Connection concurrency limiter.
pub connection_semaphore: Arc<Semaphore>,
/// DNS cache for upstream target resolution.
/// DNS cache for upstream target resolution (shared).
pub dns_cache: Arc<DnsCache>,
/// Request/latency metrics for heartbeat.
/// Reqwest client for tunnel upstream requests (shared).
pub reqwest_client: reqwest::Client,
}
/// Per-server state: one instance per Aether server connection.
pub struct ServerContext {
/// Human-readable label for logging (e.g. "server-0").
pub server_label: String,
/// Aether server URL for this connection.
pub aether_url: String,
/// Management token for this server.
pub management_token: String,
/// Resolved node name (per-server override or global fallback).
pub node_name: String,
/// Node ID assigned by this Aether server.
pub node_id: Arc<RwLock<String>>,
/// API client for this server.
pub aether_client: Arc<AetherClient>,
/// Dynamic config from this server's heartbeat ACKs.
pub dynamic: SharedDynamicConfig,
/// Per-server active connection count.
pub active_connections: Arc<AtomicU64>,
/// Per-server request/latency metrics.
pub metrics: Arc<ProxyMetrics>,
/// Reconnect attempt counter (reset on successful connection).
pub reconnect_attempts: AtomicU32,
}
/// Aggregate metrics for reporting to Aether.
aether-proxy/src/proxy/target_filter.rs → aether-proxy/src/target_filter.rs
View File
aether-proxy/src/tunnel/client.rs
+138
View File
@@ -0,0 +1,138 @@
//! WebSocket tunnel client: connect, authenticate, and run the tunnel.
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::watch;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tokio_tungstenite::tungstenite::http;
use tracing::{debug, info};
use crate::state::{AppState, ServerContext};
use super::{dispatcher, heartbeat, writer};
/// Outcome of a tunnel session.
pub enum TunnelOutcome {
/// Graceful shutdown requested by the local process.
Shutdown,
/// Remote side disconnected or connection lost — should reconnect.
Disconnected,
}
/// Connect to Aether's WebSocket tunnel endpoint and run until disconnected.
pub async fn connect_and_run(
state: &Arc<AppState>,
server: &Arc<ServerContext>,
shutdown: &mut watch::Receiver<bool>,
) -> Result<TunnelOutcome, anyhow::Error> {
let ws_url = build_tunnel_url(server);
info!(url = %ws_url, "connecting tunnel");
// Build WebSocket request with auth headers
let mut request = ws_url.into_client_request()?;
let headers = request.headers_mut();
headers.insert(
"Authorization",
http::HeaderValue::from_str(&format!("Bearer {}", server.management_token))?,
);
let node_id = server.node_id.read().unwrap().clone();
headers.insert("X-Node-Id", http::HeaderValue::from_str(&node_id)?);
headers.insert(
"X-Node-Name",
http::HeaderValue::from_str(&server.node_name)?,
);
// Connect
let (ws_stream, _response) = tokio_tungstenite::connect_async(request).await?;
info!("tunnel connected");
// Reset reconnect counter on success
server.reconnect_attempts.store(0, Ordering::Relaxed);
// Split into read/write halves
let (ws_sink, ws_read) = futures_util::StreamExt::split(ws_stream);
// Spawn writer task
let (frame_tx, writer_handle) = writer::spawn_writer(ws_sink);
// Spawn heartbeat task
let hb_handle = heartbeat::spawn(
Arc::clone(&state.config),
Arc::clone(server),
frame_tx.clone(),
shutdown.clone(),
);
// Run dispatcher (blocks until disconnect or shutdown)
let state_clone = Arc::clone(state);
let server_clone = Arc::clone(server);
let outcome = tokio::select! {
result = dispatcher::run(state_clone, server_clone, ws_read, frame_tx.clone(), hb_handle) => {
match result {
Ok(()) => TunnelOutcome::Disconnected,
Err(e) => return Err(e),
}
}
_ = shutdown.changed() => {
debug!("shutdown during tunnel dispatch");
TunnelOutcome::Shutdown
}
};
// Drop our sender; the writer will exit once all stream handler clones
// are also dropped (i.e. after they finish their in-flight work).
drop(frame_tx);
// Wait for the writer task to finish with a generous timeout — the
// dispatcher already waits up to 30s for stream handlers, so 35s here
// covers that plus a small margin.
let _ = tokio::time::timeout(Duration::from_secs(35), writer_handle).await;
info!("tunnel disconnected");
Ok(outcome)
}
/// Calculate next reconnect delay with exponential backoff + jitter.
pub fn next_reconnect_delay(state: &Arc<AppState>, server: &Arc<ServerContext>) -> Duration {
let attempt = server.reconnect_attempts.fetch_add(1, Ordering::Relaxed);
let base_ms = state.config.tunnel_reconnect_base_ms;
let max_ms = state.config.tunnel_reconnect_max_ms;
let delay_ms = base_ms.saturating_mul(1u64 << attempt.min(10)).min(max_ms);
let jitter = (delay_ms / 4).max(1);
let jitter_ms = rand_u64() % jitter;
Duration::from_millis(delay_ms + jitter_ms)
}
fn build_tunnel_url(server: &ServerContext) -> String {
let base = server.aether_url.trim_end_matches('/');
let ws_base = if base.starts_with("https://") {
base.replacen("https://", "wss://", 1)
} else if base.starts_with("http://") {
base.replacen("http://", "ws://", 1)
} else {
format!("wss://{}", base)
};
format!("{}/api/internal/proxy-tunnel", ws_base)
}
/// Simple pseudo-random u64 (no external crate needed).
fn rand_u64() -> u64 {
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
static COUNTER: AtomicU64 = AtomicU64::new(0);
let seed = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos() as u64;
let cnt = COUNTER.fetch_add(1, Ordering::Relaxed);
let mut x = seed ^ cnt;
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
x
}
aether-proxy/src/tunnel/dispatcher.rs
+201
View File
@@ -0,0 +1,201 @@
//! Frame dispatcher: reads incoming WebSocket frames and routes them.
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use bytes::Bytes;
use futures_util::StreamExt;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio_tungstenite::tungstenite::Message;
use tracing::{debug, error, warn};
use crate::state::{AppState, ServerContext};
use super::heartbeat::HeartbeatHandle;
use super::protocol::{Frame, MsgType, RequestMeta};
use super::stream_handler;
use super::writer::FrameSender;
/// Run the dispatcher loop, reading from the WebSocket stream.
pub async fn run<S>(
state: Arc<AppState>,
server: Arc<ServerContext>,
mut ws_stream: S,
frame_tx: FrameSender,
heartbeat: HeartbeatHandle,
) -> Result<(), anyhow::Error>
where
S: StreamExt<Item = Result<Message, tokio_tungstenite::tungstenite::Error>>
+ Unpin
+ Send
+ 'static,
{
// Active streams: stream_id -> body sender
let mut streams: HashMap<u32, mpsc::Sender<Frame>> = HashMap::new();
// Track spawned stream handlers so we can wait for them on shutdown
let mut handler_handles: Vec<JoinHandle<()>> = Vec::new();
let max_streams = state.config.tunnel_max_streams.unwrap_or(128) as usize;
let read_err = loop {
let msg_result = match ws_stream.next().await {
Some(r) => r,
None => break None, // stream ended
};
let msg = match msg_result {
Ok(m) => m,
Err(e) => {
error!(error = %e, "WebSocket read error");
break Some(e);
}
};
let data = match msg {
Message::Binary(data) => Bytes::from(data),
Message::Ping(_) => continue,
Message::Pong(_) => continue,
Message::Close(_) => {
debug!("received WebSocket close");
break None;
}
_ => continue,
};
let frame = match Frame::decode(data) {
Ok(f) => f,
Err(e) => {
warn!(error = %e, "failed to decode frame");
continue;
}
};
match frame.msg_type {
MsgType::RequestHeaders => {
// Parse request metadata
let meta: RequestMeta = match serde_json::from_slice(&frame.payload) {
Ok(m) => m,
Err(e) => {
warn!(stream_id = frame.stream_id, error = %e, "invalid request metadata");
let _ = frame_tx
.send(Frame::new(
frame.stream_id,
MsgType::StreamError,
0,
Bytes::from(format!("invalid request metadata: {e}")),
))
.await;
continue;
}
};
if streams.len() >= max_streams {
warn!(
stream_id = frame.stream_id,
"max concurrent streams reached"
);
let _ = frame_tx
.send(Frame::new(
frame.stream_id,
MsgType::StreamError,
0,
Bytes::from("max concurrent streams reached"),
))
.await;
continue;
}
// Create body channel and spawn handler
let (body_tx, body_rx) = mpsc::channel::<Frame>(16);
streams.insert(frame.stream_id, body_tx);
let state_clone = Arc::clone(&state);
let server_clone = Arc::clone(&server);
let tx_clone = frame_tx.clone();
let sid = frame.stream_id;
let handle = tokio::spawn(async move {
stream_handler::handle_stream(
state_clone,
server_clone,
sid,
meta,
body_rx,
tx_clone,
)
.await;
});
handler_handles.push(handle);
debug!(stream_id = frame.stream_id, "new stream started");
}
MsgType::RequestBody => {
if let Some(tx) = streams.get(&frame.stream_id) {
let is_end = frame.is_end_stream();
let sid = frame.stream_id;
let _ = tx.send(frame).await;
if is_end {
streams.remove(&sid);
}
}
}
MsgType::StreamEnd | MsgType::StreamError => {
// Client-side cancellation or end
streams.remove(&frame.stream_id);
}
MsgType::Ping => {
let _ = frame_tx
.send(Frame::control(MsgType::Pong, frame.payload))
.await;
}
MsgType::HeartbeatAck => {
heartbeat.on_ack(frame.payload).await;
}
MsgType::GoAway => {
debug!("received GOAWAY");
break None;
}
_ => {
debug!(msg_type = ?frame.msg_type, "ignoring unexpected frame type");
}
}
// Periodically clean up finished handles to avoid unbounded growth
if handler_handles.len() > max_streams {
handler_handles.retain(|h| !h.is_finished());
}
};
// Drop body senders so stream handlers waiting on body_rx will unblock
streams.clear();
// Wait for active stream handlers to finish so their frame_tx clones
// are dropped before the writer closes the sink.
drain_handlers(handler_handles).await;
match read_err {
Some(e) => Err(e.into()),
None => Ok(()),
}
}
/// Wait for all active stream handlers to finish (with a timeout).
async fn drain_handlers(handles: Vec<JoinHandle<()>>) {
if handles.is_empty() {
return;
}
let count = handles.len();
debug!(count, "waiting for active stream handlers to finish");
let _ = tokio::time::timeout(Duration::from_secs(30), async {
for h in handles {
let _ = h.await;
}
})
.await;
}
aether-proxy/src/tunnel/heartbeat.rs
+113
View File
@@ -0,0 +1,113 @@
//! Tunnel heartbeat: sends metrics over the tunnel, processes ACKs.
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
use bytes::Bytes;
use tokio::sync::watch;
use tracing::{debug, warn};
use crate::config::Config;
use crate::registration::client::RemoteConfig;
use crate::runtime;
use crate::state::ServerContext;
use super::protocol::{Frame, MsgType};
use super::writer::FrameSender;
/// Handle for the dispatcher to forward HeartbeatAck frames.
#[derive(Clone)]
pub struct HeartbeatHandle {
ack_tx: tokio::sync::mpsc::Sender<Bytes>,
}
impl HeartbeatHandle {
pub async fn on_ack(&self, payload: Bytes) {
let _ = self.ack_tx.send(payload).await;
}
}
/// Spawn the heartbeat task. Returns a handle for forwarding ACKs.
pub fn spawn(
config: Arc<Config>,
server: Arc<ServerContext>,
frame_tx: FrameSender,
mut shutdown: watch::Receiver<bool>,
) -> HeartbeatHandle {
let (ack_tx, mut ack_rx) = tokio::sync::mpsc::channel::<Bytes>(4);
let interval = Duration::from_secs(config.heartbeat_interval);
tokio::spawn(async move {
let mut ticker = tokio::time::interval(interval);
ticker.tick().await; // Skip first immediate tick
loop {
tokio::select! {
_ = ticker.tick() => {
let payload = build_heartbeat_payload(&server);
let frame = Frame::control(MsgType::HeartbeatData, payload);
if frame_tx.send(frame).await.is_err() {
break; // Writer closed
}
debug!("sent heartbeat data");
}
Some(ack_payload) = ack_rx.recv() => {
handle_ack(&server, &ack_payload);
}
_ = shutdown.changed() => {
debug!("heartbeat task shutting down");
break;
}
}
}
});
HeartbeatHandle { ack_tx }
}
fn build_heartbeat_payload(server: &ServerContext) -> Bytes {
let node_id = server.node_id.read().unwrap().clone();
let interval_requests = server.metrics.total_requests.swap(0, Ordering::Relaxed);
let interval_latency_ns = server.metrics.total_latency_ns.swap(0, Ordering::Relaxed);
let avg_latency_ms = if interval_requests > 0 {
Some(interval_latency_ns as f64 / interval_requests as f64 / 1_000_000.0)
} else {
None
};
let payload = serde_json::json!({
"node_id": node_id,
"active_connections": server.active_connections.load(Ordering::Relaxed),
"total_requests": interval_requests,
"avg_latency_ms": avg_latency_ms,
});
Bytes::from(serde_json::to_vec(&payload).unwrap_or_default())
}
fn handle_ack(server: &ServerContext, payload: &[u8]) {
if payload.is_empty() {
return;
}
#[derive(serde::Deserialize)]
struct AckPayload {
#[serde(default)]
remote_config: Option<RemoteConfig>,
#[serde(default)]
config_version: u64,
}
match serde_json::from_slice::<AckPayload>(payload) {
Ok(ack) => {
if let Some(ref rc) = ack.remote_config {
runtime::apply_remote_config(&server.dynamic, rc, ack.config_version);
}
}
Err(e) => {
warn!(error = %e, "failed to parse heartbeat ACK");
}
}
}
aether-proxy/src/tunnel/mod.rs
+53
View File
@@ -0,0 +1,53 @@
pub mod client;
pub mod dispatcher;
pub mod heartbeat;
pub mod protocol;
pub mod stream_handler;
pub mod writer;
use std::sync::Arc;
use tokio::sync::watch;
use tracing::{error, info};
use crate::state::{AppState, ServerContext};
/// Run the tunnel mode main loop (connect, dispatch, reconnect).
pub async fn run(
state: &Arc<AppState>,
server: &Arc<ServerContext>,
mut shutdown: watch::Receiver<bool>,
) {
info!(server = %server.server_label, "starting tunnel");
loop {
match client::connect_and_run(state, server, &mut shutdown).await {
Ok(client::TunnelOutcome::Shutdown) => {
info!(server = %server.server_label, "tunnel shut down gracefully");
return;
}
Ok(client::TunnelOutcome::Disconnected) => {
info!(server = %server.server_label, "tunnel disconnected, will reconnect");
}
Err(e) => {
error!(server = %server.server_label, error = %e, "tunnel connection lost");
}
}
if *shutdown.borrow() {
info!(server = %server.server_label, "shutdown requested, not reconnecting");
return;
}
let delay = client::next_reconnect_delay(state, server);
info!(server = %server.server_label, delay_ms = delay.as_millis(), "reconnecting tunnel");
tokio::select! {
_ = tokio::time::sleep(delay) => {}
_ = shutdown.changed() => {
info!(server = %server.server_label, "shutdown requested during reconnect wait");
return;
}
}
}
}
aether-proxy/src/tunnel/protocol.rs
+161
View File
@@ -0,0 +1,161 @@
//! Binary frame protocol for WebSocket tunnel multiplexing.
//!
//! Frame layout (10-byte header + variable payload):
//! ```text
//! | stream_id (4B) | msg_type (1B) | flags (1B) | payload_len (4B) | payload (NB) |
//! ```
use bytes::{Buf, BufMut, Bytes, BytesMut};
pub const HEADER_SIZE: usize = 10;
/// Frame flags.
pub mod flags {
pub const END_STREAM: u8 = 0x01;
pub const GZIP_COMPRESSED: u8 = 0x02;
}
/// Message types for the tunnel protocol.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum MsgType {
RequestHeaders = 0x01,
RequestBody = 0x02,
ResponseHeaders = 0x03,
ResponseBody = 0x04,
StreamEnd = 0x05,
StreamError = 0x06,
Ping = 0x10,
Pong = 0x11,
GoAway = 0x12,
HeartbeatData = 0x13,
HeartbeatAck = 0x14,
}
impl MsgType {
pub fn from_u8(v: u8) -> Option<Self> {
match v {
0x01 => Some(Self::RequestHeaders),
0x02 => Some(Self::RequestBody),
0x03 => Some(Self::ResponseHeaders),
0x04 => Some(Self::ResponseBody),
0x05 => Some(Self::StreamEnd),
0x06 => Some(Self::StreamError),
0x10 => Some(Self::Ping),
0x11 => Some(Self::Pong),
0x12 => Some(Self::GoAway),
0x13 => Some(Self::HeartbeatData),
0x14 => Some(Self::HeartbeatAck),
_ => None,
}
}
}
/// A single multiplexed frame.
#[derive(Debug, Clone)]
pub struct Frame {
pub stream_id: u32,
pub msg_type: MsgType,
pub flags: u8,
pub payload: Bytes,
}
impl Frame {
pub fn new(stream_id: u32, msg_type: MsgType, flags: u8, payload: impl Into<Bytes>) -> Self {
Self {
stream_id,
msg_type,
flags,
payload: payload.into(),
}
}
/// Control frame (stream_id = 0).
pub fn control(msg_type: MsgType, payload: impl Into<Bytes>) -> Self {
Self::new(0, msg_type, 0, payload)
}
pub fn is_end_stream(&self) -> bool {
self.flags & flags::END_STREAM != 0
}
pub fn is_gzip(&self) -> bool {
self.flags & flags::GZIP_COMPRESSED != 0
}
/// Encode into a binary buffer.
pub fn encode(&self) -> Bytes {
let mut buf = BytesMut::with_capacity(HEADER_SIZE + self.payload.len());
buf.put_u32(self.stream_id);
buf.put_u8(self.msg_type as u8);
buf.put_u8(self.flags);
buf.put_u32(self.payload.len() as u32);
buf.put(self.payload.clone());
buf.freeze()
}
/// Decode from a binary buffer.
pub fn decode(mut data: Bytes) -> Result<Self, ProtocolError> {
if data.len() < HEADER_SIZE {
return Err(ProtocolError::TooShort {
expected: HEADER_SIZE,
actual: data.len(),
});
}
let stream_id = data.get_u32();
let msg_type_raw = data.get_u8();
let frame_flags = data.get_u8();
let payload_len = data.get_u32() as usize;
if data.remaining() < payload_len {
return Err(ProtocolError::Incomplete {
expected: HEADER_SIZE + payload_len,
actual: HEADER_SIZE + data.remaining(),
});
}
let msg_type =
MsgType::from_u8(msg_type_raw).ok_or(ProtocolError::UnknownMsgType(msg_type_raw))?;
let payload = data.split_to(payload_len);
Ok(Self {
stream_id,
msg_type,
flags: frame_flags,
payload,
})
}
}
/// Protocol errors.
#[derive(Debug, thiserror::Error)]
pub enum ProtocolError {
#[error("frame too short: expected {expected} bytes, got {actual}")]
TooShort { expected: usize, actual: usize },
#[error("frame incomplete: expected {expected} bytes, got {actual}")]
Incomplete { expected: usize, actual: usize },
#[error("unknown message type: 0x{0:02x}")]
UnknownMsgType(u8),
}
/// JSON payload for REQUEST_HEADERS frames.
#[derive(Debug, serde::Deserialize)]
pub struct RequestMeta {
pub method: String,
pub url: String,
pub headers: std::collections::HashMap<String, String>,
#[serde(default = "default_timeout")]
pub timeout: u64,
}
fn default_timeout() -> u64 {
60
}
/// JSON payload for RESPONSE_HEADERS frames.
#[derive(Debug, serde::Serialize)]
pub struct ResponseMeta {
pub status: u16,
/// Header list preserving duplicates (e.g. multiple Set-Cookie).
pub headers: Vec<(String, String)>,
}
aether-proxy/src/tunnel/stream_handler.rs
+264
View File
@@ -0,0 +1,264 @@
//! Per-stream request handler.
//!
//! Receives request frames, executes the upstream HTTP request,
//! and sends response frames back through the writer channel.
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::{Duration, Instant};
use bytes::Bytes;
use futures_util::StreamExt;
use tokio::sync::mpsc;
use tracing::{debug, warn};
use crate::state::{AppState, ServerContext};
use crate::target_filter;
use super::protocol::{flags, Frame, MsgType, RequestMeta, ResponseMeta};
use super::writer::FrameSender;
/// Maximum response body chunk size per frame (32 KB).
const MAX_CHUNK_SIZE: usize = 32 * 1024;
/// Handle a single stream: receive body, execute upstream, send response.
pub async fn handle_stream(
state: Arc<AppState>,
server: Arc<ServerContext>,
stream_id: u32,
meta: RequestMeta,
mut body_rx: mpsc::Receiver<Frame>,
frame_tx: FrameSender,
) {
let start = Instant::now();
server.active_connections.fetch_add(1, Ordering::Relaxed);
handle_stream_inner(&state, &server, stream_id, meta, &mut body_rx, &frame_tx).await;
server.active_connections.fetch_sub(1, Ordering::Relaxed);
server.metrics.record_request(start.elapsed());
}
async fn handle_stream_inner(
state: &AppState,
server: &ServerContext,
stream_id: u32,
meta: RequestMeta,
body_rx: &mut mpsc::Receiver<Frame>,
frame_tx: &FrameSender,
) {
// Collect request body
let mut body_parts: Vec<Bytes> = Vec::new();
let mut body_done = false;
// Drain body frames
while !body_done {
match body_rx.recv().await {
Some(frame) => {
if frame.msg_type == MsgType::RequestBody {
let payload = if frame.is_gzip() {
match decompress_gzip(&frame.payload) {
Ok(d) => d,
Err(e) => {
send_error(
frame_tx,
stream_id,
&format!("gzip decompress failed: {e}"),
)
.await;
return;
}
}
} else {
frame.payload.clone()
};
if !payload.is_empty() {
body_parts.push(payload);
}
if frame.is_end_stream() {
body_done = true;
}
} else if frame.msg_type == MsgType::StreamEnd
|| frame.msg_type == MsgType::StreamError
{
body_done = true;
if frame.msg_type == MsgType::StreamError {
return; // Client cancelled
}
}
}
None => return, // Channel closed
}
}
let body: Bytes = if body_parts.is_empty() {
Bytes::new()
} else if body_parts.len() == 1 {
body_parts.into_iter().next().unwrap()
} else {
let total: usize = body_parts.iter().map(|b| b.len()).sum();
let mut combined = Vec::with_capacity(total);
for part in &body_parts {
combined.extend_from_slice(part);
}
Bytes::from(combined)
};
// Validate target
let target_url = match url::Url::parse(&meta.url) {
Ok(u) => u,
Err(e) => {
send_error(frame_tx, stream_id, &format!("invalid URL: {e}")).await;
return;
}
};
let host = match target_url.host_str() {
Some(h) => h.to_string(),
None => {
send_error(frame_tx, stream_id, "missing host in URL").await;
return;
}
};
let port = target_url.port_or_known_default().unwrap_or(443);
// DNS + target validation (dns_cache is populated as a side effect)
let dns_start = Instant::now();
{
let allowed_ports = server.dynamic.read().unwrap().allowed_ports.clone();
if let Err(e) =
target_filter::validate_target(&host, port, &allowed_ports, &state.dns_cache).await
{
send_error(frame_tx, stream_id, &format!("target blocked: {e}")).await;
return;
}
}
let dns_ms = dns_start.elapsed().as_millis() as u64;
// Execute upstream request
let client = &state.reqwest_client;
let timeout = Duration::from_secs(meta.timeout);
let method: reqwest::Method = meta.method.parse().unwrap_or(reqwest::Method::GET);
let mut req = client.request(method, &meta.url);
for (k, v) in &meta.headers {
req = req.header(k.as_str(), v.as_str());
}
let body_size = body.len();
if !body.is_empty() {
req = req.body(body);
}
req = req.timeout(timeout);
let upstream_start = Instant::now();
let response = match req.send().await {
Ok(r) => r,
Err(e) => {
let msg = if e.is_timeout() {
"upstream timeout".to_string()
} else if e.is_connect() {
format!("upstream connect error: {e}")
} else {
format!("upstream error: {e}")
};
send_error(frame_tx, stream_id, &msg).await;
return;
}
};
// Send RESPONSE_HEADERS
let status = response.status().as_u16();
let ttfb_ms = upstream_start.elapsed().as_millis() as u64;
let mut resp_headers: Vec<(String, String)> = Vec::new();
for (k, v) in response.headers() {
if let Ok(vs) = v.to_str() {
resp_headers.push((k.as_str().to_string(), vs.to_string()));
}
}
// Inject proxy timing (same format as delegate mode)
let timing = serde_json::json!({
"dns_ms": dns_ms,
"ttfb_ms": ttfb_ms,
"upstream_ms": ttfb_ms,
"upstream_processing_ms": ttfb_ms.saturating_sub(dns_ms),
"body_size": body_size,
"mode": "tunnel",
});
resp_headers.push(("x-proxy-timing".to_string(), timing.to_string()));
let resp_meta = ResponseMeta {
status,
headers: resp_headers,
};
let meta_json = serde_json::to_vec(&resp_meta).unwrap_or_default();
let _ = frame_tx
.send(Frame::new(
stream_id,
MsgType::ResponseHeaders,
0,
meta_json,
))
.await;
// Stream response body
let mut stream = response.bytes_stream();
while let Some(chunk_result) = stream.next().await {
match chunk_result {
Ok(chunk) => {
if chunk.len() <= MAX_CHUNK_SIZE {
// 大多数 chunk 无需分割，直接零拷贝发送
let _ = frame_tx
.send(Frame::new(stream_id, MsgType::ResponseBody, 0, chunk))
.await;
} else {
// 超大 chunk 按 MAX_CHUNK_SIZE 分割（使用 Bytes::slice 避免拷贝）
let mut offset = 0;
while offset < chunk.len() {
let end = (offset + MAX_CHUNK_SIZE).min(chunk.len());
let slice = chunk.slice(offset..end);
let _ = frame_tx
.send(Frame::new(stream_id, MsgType::ResponseBody, 0, slice))
.await;
offset = end;
}
}
}
Err(e) => {
warn!(stream_id, error = %e, "upstream body read error");
send_error(frame_tx, stream_id, &format!("body read error: {e}")).await;
return;
}
}
}
// Send STREAM_END
let _ = frame_tx
.send(Frame::new(
stream_id,
MsgType::StreamEnd,
flags::END_STREAM,
Bytes::new(),
))
.await;
debug!(stream_id, status, "stream completed");
}
async fn send_error(tx: &FrameSender, stream_id: u32, msg: &str) {
let _ = tx
.send(Frame::new(
stream_id,
MsgType::StreamError,
0,
Bytes::from(msg.to_string()),
))
.await;
}
fn decompress_gzip(data: &[u8]) -> Result<Bytes, std::io::Error> {
use flate2::read::GzDecoder;
use std::io::Read;
let mut decoder = GzDecoder::new(data);
let mut buf = Vec::new();
decoder.read_to_end(&mut buf)?;
Ok(Bytes::from(buf))
}
aether-proxy/src/tunnel/writer.rs
+37
View File
@@ -0,0 +1,37 @@
//! Dedicated WebSocket writer task.
//!
//! All frame writes go through an mpsc channel to a single writer task,
//! avoiding contention on the WebSocket sink.
use futures_util::SinkExt;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio_tungstenite::tungstenite::Message;
use tracing::{debug, error};
use super::protocol::Frame;
/// Sender half — cloned by stream handlers and heartbeat.
pub type FrameSender = mpsc::Sender<Frame>;
/// Spawn the writer task. Returns the sender and a JoinHandle for cleanup.
pub fn spawn_writer<S>(mut sink: S) -> (FrameSender, JoinHandle<()>)
where
S: SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin + Send + 'static,
{
let (tx, mut rx) = mpsc::channel::<Frame>(256);
let handle = tokio::spawn(async move {
while let Some(frame) = rx.recv().await {
let data = frame.encode();
if let Err(e) = sink.send(Message::Binary(data.into())).await {
error!(error = %e, "failed to write frame to WebSocket");
break;
}
}
debug!("writer task exiting");
let _ = sink.close().await;
});
(tx, handle)
}