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

refactor(proxy): 简化 HMAC 认证、移除明文 HTTP 代理并优化 setup 流程

Browse Source 
- HMAC 签名移除 node_id，仅使用 timestamp，消除重注册时的认证竞态问题
- 删除 plain HTTP forward proxy（plain.rs），仅保留 CONNECT 隧道和 delegate，非支持方法返回 405
- 提取共享 BoxBody 类型和 empty_box_body() 到 proxy/mod.rs
- CLI 解析重构为 clap 原生 subcommand，启用 subcommand_negates_reqs
- setup 向导返回 SetupOutcome 枚举，支持保存后自动启动 proxy
- setup TUI 增加未保存变更的退出确认（pending_quit）
- validate_target 改为 async，使用 tokio::net::lookup_host 避免阻塞 DNS
- 显式初始化 rustls ring CryptoProvider
- ConfigFile 新增 inject_env_override() 用于 setup 后重载配置
- Python 端 HMAC 签名同步移除 node_id，缓存时间桶从 120s 调整为 240s
This commit is contained in:
fawney19
2026-02-08 16:10:13 +08:00
parent f07cae540e
commit 183e7c60e1
15 changed files with 261 additions and 357 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,7 +16,6 @@ use crate::proxy::target_filter;
pub async fn handle_connect(
req: Request<Incoming>,
config: Arc<Config>,
node_id: &str,
allowed_ports: &HashSet<u16>,
timestamp_tolerance: u64,
) -> Response<http_body_util::Empty<bytes::Bytes>> {
@@ -27,7 +26,7 @@ pub async fn handle_connect(
.and_then(|v| v.to_str().ok());
// HMAC authentication
if let Err(e) = auth::validate_proxy_auth(proxy_auth, &config, node_id, timestamp_tolerance) {
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());
}
@@ -45,7 +44,7 @@ pub async fn handle_connect(
let port = authority.port_u16().unwrap_or(443);
// Target filter: private IP + port whitelist
let target_addr = match target_filter::validate_target(&host, port, allowed_ports) {
let target_addr = match target_filter::validate_target(&host, port, allowed_ports).await {
Ok(addr) => addr,
Err(e) => {
warn!(host = %host, port, error = %e, "CONNECT target rejected");

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

@@ -10,9 +10,9 @@ use serde::Deserialize;
use tracing::{debug, warn};
use url::Url;
use super::BoxBody;
use crate::auth;
use crate::config::Config;
use crate::proxy::plain::BoxBody;
use crate::proxy::target_filter;
/// Delegation request payload sent by Aether.
@@ -34,7 +34,6 @@ struct DelegateRequest {
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,
@@ -45,7 +44,7 @@ pub async fn handle_delegate(
.get("authorization")
.and_then(|v| v.to_str().ok());
if let Err(e) = auth::validate_proxy_auth(auth_header, &config, node_id, timestamp_tolerance) {
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());
}
@@ -87,7 +86,7 @@ pub async fn handle_delegate(
let port = parsed_url.port_or_known_default().unwrap_or(443);
if let Err(e) = target_filter::validate_target(&host, port, allowed_ports) {
if let Err(e) = target_filter::validate_target(&host, port, allowed_ports).await {
warn!(host = %host, port, error = %e, "delegate target rejected");
return error_response(403, "target_not_allowed", &e.to_string());
}
@@ -165,7 +164,7 @@ pub async fn handle_delegate(
builder
.body(stream_body)
.unwrap_or_else(|_| Response::builder().status(500).body(empty_box()).unwrap())
.unwrap_or_else(|_| Response::builder().status(500).body(super::empty_box_body()).unwrap())
}
// ── Sanitisation ─────────────────────────────────────────────────────────────
@@ -195,12 +194,6 @@ fn sanitize_upstream_error(msg: &str) -> String {
// ── 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,

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

@@ -1,6 +1,18 @@
pub mod connect;
pub mod delegate;
pub mod plain;
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()
}

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

@@ -1,179 +0,0 @@
use std::collections::HashSet;
use std::sync::Arc;
use http_body_util::{BodyExt, Full};
use hyper::body::Incoming;
use hyper::{Request, Response};
use tracing::{debug, warn};
use crate::auth;
use crate::config::Config;
use crate::proxy::target_filter;
/// Boxed body that unifies `Full` (error responses) and `Incoming` (streamed upstream).
pub type BoxBody =
http_body_util::combinators::BoxBody<bytes::Bytes, Box<dyn std::error::Error + Send + Sync>>;
/// Handle plain HTTP forward proxy requests (non-CONNECT).
///
/// Flow: validate auth -> check target filter -> forward request -> **stream** response
pub async fn handle_plain(
req: Request<Incoming>,
config: Arc<Config>,
node_id: &str,
allowed_ports: &HashSet<u16>,
timestamp_tolerance: u64,
) -> Response<BoxBody> {
// 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, node_id, timestamp_tolerance) {
warn!(error = %e, "HTTP proxy auth failed");
return proxy_auth_required(&e.to_string());
}
// Parse target from absolute URI
let uri = req.uri().clone();
let host = match uri.host() {
Some(h) => h.to_string(),
None => {
warn!(uri = %uri, "HTTP proxy request missing host");
return bad_request("missing host in URI");
}
};
let port = uri.port_u16().unwrap_or(80);
// Target filter
let target_addr = match target_filter::validate_target(&host, port, allowed_ports) {
Ok(addr) => addr,
Err(e) => {
warn!(host = %host, port, error = %e, "HTTP proxy target rejected");
return forbidden(&e.to_string());
}
};
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 mut builder = Request::builder()
.method(req.method())
.uri(path_and_query)
.version(req.version());
// Copy headers, skipping proxy-specific and forwarding-related ones
for (name, value) in req.headers() {
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);
}
// Collect the incoming request body (client payloads are small)
let body_bytes = match req.into_body().collect().await {
Ok(collected) => collected.to_bytes(),
Err(e) => {
warn!(error = %e, "failed to read request body");
return bad_gateway("failed to read request body");
}
};
// Connect and send via raw TCP + hyper client
let stream = match tokio::net::TcpStream::connect(target_addr).await {
Ok(s) => s,
Err(e) => {
warn!(target = %target_addr, error = %e, "HTTP proxy connection failed");
return bad_gateway(&format!("connection failed: {}", e));
}
};
let io = hyper_util::rt::TokioIo::new(stream);
let (mut sender, conn) = match hyper::client::conn::http1::handshake(io).await {
Ok(pair) => pair,
Err(e) => {
warn!(error = %e, "HTTP handshake failed");
return bad_gateway(&format!("handshake failed: {}", e));
}
};
tokio::task::spawn(async move {
if let Err(e) = conn.await {
debug!(error = %e, "HTTP proxy client connection error");
}
});
let outgoing = builder
.body(Full::new(body_bytes))
.expect("failed to build outgoing request");
match sender.send_request(outgoing).await {
Ok(resp) => {
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
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { Box::new(e) })
.boxed();
Response::from_parts(parts, body)
}
Err(e) => {
warn!(error = %e, "HTTP proxy request failed");
bad_gateway(&format!("upstream request failed: {}", e))
}
}
}
// ── 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 proxy_auth_required(msg: &str) -> Response<BoxBody> {
Response::builder()
.status(407)
.header("Proxy-Authenticate", "HMAC-SHA256")
.header("X-Error", msg)
.body(empty_box())
.unwrap()
}
fn forbidden(msg: &str) -> Response<BoxBody> {
Response::builder()
.status(403)
.header("X-Error", msg)
.body(empty_box())
.unwrap()
}
fn bad_request(msg: &str) -> Response<BoxBody> {
Response::builder()
.status(400)
.header("X-Error", msg)
.body(empty_box())
.unwrap()
}
fn bad_gateway(msg: &str) -> Response<BoxBody> {
Response::builder()
.status(502)
.header("X-Error", msg)
.body(empty_box())
.unwrap()
}

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

@@ -7,20 +7,21 @@ 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::{Method, Request, Response};
use hyper_util::rt::TokioIo;
use tokio::net::TcpListener;
use tokio::sync::watch;
use tracing::{debug, info, warn};
use crate::proxy::{connect, delegate, plain, tls};
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
/// - Other HTTP requests -> plain forward proxy 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
@@ -104,38 +105,24 @@ 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>,
>;
// Snapshot current dynamic values (may be updated by remote config)
let current_node_id = node_id.read().unwrap().clone();
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,
&current_node_id,
&allowed_ports,
timestamp_tolerance,
)
.await;
let resp =
connect::handle_connect(req, config, &allowed_ports, timestamp_tolerance).await;
let resp = resp.map(|_| -> BoxBody {
http_body_util::Empty::new()
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { match e {} })
@@ -147,7 +134,6 @@ where
let resp = delegate::handle_delegate(
req,
config,
&current_node_id,
&allowed_ports,
timestamp_tolerance,
&delegate_client,
@@ -155,15 +141,14 @@ where
.await;
Ok(resp)
} else {
let resp = plain::handle_plain(
req,
config,
&current_node_id,
&allowed_ports,
timestamp_tolerance,
)
.await;
Ok(resp)
// Only CONNECT tunnels and /_aether/delegate are supported;
// plain HTTP forward proxy was removed (all API traffic is HTTPS).
Ok(Response::builder()
.status(405)
.header("Allow", "CONNECT")
.header("Content-Length", "0")
.body(crate::proxy::empty_box_body())
.unwrap())
}
}
});

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

@@ -1,5 +1,5 @@
use std::collections::HashSet;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, ToSocketAddrs};
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
/// Check if an IP address belongs to a private/reserved network.
fn is_private_ip(ip: &IpAddr) -> bool {
@@ -82,8 +82,11 @@ impl std::fmt::Display for FilterError {
/// Validate that the target host:port is allowed.
///
/// Uses async DNS resolution (via `tokio::net::lookup_host`) to avoid
/// blocking the async runtime on potentially slow DNS lookups.
///
/// Returns the resolved socket address to connect to.
pub fn validate_target(
pub async fn validate_target(
host: &str,
port: u16,
allowed_ports: &HashSet<u16>,
@@ -93,7 +96,7 @@ pub fn validate_target(
return Err(FilterError::PortNotAllowed(port));
}
// Try parsing as IP directly
// Try parsing as IP directly (no DNS needed)
if let Ok(ip) = host.parse::<IpAddr>() {
if is_private_ip(&ip) {
return Err(FilterError::PrivateIp(ip));
@@ -101,10 +104,10 @@ pub fn validate_target(
return Ok(SocketAddr::new(ip, port));
}
// DNS resolution with private IP check (DNS rebinding protection)
// Async DNS resolution with private IP check (DNS rebinding protection)
let addr_str = format!("{}:{}", host, port);
let addrs: Vec<SocketAddr> = addr_str
.to_socket_addrs()
let addrs: Vec<SocketAddr> = tokio::net::lookup_host(&addr_str)
.await
.map_err(|_| FilterError::DnsResolutionFailed(host.to_string()))?
.collect();
@@ -157,21 +160,21 @@ mod tests {
))));
}
#[test]
fn test_port_not_allowed() {
let result = validate_target("8.8.8.8", 22, &ports());
#[tokio::test]
async fn test_port_not_allowed() {
let result = validate_target("8.8.8.8", 22, &ports()).await;
assert!(matches!(result, Err(FilterError::PortNotAllowed(22))));
}
#[test]
fn test_private_ip_blocked() {
let result = validate_target("127.0.0.1", 80, &ports());
#[tokio::test]
async fn test_private_ip_blocked() {
let result = validate_target("127.0.0.1", 80, &ports()).await;
assert!(matches!(result, Err(FilterError::PrivateIp(_))));
}
#[test]
fn test_public_ip_allowed() {
let result = validate_target("8.8.8.8", 443, &ports());
#[tokio::test]
async fn test_public_ip_allowed() {
let result = validate_target("8.8.8.8", 443, &ports()).await;
assert!(result.is_ok());
}
}