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

refactor: 代理节点架构重构与功能增强

Browse Source 
aether-proxy:
- 重构 main.rs，拆分为 app/state/hardware/net 模块
- setup.rs 拆分为 setup/tui.rs + setup/service.rs，支持 systemd 服务管理子命令
- 新增 delegate 端点，支持后端通过代理节点转发请求而非传统 CONNECT 代理
- 注册时上报硬件信息（CPU/内存/fd_limit）和估算最大并发数
- 心跳上报活跃连接数，支持远程下发 node_name 配置
- HTTP 转发时剥离 X-Forwarded-* 等敏感头部
- 切换到 rustls-tls，降低日志级别减少噪音

后端:
- 从 http_client.py 提取代理相关逻辑至 proxy_node/resolver.py
- 从 routes.py 提取业务逻辑至 proxy_node/service.py
- handler 支持 delegate 模式（通过代理节点 HTTP 端点转发而非 CONNECT 隧道）
- ProxyNode 模型新增 hardware_info 和 estimated_max_concurrency 字段

前端:
- 新增 HardwareTooltip 组件展示节点硬件信息
- 远程配置支持下发 node_name
This commit is contained in:
fawney19
2026-02-08 13:33:08 +08:00
parent 254d30d32d
commit 519ad67eb1
44 changed files with 3339 additions and 1709 deletions
Download Patch File Download Diff File Expand all files Collapse all files

200
aether-proxy/src/app.rs Normal file
View File

@@ -0,0 +1,200 @@
//! 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::{Arc, RwLock};
use tokio::signal;
use tokio::sync::watch;
use tracing::{error, info};
use crate::config::Config;
use crate::net;
use crate::registration::client::AetherClient;
use crate::runtime::{self, DynamicConfig};
use crate::state::AppState;
use crate::{hardware, proxy};
/// Run the full application lifecycle after config has been parsed.
pub async fn run(mut config: Config) -> anyhow::Result<()> {
init_tracing(&config);
info!(
version = env!("CARGO_PKG_VERSION"),
port = config.listen_port,
node_name = %config.node_name,
"aether-proxy starting"
);
// Resolve public IP
let public_ip = match &config.public_ip {
Some(ip) => ip.clone(),
None => net::detect_public_ip().await?,
};
info!(public_ip = %public_ip, "using public IP");
// Auto-detect region if not configured
if config.node_region.is_none() {
if let Some(region) = net::detect_region(&public_ip).await {
config.node_region = Some(region);
}
}
// 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)
let hw_info = hardware::collect();
// Register with Aether
let aether_client = Arc::new(AetherClient::new(&config));
let node_id = aether_client
.register(
&config,
&public_ip,
config.enable_tls,
tls_fingerprint.as_deref(),
Some(&hw_info),
)
.await?;
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).
// No overall timeout — SSE streams can last indefinitely.
// Connect timeout limits connection establishment; Aether controls
// first-byte / idle timeouts on its own side.
let delegate_client = reqwest::Client::builder()
.connect_timeout(std::time::Duration::from_secs(30))
.pool_max_idle_per_host(20)
.pool_idle_timeout(std::time::Duration::from_secs(90))
.build()
.expect("failed to create delegate HTTP client");
// 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)),
});
// Shutdown signal channel
let (shutdown_tx, shutdown_rx) = watch::channel(false);
// Start heartbeat task
let heartbeat_handle = {
let state = Arc::clone(&state);
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");
}
// Wait for tasks to finish
let _ = tokio::join!(heartbeat_handle, server_handle);
info!("aether-proxy stopped");
Ok(())
}
fn init_tracing(config: &Config) {
use tracing_subscriber::prelude::*;
use tracing_subscriber::{reload, EnvFilter};
let filter = EnvFilter::try_new(&config.log_level).unwrap_or_else(|_| EnvFilter::new("info"));
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);
}
}));
if config.log_json {
tracing_subscriber::registry()
.with(filter_layer)
.with(tracing_subscriber::fmt::layer().json())
.init();
} else {
tracing_subscriber::registry()
.with(filter_layer)
.with(tracing_subscriber::fmt::layer())
.init();
}
}
async fn wait_for_shutdown() {
let ctrl_c = async {
signal::ctrl_c()
.await
.expect("failed to install Ctrl+C handler");
};
#[cfg(unix)]
let terminate = async {
signal::unix::signal(signal::unix::SignalKind::terminate())
.expect("failed to install SIGTERM handler")
.recv()
.await;
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! {
_ = ctrl_c => {},
_ = terminate => {},
}
}

25
aether-proxy/src/auth/hmac.rs
View File

@@ -24,7 +24,9 @@ impl std::fmt::Display for AuthError {
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::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"),
@@ -60,9 +62,7 @@ pub fn validate_proxy_auth(
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)?;
let (username, password) = decoded.split_once(':').ok_or(AuthError::InvalidBasicAuth)?;
if username != "hmac" {
return Err(AuthError::InvalidUsername);
@@ -82,11 +82,7 @@ pub fn validate_proxy_auth(
.expect("system clock before epoch")
.as_secs();
let diff = if now > timestamp {
now - timestamp
} else {
timestamp - now
};
let diff = now.abs_diff(timestamp);
if diff > timestamp_tolerance {
return Err(AuthError::TimestampExpired);
@@ -129,6 +125,9 @@ mod tests {
timestamp_tolerance: 300,
log_level: "info".to_string(),
log_json: false,
enable_tls: false,
tls_cert: String::new(),
tls_key: String::new(),
}
}
@@ -138,8 +137,7 @@ mod tests {
.unwrap()
.as_secs();
let payload = format!("{}\n{}", now, node_id);
let mut mac =
HmacSha256::new_from_slice(config.hmac_key.as_bytes()).unwrap();
let mut mac = HmacSha256::new_from_slice(config.hmac_key.as_bytes()).unwrap();
mac.update(payload.as_bytes());
let sig = hex::encode(mac.finalize().into_bytes());
let cred = format!("hmac:{}.{}", now, sig);
@@ -151,7 +149,10 @@ mod tests {
fn test_valid_auth() {
let config = make_config();
let header = make_valid_auth(&config, "node-1");
assert!(validate_proxy_auth(Some(&header), &config, "node-1", config.timestamp_tolerance).is_ok());
assert!(
validate_proxy_auth(Some(&header), &config, "node-1", config.timestamp_tolerance)
.is_ok()
);
}
#[test]

12
aether-proxy/src/config.rs
View File

@@ -64,11 +64,19 @@ pub struct Config {
pub enable_tls: bool,
/// Path to TLS certificate PEM file
#[arg(long, env = "AETHER_PROXY_TLS_CERT", default_value = "aether-proxy-cert.pem")]
#[arg(
long,
env = "AETHER_PROXY_TLS_CERT",
default_value = "aether-proxy-cert.pem"
)]
pub tls_cert: String,
/// Path to TLS private key PEM file
#[arg(long, env = "AETHER_PROXY_TLS_KEY", default_value = "aether-proxy-key.pem")]
#[arg(
long,
env = "AETHER_PROXY_TLS_KEY",
default_value = "aether-proxy-key.pem"
)]
pub tls_key: String,
}

79
aether-proxy/src/hardware.rs Normal file
View File

@@ -0,0 +1,79 @@
use serde::Serialize;
use sysinfo::System;
use tracing::info;
/// Hardware information collected at startup.
///
/// The struct is `Serialize`-able so it can be sent directly as the
/// `hardware_info` JSON bag in the registration request. New fields
/// can be added without database schema migrations.
#[derive(Debug, Clone, Serialize)]
pub struct HardwareInfo {
pub cpu_cores: u32,
pub total_memory_mb: u64,
pub os_info: String,
pub fd_limit: u64,
#[serde(skip)]
pub estimated_max_concurrency: u64,
}
/// Collect hardware information and estimate max concurrency.
///
/// Should be called once at startup -- hardware does not change at runtime.
pub fn collect() -> HardwareInfo {
let sys = System::new_all();
let cpu_cores = sys.cpus().len() as u32;
let total_memory_mb = sys.total_memory() / (1024 * 1024);
let os_info = format!(
"{} {}",
System::name().unwrap_or_else(|| "Unknown".into()),
System::os_version().unwrap_or_default(),
)
.trim()
.to_string();
// Estimate max concurrent connections:
// - Each tokio async task uses ~8-16 KB stack + heap buffers
// - OS file descriptor limit is often the real bottleneck
// - Conservative formula: min(fd_limit - 100, ram_mb * 40, cpu_cores * 2000)
let fd_limit = get_fd_limit();
let by_fd = fd_limit.saturating_sub(100);
let by_ram = total_memory_mb.saturating_mul(40);
let by_cpu = (cpu_cores as u64).saturating_mul(2000);
let estimated_max_concurrency = by_fd.min(by_ram).min(by_cpu);
info!(
cpu_cores,
total_memory_mb,
os_info = %os_info,
fd_limit,
estimated_max_concurrency,
"hardware info collected"
);
HardwareInfo {
cpu_cores,
total_memory_mb,
os_info,
fd_limit,
estimated_max_concurrency,
}
}
/// Read the soft file-descriptor limit (RLIMIT_NOFILE).
fn get_fd_limit() -> u64 {
#[cfg(unix)]
{
let mut rlim = libc::rlimit {
rlim_cur: 0,
rlim_max: 0,
};
let ret = unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut rlim) };
if ret == 0 {
return rlim.rlim_cur;
}
}
// Fallback for non-unix or error
1024
}

202
aether-proxy/src/main.rs
View File

@@ -1,21 +1,19 @@
mod app;
mod auth;
mod config;
mod hardware;
mod net;
mod proxy;
mod registration;
mod runtime;
mod setup;
mod state;
use std::path::PathBuf;
use std::sync::{Arc, RwLock};
use clap::Parser;
use tokio::signal;
use tokio::sync::watch;
use tracing::{error, info};
use config::Config;
use registration::client::{detect_public_ip, AetherClient};
use runtime::DynamicConfig;
/// Default config file name.
const DEFAULT_CONFIG: &str = "aether-proxy.toml";
@@ -24,186 +22,54 @@ const DEFAULT_CONFIG: &str = "aether-proxy.toml";
async fn main() -> anyhow::Result<()> {
let args: Vec<String> = std::env::args().collect();
// ── Handle `setup` subcommand before clap parsing ────────────────────
if args.len() > 1 && args[1] == "setup" {
let path = args
.get(2)
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from(DEFAULT_CONFIG));
return setup::run(path);
// Handle subcommands before clap parsing (these don't need Config)
if args.len() > 1 {
match args[1].as_str() {
"setup" => {
let path = args
.get(2)
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from(DEFAULT_CONFIG));
return setup::run(path);
}
"start" => return setup::service::cmd_start(),
"status" => return setup::service::cmd_status(),
"logs" => return setup::service::cmd_logs(),
"restart" => return setup::service::cmd_restart(),
"stop" => return setup::service::cmd_stop(),
"uninstall" => return setup::service::cmd_uninstall(),
_ => {} // fall through to clap (--help, --version, config args)
}
}
// ── Load config file as env-var defaults (before clap) ───────────────
let config_file_path = std::env::var("AETHER_PROXY_CONFIG")
.unwrap_or_else(|_| DEFAULT_CONFIG.to_string());
// Load config file as env-var defaults (before clap)
let config_file_path =
std::env::var("AETHER_PROXY_CONFIG").unwrap_or_else(|_| DEFAULT_CONFIG.to_string());
if std::path::Path::new(&config_file_path).exists() {
if let Ok(file_cfg) = config::ConfigFile::load(std::path::Path::new(&config_file_path)) {
file_cfg.inject_env();
}
}
// ── Parse config; fall back to setup TUI if required args are missing ─
// Parse config; fall back to setup TUI if required args are missing
let config = match Config::try_parse() {
Ok(c) => c,
Err(e) => {
if e.kind() == clap::error::ErrorKind::MissingRequiredArgument {
eprintln!("缺少必要配置，启动交互式配置向导...\n");
eprintln!("Missing required config, launching setup wizard...\n");
return setup::run(PathBuf::from(&config_file_path));
}
e.exit();
}
};
// Initialize tracing (with hot-reload support)
init_tracing(&config);
info!(
version = env!("CARGO_PKG_VERSION"),
port = config.listen_port,
node_name = %config.node_name,
"aether-proxy starting"
);
// Resolve public IP
let public_ip = match &config.public_ip {
Some(ip) => ip.clone(),
None => detect_public_ip().await?,
};
info!(public_ip = %public_ip, "using public IP");
// Register with Aether
let aether_client = Arc::new(AetherClient::new(&config));
// 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)
};
let node_id = aether_client
.register(&config, &public_ip, config.enable_tls, tls_fingerprint.as_deref())
.await?;
info!(node_id = %node_id, "node registered");
let node_id = Arc::new(RwLock::new(node_id));
// Dynamic config (hot-reloadable via heartbeat)
let dynamic = Arc::new(RwLock::new(DynamicConfig::from_config(&config)));
// Shutdown signal channel
let (shutdown_tx, shutdown_rx) = watch::channel(false);
let config = Arc::new(config);
// Start heartbeat task
let heartbeat_handle = {
let client = Arc::clone(&aether_client);
let node_id = Arc::clone(&node_id);
let config = Arc::clone(&config);
let dynamic = Arc::clone(&dynamic);
let public_ip = public_ip.clone();
let fingerprint = tls_fingerprint.clone();
let rx = shutdown_rx.clone();
tokio::spawn(async move {
registration::heartbeat::run(client, node_id, config, public_ip, fingerprint, dynamic, rx).await;
})
};
// Start proxy server
let server_handle = {
let config = Arc::clone(&config);
let node_id = Arc::clone(&node_id);
let dynamic = Arc::clone(&dynamic);
let rx = shutdown_rx.clone();
let tls = tls_acceptor.clone();
tokio::spawn(async move {
if let Err(e) = proxy::server::run(config, node_id, dynamic, tls, 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 = node_id.read().unwrap().clone();
if let Err(e) = aether_client.unregister(&current_node_id).await {
error!(error = %e, "unregister failed during shutdown");
// Warn if systemd service is already running (would cause port conflict)
if setup::service::is_service_active() {
eprintln!("Warning: systemd service is already running.");
eprintln!("Use `aether-proxy stop` to stop it first, or manage via subcommands:");
eprintln!(" aether-proxy status / logs / restart / stop");
std::process::exit(1);
}
// Wait for tasks to finish
let _ = tokio::join!(heartbeat_handle, server_handle);
info!("aether-proxy stopped");
Ok(())
}
fn init_tracing(config: &Config) {
use tracing_subscriber::prelude::*;
use tracing_subscriber::{reload, EnvFilter};
let filter =
EnvFilter::try_new(&config.log_level).unwrap_or_else(|_| EnvFilter::new("info"));
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);
}
}));
if config.log_json {
tracing_subscriber::registry()
.with(filter_layer)
.with(tracing_subscriber::fmt::layer().json())
.init();
} else {
tracing_subscriber::registry()
.with(filter_layer)
.with(tracing_subscriber::fmt::layer())
.init();
}
}
async fn wait_for_shutdown() {
let ctrl_c = async {
signal::ctrl_c()
.await
.expect("failed to install Ctrl+C handler");
};
#[cfg(unix)]
let terminate = async {
signal::unix::signal(signal::unix::SignalKind::terminate())
.expect("failed to install SIGTERM handler")
.recv()
.await;
};
#[cfg(not(unix))]
let terminate = std::future::pending::<()>();
tokio::select! {
_ = ctrl_c => {},
_ = terminate => {},
}
app::run(config).await
}

86
aether-proxy/src/net.rs Normal file
View File

@@ -0,0 +1,86 @@
//! Network utility functions (public IP detection, region detection).
//!
//! These are standalone helpers not tied to any specific client or service.
use reqwest::Client;
use tracing::{debug, info};
/// Auto-detect public IP by querying external services.
pub async fn detect_public_ip() -> anyhow::Result<String> {
let endpoints = [
"https://api.ipify.org",
"https://ifconfig.me/ip",
"https://icanhazip.com",
];
let client = Client::builder()
.timeout(std::time::Duration::from_secs(5))
.build()?;
for endpoint in &endpoints {
match client.get(*endpoint).send().await {
Ok(resp) if resp.status().is_success() => {
let ip = resp.text().await?.trim().to_string();
if !ip.is_empty() {
info!(ip = %ip, source = %endpoint, "detected public IP");
return Ok(ip);
}
}
Ok(resp) => {
debug!(endpoint = %endpoint, status = %resp.status(), "IP detection failed");
}
Err(e) => {
debug!(endpoint = %endpoint, error = %e, "IP detection failed");
}
}
}
anyhow::bail!("failed to detect public IP from any source; use --public-ip")
}
/// Auto-detect geographic region from a public IP address.
///
/// Uses multiple providers with HTTPS preferred. Falls back to ip-api.com
/// over plain HTTP (their free tier doesn't support HTTPS).
/// This is best-effort and non-sensitive -- region detection should never
/// block startup.
pub async fn detect_region(ip: &str) -> Option<String> {
// Try HTTPS provider first
let https_url = format!("https://ipinfo.io/{}/country", ip);
let client = Client::builder()
.timeout(std::time::Duration::from_secs(5))
.build()
.ok()?;
// Try ipinfo.io (HTTPS, returns plain text country code)
if let Ok(resp) = client.get(&https_url).send().await {
if resp.status().is_success() {
if let Ok(text) = resp.text().await {
let code = text.trim();
if !code.is_empty() && code.len() <= 3 {
info!(region = %code, ip = %ip, source = "ipinfo.io", "detected region");
return Some(code.to_string());
}
}
}
}
// Fallback: ip-api.com (HTTP only on free tier, non-sensitive data)
let http_url = format!("http://ip-api.com/json/{}?fields=countryCode", ip);
match client.get(&http_url).send().await {
Ok(resp) if resp.status().is_success() => {
let body: serde_json::Value = resp.json().await.ok()?;
let code = body.get("countryCode")?.as_str()?;
if code.is_empty() {
return None;
}
info!(region = %code, ip = %ip, source = "ip-api.com", "detected region");
Some(code.to_string())
}
_ => {
debug!(ip = %ip, "region detection failed");
None
}
}
}

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

@@ -4,7 +4,7 @@ use std::sync::Arc;
use hyper::body::Incoming;
use hyper::{Request, Response};
use tokio::net::TcpStream;
use tracing::{debug, info, warn};
use tracing::{debug, warn};
use crate::auth;
use crate::config::Config;
@@ -53,7 +53,7 @@ pub async fn handle_connect(
}
};
info!(target = %target_addr, "CONNECT tunnel establishing");
debug!(target = %target_addr, "CONNECT tunnel establishing");
// Connect to target
let target_stream = match TcpStream::connect(target_addr).await {
@@ -65,28 +65,29 @@ pub async fn handle_connect(
};
// Respond 200 and upgrade connection to raw TCP tunnel
let target_display = target_addr.to_string();
tokio::task::spawn(async move {
match hyper::upgrade::on(req).await {
Ok(upgraded) => {
let mut upgraded =
hyper_util::rt::TokioIo::new(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)) => {
info!(
debug!(
target = %target_display,
from_client,
from_target,
"CONNECT tunnel closed"
);
}
Err(e) => {
debug!(error = %e, "CONNECT tunnel error");
debug!(target = %target_display, error = %e, "CONNECT tunnel error");
}
}
}
Err(e) => {
warn!(error = %e, "CONNECT upgrade failed");
warn!(target = %target_display, error = %e, "CONNECT upgrade failed");
}
}
});

221
aether-proxy/src/proxy/delegate.rs Normal file
View File

@@ -0,0 +1,221 @@
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;
use futures_util::TryStreamExt;
use http_body_util::{BodyExt, Full, Limited, StreamBody};
use hyper::body::{Frame, Incoming};
use hyper::{Request, Response};
use serde::Deserialize;
use tracing::{debug, warn};
use url::Url;
use crate::auth;
use crate::config::Config;
use crate::proxy::plain::BoxBody;
use crate::proxy::target_filter;
/// Delegation request payload sent by Aether.
#[derive(Debug, Deserialize)]
struct DelegateRequest {
method: String,
url: String,
headers: HashMap<String, String>,
body: Option<String>,
/// Accepted but not used on the proxy side — Aether controls timeouts.
#[allow(dead_code)]
timeout: Option<u64>,
}
/// Handle delegation requests: Aether sends a full request description,
/// and the proxy issues the actual upstream HTTP call using its own TLS stack.
///
/// Endpoint: POST /_aether/delegate
pub async fn handle_delegate(
req: Request<Incoming>,
config: Arc<Config>,
node_id: &str,
allowed_ports: &HashSet<u16>,
timestamp_tolerance: u64,
http_client: &reqwest::Client,
) -> Response<BoxBody> {
// Authenticate via Authorization header (same HMAC scheme as Proxy-Authorization)
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, node_id, timestamp_tolerance) {
warn!(error = %e, "delegate auth failed");
return error_response(401, "authentication_failed", &e.to_string());
}
// Read and parse request body (limit to 10 MB to prevent OOM)
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, "failed to read delegate request body");
return error_response(413, "payload_too_large", "request body exceeds 10MB limit");
}
};
let delegate_req: DelegateRequest = match serde_json::from_slice(&body_bytes) {
Ok(r) => r,
Err(e) => {
warn!(error = %e, "invalid delegate request JSON");
return error_response(400, "bad_request", &format!("invalid JSON: {}", e));
}
};
// Target filter: validate the upstream URL against allowed ports and private IP checks
let parsed_url = match Url::parse(&delegate_req.url) {
Ok(u) => u,
Err(e) => {
warn!(url = %delegate_req.url, error = %e, "invalid delegate 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 = %delegate_req.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);
if let Err(e) = target_filter::validate_target(&host, port, allowed_ports) {
warn!(host = %host, port, error = %e, "delegate target rejected");
return error_response(403, "target_not_allowed", &e.to_string());
}
debug!(
method = %delegate_req.method,
url = %delegate_req.url,
"delegate request"
);
// Build upstream request
let method = match delegate_req.method.parse::<reqwest::Method>() {
Ok(m) => m,
Err(e) => {
warn!(error = %e, method = %delegate_req.method, "invalid HTTP method");
return error_response(400, "bad_request", &format!("invalid method: {}", e));
}
};
let mut upstream_req = http_client.request(method, &delegate_req.url);
// NOTE: We intentionally do NOT set a per-request timeout here.
// reqwest's `.timeout()` caps the *entire* request including body streaming,
// which would truncate long-lived SSE streams. The delegate_client already
// has a 30s connect_timeout for connection establishment, and Aether controls
// first-byte / idle timeouts on its own side via asyncio.
// Set headers (skip `host` — reqwest sets it from the URL automatically,
// and a duplicate Host header can confuse certain upstreams)
for (name, value) in &delegate_req.headers {
if name.eq_ignore_ascii_case("host") {
continue;
}
upstream_req = upstream_req.header(name.as_str(), value.as_str());
}
// Set body
if let Some(body) = delegate_req.body {
upstream_req = upstream_req.body(body);
}
// Send upstream request
let upstream_resp = match upstream_req.send().await {
Ok(resp) => resp,
Err(e) => {
warn!(url = %delegate_req.url, error = %e, "delegate upstream request failed");
// Sanitize: strip URL details from error message to avoid leaking
// API keys or paths that may appear in query strings / paths.
let safe_detail = sanitize_upstream_error(&e.to_string());
if e.is_timeout() {
return error_response(504, "upstream_timeout", &safe_detail);
}
return error_response(502, "upstream_connection_failed", &safe_detail);
}
};
// Build response: pass through upstream status + headers, stream body back
let status = upstream_resp.status().as_u16();
let upstream_headers = upstream_resp.headers().clone();
debug!(url = %delegate_req.url, status, "delegate upstream response");
// Stream the response body
let body_stream = upstream_resp
.bytes_stream()
.map_ok(Frame::data)
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { Box::new(e) });
let stream_body: BoxBody = StreamBody::new(body_stream).boxed();
let mut builder = Response::builder().status(status);
for (name, value) in upstream_headers.iter() {
builder = builder.header(name, value);
}
builder
.body(stream_body)
.unwrap_or_else(|_| Response::builder().status(500).body(empty_box()).unwrap())
}
// ── 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 empty_box() -> BoxBody {
Full::new(bytes::Bytes::new())
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { match e {} })
.boxed()
}
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()
}

1
aether-proxy/src/proxy/mod.rs
View File

@@ -1,4 +1,5 @@
pub mod connect;
pub mod delegate;
pub mod plain;
pub mod server;
pub mod target_filter;

24
aether-proxy/src/proxy/plain.rs
View File

@@ -4,7 +4,7 @@ use std::sync::Arc;
use http_body_util::{BodyExt, Full};
use hyper::body::Incoming;
use hyper::{Request, Response};
use tracing::{debug, info, warn};
use tracing::{debug, warn};
use crate::auth;
use crate::config::Config;
@@ -56,22 +56,28 @@ pub async fn handle_plain(
}
};
info!(target = %target_addr, method = %req.method(), "HTTP proxy forwarding");
let method = req.method().clone();
debug!(target = %target_addr, method = %method, "HTTP proxy forwarding");
// Build outgoing request (strip proxy headers, use relative URI)
let path_and_query = uri
.path_and_query()
.map(|pq| pq.as_str())
.unwrap_or("/");
let path_and_query = uri.path_and_query().map(|pq| pq.as_str()).unwrap_or("/");
let mut builder = Request::builder()
.method(req.method())
.uri(path_and_query)
.version(req.version());
// Copy headers, skipping proxy-specific ones
// Copy headers, skipping proxy-specific and forwarding-related ones
for (name, value) in req.headers() {
if name == "proxy-authorization" || name == "proxy-connection" {
if name == "proxy-authorization"
|| name == "proxy-connection"
|| name == "x-forwarded-for"
|| name == "x-forwarded-host"
|| name == "x-forwarded-proto"
|| name == "x-real-ip"
|| name == "forwarded"
|| name == "via"
{
continue;
}
builder = builder.header(name, value);
@@ -116,7 +122,7 @@ pub async fn handle_plain(
match sender.send_request(outgoing).await {
Ok(resp) => {
info!(target = %target_addr, status = resp.status().as_u16(), "HTTP proxy response");
debug!(target = %target_addr, method = %method, status = resp.status().as_u16(), "HTTP proxy response");
// Stream the response body directly — no buffering
let (parts, body) = resp.into_parts();
let body: BoxBody = body

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

@@ -1,21 +1,20 @@
use std::net::SocketAddr;
use std::sync::{Arc, RwLock};
use std::sync::atomic::Ordering;
use std::sync::Arc;
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};
use hyper::rt::{Read, Write};
use hyper_util::rt::TokioIo;
use tokio::net::TcpListener;
use tokio::sync::watch;
use tokio_rustls::TlsAcceptor;
use tracing::{debug, info, warn};
use crate::config::Config;
use crate::proxy::{connect, plain, tls};
use crate::runtime::SharedDynamicConfig;
use crate::proxy::{connect, delegate, plain, tls};
use crate::state::AppState;
/// Start the proxy server.
///
@@ -23,20 +22,17 @@ use crate::runtime::SharedDynamicConfig;
/// - CONNECT requests -> tunnel handler
/// - Other HTTP requests -> plain forward proxy handler
///
/// When `tls_acceptor` is provided, the server operates in dual-stack mode:
/// 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(
config: Arc<Config>,
node_id: Arc<RwLock<String>>,
dynamic: SharedDynamicConfig,
tls_acceptor: Option<TlsAcceptor>,
state: &Arc<AppState>,
mut shutdown_rx: watch::Receiver<bool>,
) -> anyhow::Result<()> {
let addr = SocketAddr::from(([0, 0, 0, 0], config.listen_port));
let addr = SocketAddr::from(([0, 0, 0, 0], state.config.listen_port));
let listener = TcpListener::bind(addr).await?;
if tls_acceptor.is_some() {
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)");
@@ -53,26 +49,22 @@ pub async fn run(
}
};
info!(peer = %peer_addr, "new connection");
debug!(peer = %peer_addr, "new connection");
let config = Arc::clone(&config);
let node_id = Arc::clone(&node_id);
let dynamic = Arc::clone(&dynamic);
let tls_acceptor = tls_acceptor.clone();
let state = Arc::clone(state);
state.active_connections.fetch_add(1, Ordering::Relaxed);
tokio::task::spawn(async move {
// Dual-stack: peek first byte to decide TLS vs plain HTTP
if let Some(acceptor) = &tls_acceptor {
if let Some(ref acceptor) = state.tls_acceptor {
if tls::is_tls_client_hello(&stream).await {
match acceptor.accept(stream).await {
match acceptor.clone().accept(stream).await {
Ok(tls_stream) => {
debug!(peer = %peer_addr, "TLS handshake ok");
serve_connection(
TokioIo::new(tls_stream),
peer_addr,
config,
node_id,
dynamic,
&state,
)
.await;
}
@@ -80,6 +72,7 @@ pub async fn run(
debug!(peer = %peer_addr, error = %e, "TLS handshake failed");
}
}
state.active_connections.fetch_sub(1, Ordering::Relaxed);
return;
}
}
@@ -88,11 +81,11 @@ pub async fn run(
serve_connection(
TokioIo::new(stream),
peer_addr,
config,
node_id,
dynamic,
&state,
)
.await;
state.active_connections.fetch_sub(1, Ordering::Relaxed);
});
}
_ = shutdown_rx.changed() => {
@@ -106,22 +99,26 @@ pub async fn run(
}
/// Serve a single HTTP/1.1 connection (works over both plain TCP and TLS).
async fn serve_connection<I>(
io: I,
peer_addr: SocketAddr,
config: Arc<Config>,
node_id: Arc<RwLock<String>>,
dynamic: SharedDynamicConfig,
) where
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 node_id = Arc::clone(&state.node_id);
let dynamic = Arc::clone(&state.dynamic);
let delegate_client = state.delegate_client.clone();
let service = service_fn(move |req: Request<Incoming>| {
let config = Arc::clone(&config);
let node_id = Arc::clone(&node_id);
let dynamic = Arc::clone(&dynamic);
let delegate_client = delegate_client.clone();
async move {
type BoxBody = http_body_util::combinators::BoxBody<bytes::Bytes, Box<dyn std::error::Error + Send + Sync>>;
type BoxBody = http_body_util::combinators::BoxBody<
bytes::Bytes,
Box<dyn std::error::Error + Send + Sync>,
>;
// Snapshot current dynamic values (may be updated by remote config)
let current_node_id = node_id.read().unwrap().clone();
@@ -145,6 +142,18 @@ async fn serve_connection<I>(
.boxed()
});
Ok::<_, hyper::Error>(resp)
} else if req.uri().path() == "/_aether/delegate" && req.method() == hyper::Method::POST
{
let resp = delegate::handle_delegate(
req,
config,
&current_node_id,
&allowed_ports,
timestamp_tolerance,
&delegate_client,
)
.await;
Ok(resp)
} else {
let resp = plain::handle_plain(
req,
@@ -154,7 +163,6 @@ async fn serve_connection<I>(
timestamp_tolerance,
)
.await;
// plain::handle_plain already returns BoxBody (streaming)
Ok(resp)
}
}

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

@@ -25,10 +25,7 @@ pub fn ensure_self_signed_cert(cert_path: &Path, key_path: &Path) -> anyhow::Res
info!("generating self-signed TLS certificate");
let mut params = CertificateParams::new(vec![
"localhost".into(),
"aether-proxy".into(),
])?;
let mut params = CertificateParams::new(vec!["localhost".into(), "aether-proxy".into()])?;
params.distinguished_name = rcgen::DistinguishedName::new();
params
.distinguished_name
@@ -65,8 +62,8 @@ pub fn build_tls_acceptor(cert_path: &Path, key_path: &Path) -> anyhow::Result<T
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<_>, _>>()?;
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());
@@ -89,8 +86,7 @@ pub fn build_tls_acceptor(cert_path: &Path, key_path: &Path) -> anyhow::Result<T
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<_>, _>>()?;
rustls_pemfile::certs(&mut BufReader::new(cert_file)).collect::<Result<Vec<_>, _>>()?;
let cert = certs
.first()

53
aether-proxy/src/registration/client.rs
View File

@@ -3,6 +3,7 @@ use serde::{Deserialize, Serialize};
use tracing::{debug, error, info, warn};
use crate::config::Config;
use crate::hardware::HardwareInfo;
/// Heartbeat-specific error that distinguishes "node not found" (needs
/// re-registration) from transient / other failures.
@@ -35,6 +36,10 @@ struct RegisterRequest {
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>,
}
#[derive(Debug, Deserialize)]
@@ -56,6 +61,7 @@ struct HeartbeatRequest {
/// Remote configuration pushed by the Aether management backend.
#[derive(Debug, Clone, Deserialize)]
pub struct RemoteConfig {
pub node_name: Option<String>,
pub allowed_ports: Option<Vec<u16>>,
pub log_level: Option<String>,
pub heartbeat_interval: Option<u64>,
@@ -119,6 +125,7 @@ impl AetherClient {
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 {
@@ -129,6 +136,8 @@ impl AetherClient {
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),
};
info!(
@@ -215,10 +224,13 @@ impl AetherClient {
config_version,
}
}
Err(_) => HeartbeatResult {
remote_config: None,
config_version: 0,
},
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");
@@ -260,36 +272,3 @@ impl AetherClient {
}
}
}
/// Auto-detect public IP by querying external services.
pub async fn detect_public_ip() -> anyhow::Result<String> {
let endpoints = [
"https://api.ipify.org",
"https://ifconfig.me/ip",
"https://icanhazip.com",
];
let client = Client::builder()
.timeout(std::time::Duration::from_secs(5))
.build()?;
for endpoint in &endpoints {
match client.get(*endpoint).send().await {
Ok(resp) if resp.status().is_success() => {
let ip = resp.text().await?.trim().to_string();
if !ip.is_empty() {
info!(ip = %ip, source = %endpoint, "detected public IP");
return Ok(ip);
}
}
Ok(resp) => {
debug!(endpoint = %endpoint, status = %resp.status(), "IP detection failed");
}
Err(e) => {
debug!(endpoint = %endpoint, error = %e, "IP detection failed");
}
}
}
anyhow::bail!("failed to detect public IP from any source; use --public-ip")
}

55
aether-proxy/src/registration/heartbeat.rs
View File

@@ -1,11 +1,12 @@
use std::sync::{Arc, RwLock};
use std::sync::atomic::Ordering;
use std::sync::Arc;
use tokio::sync::watch;
use tracing::{debug, error, info, warn};
use crate::config::Config;
use crate::registration::client::{AetherClient, HeartbeatError};
use crate::runtime::{self, SharedDynamicConfig};
use crate::registration::client::HeartbeatError;
use crate::runtime;
use crate::state::AppState;
/// Run periodic heartbeat task until shutdown signal.
///
@@ -16,19 +17,11 @@ use crate::runtime::{self, SharedDynamicConfig};
/// 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(
client: Arc<AetherClient>,
node_id: Arc<RwLock<String>>,
config: Arc<Config>,
public_ip: String,
tls_fingerprint: Option<String>,
dynamic: SharedDynamicConfig,
mut shutdown_rx: watch::Receiver<bool>,
) {
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 = dynamic.read().unwrap().heartbeat_interval;
let initial_interval = state.dynamic.read().unwrap().heartbeat_interval;
tokio::select! {
_ = tokio::time::sleep(std::time::Duration::from_secs(initial_interval)) => {}
_ = shutdown_rx.changed() => {
@@ -38,12 +31,17 @@ pub async fn run(
}
loop {
let current_node_id = node_id.read().unwrap().clone();
let current_node_id = state.node_id.read().unwrap().clone();
let active_conns = state.active_connections.load(Ordering::Relaxed) as i64;
match client.heartbeat(&current_node_id, None, None, None).await {
match state
.aether_client
.heartbeat(&current_node_id, Some(active_conns), None, None)
.await
{
Ok(result) => {
if consecutive_failures > 0 {
debug!(
info!(
previous_failures = consecutive_failures,
"heartbeat recovered"
);
@@ -52,7 +50,7 @@ pub async fn run(
// Apply remote config if present and version changed
if let Some(ref remote) = result.remote_config {
runtime::apply_remote_config(&dynamic, remote, result.config_version);
runtime::apply_remote_config(&state.dynamic, remote, result.config_version);
}
}
Err(HeartbeatError::NodeNotFound(_)) => {
@@ -60,19 +58,24 @@ pub async fn run(
old_node_id = %current_node_id,
"node not found, re-registering"
);
match client.register(
&config,
&public_ip,
config.enable_tls,
tls_fingerprint.as_deref(),
).await {
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"
);
*node_id.write().unwrap() = new_id;
*state.node_id.write().unwrap() = new_id;
consecutive_failures = 0;
}
Err(e) => {
@@ -96,7 +99,7 @@ pub async fn run(
}
// Read interval from dynamic config (may have been updated remotely)
let interval_secs = dynamic.read().unwrap().heartbeat_interval;
let interval_secs = state.dynamic.read().unwrap().heartbeat_interval;
tokio::select! {
_ = tokio::time::sleep(std::time::Duration::from_secs(interval_secs)) => {}

11
aether-proxy/src/runtime.rs
View File

@@ -14,6 +14,7 @@ use crate::config::Config;
/// Configuration that can be changed at runtime without restart.
#[derive(Debug)]
pub struct DynamicConfig {
pub node_name: String,
pub allowed_ports: HashSet<u16>,
pub timestamp_tolerance: u64,
pub log_level: String,
@@ -27,6 +28,7 @@ impl DynamicConfig {
/// Initialize from static config (startup defaults).
pub fn from_config(config: &Config) -> Self {
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(),
@@ -54,7 +56,7 @@ pub fn set_log_reloader(f: Box<dyn Fn(&str) + Send + Sync>) {
/// Returns `true` if the config was actually changed.
pub fn apply_remote_config(
dynamic: &SharedDynamicConfig,
remote: &super::registration::client::RemoteConfig,
remote: &crate::registration::client::RemoteConfig,
version: u64,
) -> bool {
let mut cfg = dynamic.write().unwrap();
@@ -65,6 +67,13 @@ pub fn apply_remote_config(
let mut changed = Vec::new();
if let Some(ref name) = remote.node_name {
if *name != cfg.node_name {
changed.push(format!("node_name → {}", name));
cfg.node_name = name.clone();
}
}
if let Some(ref ports) = remote.allowed_ports {
let new_set: HashSet<u16> = ports.iter().copied().collect();
if new_set != cfg.allowed_ports {

4
aether-proxy/src/setup/mod.rs Normal file
View File

@@ -0,0 +1,4 @@
pub(crate) mod service;
mod tui;
pub use self::tui::run;

253
aether-proxy/src/setup/service.rs Normal file
View File

@@ -0,0 +1,253 @@
//! Systemd service installation for aether-proxy.
//!
//! Called from the setup TUI when the user enables "Install Service".
//! The unit file points to the binary and config at their current
//! absolute paths -- no files are copied.
use std::path::Path;
use std::process::Command;
const UNIT_PATH: &str = "/etc/systemd/system/aether-proxy.service";
const SERVICE_NAME: &str = "aether-proxy";
/// Whether systemd service installation is possible (systemd present + root).
pub fn is_available() -> bool {
is_systemd_available() && is_root()
}
/// Install aether-proxy as a systemd service. Must be run as root.
pub fn install_service(config_path: &Path) -> anyhow::Result<()> {
if !is_systemd_available() {
anyhow::bail!("systemd not available");
}
if !is_root() {
anyhow::bail!("root required, use: sudo aether-proxy setup");
}
let exe_path = std::env::current_exe()?.canonicalize()?;
let exe_str = exe_path
.to_str()
.ok_or_else(|| anyhow::anyhow!("binary path contains invalid UTF-8"))?;
let config_abs = std::fs::canonicalize(config_path)?;
let config_str = config_abs
.to_str()
.ok_or_else(|| anyhow::anyhow!("config path contains invalid UTF-8"))?;
let working_dir = config_abs
.parent()
.unwrap_or_else(|| Path::new("/"))
.to_str()
.unwrap_or("/");
// Stop existing service if running (ignore errors)
if Path::new(UNIT_PATH).exists() {
eprintln!(" Stopping existing service...");
let _ = Command::new("systemctl")
.args(["stop", SERVICE_NAME])
.status();
}
// Write unit file
eprintln!(" Generating systemd unit file...");
eprintln!(" Binary: {}", exe_str);
eprintln!(" Config: {}", config_str);
eprintln!(" WorkDir: {}", working_dir);
let unit_content = format!(
"[Unit]\n\
Description=Aether Proxy\n\
After=network.target\n\
\n\
[Service]\n\
Type=simple\n\
WorkingDirectory={working_dir}\n\
Environment=AETHER_PROXY_CONFIG={config_str}\n\
ExecStart={exe_str}\n\
Restart=on-failure\n\
RestartSec=5\n\
LimitNOFILE=65535\n\
\n\
[Install]\n\
WantedBy=multi-user.target\n",
);
std::fs::write(UNIT_PATH, &unit_content)?;
// Reload and enable
eprintln!(" Enabling and starting service...");
run_cmd("systemctl", &["daemon-reload"])?;
run_cmd("systemctl", &["enable", "--now", SERVICE_NAME])?;
// Verify
eprintln!();
let output = Command::new("systemctl")
.args(["is-active", SERVICE_NAME])
.output()?;
let state = String::from_utf8_lossy(&output.stdout).trim().to_string();
if state == "active" {
eprintln!(" Service started successfully!");
} else {
eprintln!(" Service state: {} (check logs)", state);
}
eprintln!();
eprintln!(" Commands:");
eprintln!(" sudo systemctl status {} # status", SERVICE_NAME);
eprintln!(" sudo systemctl restart {} # restart", SERVICE_NAME);
eprintln!(" sudo journalctl -u {} -f # logs", SERVICE_NAME);
eprintln!();
Ok(())
}
fn is_systemd_available() -> bool {
Command::new("systemctl")
.arg("--version")
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false)
}
fn is_root() -> bool {
#[cfg(unix)]
{
unsafe { libc::geteuid() == 0 }
}
#[cfg(not(unix))]
{
false
}
}
/// Whether a systemd unit file is currently installed.
pub fn is_installed() -> bool {
Path::new(UNIT_PATH).exists()
}
/// Remove the systemd service (called from setup TUI when Install Service is toggled off).
pub fn uninstall_service() -> anyhow::Result<()> {
if !Path::new(UNIT_PATH).exists() {
return Ok(());
}
eprintln!(" Stopping and removing existing service...");
let _ = Command::new("systemctl")
.args(["disable", "--now", SERVICE_NAME])
.status();
std::fs::remove_file(UNIT_PATH)?;
eprintln!(" Removed {}", UNIT_PATH);
run_cmd("systemctl", &["daemon-reload"])?;
eprintln!(" Service uninstalled.");
eprintln!();
Ok(())
}
/// Check if the systemd service is currently active.
pub fn is_service_active() -> bool {
std::path::Path::new(UNIT_PATH).exists()
&& Command::new("systemctl")
.args(["is-active", "--quiet", SERVICE_NAME])
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false)
}
// ── CLI subcommands (systemd wrappers) ──────────────────────────────────────
fn ensure_service_installed() -> anyhow::Result<()> {
if !std::path::Path::new(UNIT_PATH).exists() {
anyhow::bail!("service not installed, run `sudo aether-proxy setup` first");
}
Ok(())
}
fn ensure_root_and_service() -> anyhow::Result<()> {
ensure_service_installed()?;
if !is_root() {
anyhow::bail!("root required, use: sudo aether-proxy <command>");
}
Ok(())
}
/// `aether-proxy status` -- show service status.
pub fn cmd_status() -> anyhow::Result<()> {
ensure_service_installed()?;
let status = Command::new("systemctl")
.args(["status", SERVICE_NAME])
.status()?;
// systemctl status returns non-zero when inactive; that's fine
std::process::exit(status.code().unwrap_or(1));
}
/// `aether-proxy logs` -- tail service logs.
pub fn cmd_logs() -> anyhow::Result<()> {
ensure_service_installed()?;
let status = Command::new("journalctl")
.args(["-u", SERVICE_NAME, "-f", "--no-pager", "-n", "100"])
.status()?;
std::process::exit(status.code().unwrap_or(1));
}
/// `aether-proxy start` -- start the service.
pub fn cmd_start() -> anyhow::Result<()> {
ensure_root_and_service()?;
run_cmd("systemctl", &["start", SERVICE_NAME])?;
eprintln!(" Service started.");
Ok(())
}
/// `aether-proxy restart` -- restart the service.
pub fn cmd_restart() -> anyhow::Result<()> {
ensure_root_and_service()?;
run_cmd("systemctl", &["restart", SERVICE_NAME])?;
eprintln!(" Service restarted.");
Ok(())
}
/// `aether-proxy stop` -- stop the service.
pub fn cmd_stop() -> anyhow::Result<()> {
ensure_root_and_service()?;
run_cmd("systemctl", &["stop", SERVICE_NAME])?;
eprintln!(" Service stopped.");
Ok(())
}
/// `aether-proxy uninstall` -- disable and remove the systemd service.
pub fn cmd_uninstall() -> anyhow::Result<()> {
ensure_root_and_service()?;
eprintln!(" Stopping and disabling service...");
let _ = Command::new("systemctl")
.args(["disable", "--now", SERVICE_NAME])
.status();
if std::path::Path::new(UNIT_PATH).exists() {
std::fs::remove_file(UNIT_PATH)?;
eprintln!(" Removed {}", UNIT_PATH);
}
run_cmd("systemctl", &["daemon-reload"])?;
eprintln!(" Service uninstalled.");
eprintln!();
eprintln!(" Config file and TLS certs are preserved. Remove manually if needed.");
Ok(())
}
fn run_cmd(program: &str, args: &[&str]) -> anyhow::Result<()> {
let display = format!("{} {}", program, args.join(" "));
eprintln!(" > {}", display);
let status = Command::new(program).args(args).status()?;
if !status.success() {
anyhow::bail!("command failed: {}", display);
}
Ok(())
}

193
aether-proxy/src/setup.rs → aether-proxy/src/setup/tui.rs
View File

@@ -32,7 +32,6 @@ enum FieldKind {
Secret,
Number,
Bool,
PortList,
LogLevel,
}
@@ -102,14 +101,6 @@ impl App {
required: true,
help: "代理服务监听端口",
},
Field {
label: "Public IP",
key: "public_ip",
value: String::new(),
kind: FieldKind::Text,
required: false,
help: "节点公网 IP (留空则自动检测)",
},
Field {
label: "Node Name",
key: "node_name",
@@ -118,53 +109,13 @@ impl App {
required: true,
help: "节点名称，用于在 Aether 后台识别",
},
Field {
label: "Node Region",
key: "node_region",
value: String::new(),
kind: FieldKind::Text,
required: false,
help: "节点区域标识 (如 ap-northeast-1)",
},
Field {
label: "Heartbeat Interval",
key: "heartbeat_interval",
value: "30".into(),
kind: FieldKind::Number,
required: true,
help: "心跳上报间隔 (秒)",
},
Field {
label: "Allowed Ports",
key: "allowed_ports",
value: "80, 443, 8080, 8443".into(),
kind: FieldKind::PortList,
required: true,
help: "允许代理的目标端口，逗号分隔",
},
Field {
label: "Timestamp Tolerance",
key: "timestamp_tolerance",
value: "300".into(),
kind: FieldKind::Number,
required: true,
help: "HMAC 时间戳容差窗口 (秒)",
},
Field {
label: "Enable TLS",
key: "enable_tls",
value: "true".into(),
kind: FieldKind::Bool,
required: true,
help: "启用 TLS 加密 (双栈模式, 同时接受 HTTP 和 TLS)",
},
Field {
label: "Log Level",
key: "log_level",
value: "info".into(),
kind: FieldKind::LogLevel,
required: true,
help: "日志级别 — Enter 切换: trace / debug / info / warn / error",
help: "日志级别 -- Enter 切换: trace / debug / info / warn / error",
},
Field {
label: "Log JSON",
@@ -172,7 +123,20 @@ impl App {
value: "false".into(),
kind: FieldKind::Bool,
required: true,
help: "是否以 JSON 格式输出日志 — Enter 切换",
help: "是否以 JSON 格式输出日志 -- Enter 切换",
},
Field {
label: "Install Service",
key: "install_service",
value: if super::service::is_available() {
"true"
} else {
"false"
}
.into(),
kind: FieldKind::Bool,
required: true,
help: "注册为 systemd 开机启动服务 (需要 root 权限) -- Enter 切换",
},
],
selected: 0,
@@ -202,17 +166,9 @@ impl App {
"management_token" => cfg.management_token.clone(),
"hmac_key" => cfg.hmac_key.clone(),
"listen_port" => cfg.listen_port.map(|v| v.to_string()),
"public_ip" => cfg.public_ip.clone(),
"node_name" => cfg.node_name.clone(),
"node_region" => cfg.node_region.clone(),
"heartbeat_interval" => cfg.heartbeat_interval.map(|v| v.to_string()),
"allowed_ports" => cfg.allowed_ports.as_ref().map(|p| {
p.iter().map(|v| v.to_string()).collect::<Vec<_>>().join(", ")
}),
"timestamp_tolerance" => cfg.timestamp_tolerance.map(|v| v.to_string()),
"log_level" => cfg.log_level.clone(),
"log_json" => cfg.log_json.map(|v| v.to_string()),
"enable_tls" => cfg.enable_tls.map(|v| v.to_string()),
_ => None,
};
if let Some(v) = val {
@@ -235,19 +191,15 @@ impl App {
management_token: get("management_token"),
hmac_key: get("hmac_key"),
listen_port: get("listen_port").and_then(|v| v.parse().ok()),
public_ip: get("public_ip"),
public_ip: None,
node_name: get("node_name"),
node_region: get("node_region"),
heartbeat_interval: get("heartbeat_interval").and_then(|v| v.parse().ok()),
allowed_ports: get("allowed_ports").map(|v| {
v.split(',')
.filter_map(|s| s.trim().parse().ok())
.collect()
}),
timestamp_tolerance: get("timestamp_tolerance").and_then(|v| v.parse().ok()),
node_region: None,
heartbeat_interval: None,
allowed_ports: None,
timestamp_tolerance: None,
log_level: get("log_level"),
log_json: get("log_json").and_then(|v| v.parse().ok()),
enable_tls: get("enable_tls").and_then(|v| v.parse().ok()),
enable_tls: None,
tls_cert: None,
tls_key: None,
}
@@ -259,7 +211,7 @@ impl App {
self.modified = false;
self.saved_once = true;
self.message = Some((
format!("✓ 已保存到 {}", self.config_path.display()),
format!("saved to {}", self.config_path.display()),
Instant::now(),
false,
));
@@ -304,7 +256,7 @@ impl App {
KeyCode::Char('q') | KeyCode::Esc => return true,
KeyCode::Char('s') if key.modifiers.contains(KeyModifiers::CONTROL) => {
if let Err(e) = self.save() {
self.message = Some((format!("✗ {}", e), Instant::now(), true));
self.message = Some((format!("error: {}", e), Instant::now(), true));
}
}
KeyCode::Up | KeyCode::Char('k') => {
@@ -321,16 +273,30 @@ impl App {
let field = &self.fields[self.selected];
match field.kind {
FieldKind::Bool => {
let toggled = if field.value == "true" { "false" } else { "true" };
self.fields[self.selected].value = toggled.into();
self.modified = true;
let toggled = if field.value == "true" {
"false"
} else {
"true"
};
// Block enabling service install without root/systemd
if field.key == "install_service"
&& toggled == "true"
&& !super::service::is_available()
{
self.message = Some((
"requires root with systemd, use: sudo aether-proxy setup".into(),
Instant::now(),
true,
));
} else {
self.fields[self.selected].value = toggled.into();
self.modified = true;
}
}
FieldKind::LogLevel => {
const LEVELS: &[&str] =
&["trace", "debug", "info", "warn", "error"];
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();
self.fields[self.selected].value = LEVELS[(idx + 1) % LEVELS.len()].into();
self.modified = true;
}
_ => {
@@ -343,7 +309,7 @@ impl App {
KeyCode::Tab => {
// Quick save shortcut
if let Err(e) = self.save() {
self.message = Some((format!("✗ {}", e), Instant::now(), true));
self.message = Some((format!("error: {}", e), Instant::now(), true));
}
}
_ => {}
@@ -354,7 +320,7 @@ impl App {
fn handle_edit(&mut self, key: KeyEvent) {
match key.code {
KeyCode::Esc => {
// Cancel — discard changes to this field
// Cancel -- discard changes to this field
self.mode = Mode::Normal;
}
KeyCode::Enter => {
@@ -363,8 +329,7 @@ impl App {
self.modified = true;
self.mode = Mode::Normal;
} else {
self.message =
Some(("✗ 格式无效".into(), Instant::now(), true));
self.message = Some(("invalid format".into(), Instant::now(), true));
}
}
KeyCode::Backspace => {
@@ -405,12 +370,6 @@ impl App {
let buf = &self.edit_buffer;
match kind {
FieldKind::Number => buf.is_empty() || buf.parse::<u64>().is_ok(),
FieldKind::PortList => {
buf.is_empty()
|| buf
.split(',')
.all(|s| s.trim().is_empty() || s.trim().parse::<u16>().is_ok())
}
_ => true,
}
}
@@ -468,7 +427,7 @@ fn render_fields(f: &mut Frame, app: &mut App, area: Rect) {
}
let selected = i == app.selected;
let indicator = if selected { " ▸ " } else { " " };
let indicator = if selected { " > " } else { " " };
let label_style = if selected {
Style::default()
@@ -482,10 +441,7 @@ fn render_fields(f: &mut Frame, app: &mut App, area: Rect) {
// Value display
let (value_text, value_style) = if app.mode == Mode::Editing && selected {
(
app.edit_buffer.clone(),
Style::default().fg(Color::Yellow),
)
(app.edit_buffer.clone(), Style::default().fg(Color::Yellow))
} else {
field_display(field)
};
@@ -518,9 +474,9 @@ fn render_fields(f: &mut Frame, app: &mut App, area: Rect) {
fn field_display(field: &Field) -> (String, Style) {
if field.value.is_empty() {
let text = if field.required {
"(必填)".into()
"(required)".into()
} else {
"—".into()
"-".into()
};
let color = if field.required {
Color::Red
@@ -532,14 +488,14 @@ fn field_display(field: &Field) -> (String, Style) {
match field.kind {
FieldKind::Secret => (
"•".repeat(field.value.len().min(20)),
"*".repeat(field.value.len().min(20)),
Style::default().fg(Color::White),
),
FieldKind::Bool => {
if field.value == "true" {
("✓ 开启".into(), Style::default().fg(Color::Green))
("[x] on".into(), Style::default().fg(Color::Green))
} else {
("✗ 关闭".into(), Style::default().fg(Color::DarkGray))
("[ ] off".into(), Style::default().fg(Color::DarkGray))
}
}
FieldKind::LogLevel => {
@@ -561,9 +517,9 @@ fn render_footer(f: &mut Frame, app: &App, area: Rect) {
let help = app.fields[app.selected].help;
let keybindings = if app.mode == Mode::Editing {
"Enter 确认 Esc 取消"
"Enter confirm Esc cancel"
} else {
"↑↓ 选择 Enter 编辑 ^S 保存 q 退出"
"Up/Down select Enter edit ^S save q quit"
};
let mut status_spans: Vec<Span> = vec![Span::styled(
@@ -631,11 +587,40 @@ pub fn run(config_path: PathBuf) -> anyhow::Result<()> {
// Post-TUI message
if app.saved_once {
eprintln!();
eprintln!(" 配置已保存到 {}", config_path.display());
eprintln!();
eprintln!(" 启动方式:");
eprintln!(" aether-proxy (自动读取 {})", config_path.display());
eprintln!(" Config saved to {}", config_path.display());
eprintln!();
let wants_service = app
.fields
.iter()
.find(|f| f.key == "install_service")
.map(|f| f.value == "true")
.unwrap_or(false);
if wants_service {
match super::service::install_service(&config_path) {
Ok(()) => {}
Err(e) => {
eprintln!(" Service install failed: {}", e);
eprintln!();
}
}
} else {
// Uninstall service if it was previously installed
if super::service::is_installed() {
if let Err(e) = super::service::uninstall_service() {
eprintln!(" Service uninstall failed: {}", e);
eprintln!();
}
}
eprintln!(" Run with:");
eprintln!(
" aether-proxy (auto-reads {})",
config_path.display()
);
eprintln!();
}
}
Ok(())

30
aether-proxy/src/state.rs Normal file
View File

@@ -0,0 +1,30 @@
//! 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;
use std::sync::{Arc, RwLock};
use tokio_rustls::TlsAcceptor;
use crate::config::Config;
use crate::hardware::HardwareInfo;
use crate::registration::client::AetherClient;
use crate::runtime::SharedDynamicConfig;
/// Central application state shared across all tasks.
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 reqwest client for delegate mode (proxy issues upstream requests directly).
pub delegate_client: reqwest::Client,
/// Active connection count for metrics reporting.
pub active_connections: Arc<AtomicU64>,
}