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

feat: aether-proxy TLS 双栈支持与自签名证书自动生成

Browse Source 
- aether-proxy 新增 TLS 模块：自签名证书生成、TLS acceptor 构建、证书 SHA-256 指纹计算
- 代理服务器支持 HTTP+TLS 双栈模式，通过 peek 首字节区分 TLS ClientHello 与普通 HTTP
- 注册与心跳上报 tls_enabled 和 tls_cert_fingerprint 字段
- Python 侧 httpx 代理适配：TLS 代理使用 httpx.Proxy + CERT_NONE ssl_context
- ProxyNode 模型新增 tls_enabled/tls_cert_fingerprint 字段及对应迁移
This commit is contained in:
fawney19
2026-02-07 23:18:58 +08:00
parent 10bd14c223
commit 5384ffd403
17 changed files with 588 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -58,6 +58,18 @@ pub struct Config {
/// Output logs as JSON
#[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
#[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")]
pub tls_key: String,
}
// ---------------------------------------------------------------------------
@@ -92,6 +104,12 @@ pub struct ConfigFile {
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>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tls_cert: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub tls_key: Option<String>,
}
impl ConfigFile {
@@ -133,6 +151,9 @@ impl ConfigFile {
set!("AETHER_PROXY_TIMESTAMP_TOLERANCE", self.timestamp_tolerance);
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);
// allowed_ports needs special handling (comma-separated)
if let Some(ref ports) = self.allowed_ports {

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

@@ -73,7 +73,26 @@ async fn main() -> anyhow::Result<()> {
// Register with Aether
let aether_client = Arc::new(AetherClient::new(&config));
let node_id = aether_client.register(&config, &public_ip).await?;
// 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");
@@ -94,9 +113,10 @@ async fn main() -> anyhow::Result<()> {
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, dynamic, rx).await;
registration::heartbeat::run(client, node_id, config, public_ip, fingerprint, dynamic, rx).await;
})
};
@@ -106,8 +126,9 @@ async fn main() -> anyhow::Result<()> {
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, rx).await {
if let Err(e) = proxy::server::run(config, node_id, dynamic, tls, rx).await {
error!(error = %e, "proxy server error");
}
})

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

@@ -2,3 +2,4 @@ pub mod connect;
pub mod plain;
pub mod server;
pub mod target_filter;
pub mod tls;

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

@@ -6,13 +6,15 @@ use hyper::body::Incoming;
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};
use crate::proxy::{connect, plain, tls};
use crate::runtime::SharedDynamicConfig;
/// Start the proxy server.
@@ -20,15 +22,25 @@ use crate::runtime::SharedDynamicConfig;
/// Listens for incoming TCP connections and dispatches:
/// - CONNECT requests -> tunnel handler
/// - Other HTTP requests -> plain forward proxy handler
///
/// When `tls_acceptor` is provided, 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>,
mut shutdown_rx: watch::Receiver<bool>,
) -> anyhow::Result<()> {
let addr = SocketAddr::from(([0, 0, 0, 0], config.listen_port));
let listener = TcpListener::bind(addr).await?;
info!(addr = %addr, "proxy server listening");
if tls_acceptor.is_some() {
info!(addr = %addr, "proxy server listening (HTTP+TLS dual-stack)");
} else {
info!(addr = %addr, "proxy server listening (HTTP only)");
}
loop {
tokio::select! {
@@ -46,66 +58,41 @@ pub async fn run(
let config = Arc::clone(&config);
let node_id = Arc::clone(&node_id);
let dynamic = Arc::clone(&dynamic);
let tls_acceptor = tls_acceptor.clone();
tokio::task::spawn(async move {
let io = TokioIo::new(stream);
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);
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 = resp.map(|_| -> BoxBody {
http_body_util::Empty::new()
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { match e {} })
.boxed()
});
Ok::<_, hyper::Error>(resp)
} else {
let resp = plain::handle_plain(
req,
config,
&current_node_id,
&allowed_ports,
timestamp_tolerance,
)
.await;
// plain::handle_plain already returns BoxBody (streaming)
Ok(resp)
// Dual-stack: peek first byte to decide TLS vs plain HTTP
if let Some(acceptor) = &tls_acceptor {
if tls::is_tls_client_hello(&stream).await {
match acceptor.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,
)
.await;
}
Err(e) => {
debug!(peer = %peer_addr, error = %e, "TLS handshake failed");
}
}
}
});
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");
return;
}
}
// Plain HTTP
serve_connection(
TokioIo::new(stream),
peer_addr,
config,
node_id,
dynamic,
)
.await;
});
}
_ = shutdown_rx.changed() => {
@@ -117,3 +104,71 @@ pub async fn run(
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,
config: Arc<Config>,
node_id: Arc<RwLock<String>>,
dynamic: SharedDynamicConfig,
) where
I: Read + Write + Unpin + Send + 'static,
{
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);
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 = resp.map(|_| -> BoxBody {
http_body_util::Empty::new()
.map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { match e {} })
.boxed()
});
Ok::<_, hyper::Error>(resp)
} else {
let resp = plain::handle_plain(
req,
config,
&current_node_id,
&allowed_ports,
timestamp_tolerance,
)
.await;
// plain::handle_plain already returns BoxBody (streaming)
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");
}
}
}

116
aether-proxy/src/proxy/tls.rs Normal file
View File

@@ -0,0 +1,116 @@
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};
/// 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 config = rustls::ServerConfig::builder()
.with_no_client_auth()
.with_single_cert(certs, key)?;
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
}
}
}

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

@@ -31,6 +31,10 @@ 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>,
}
#[derive(Debug, Deserialize)]
@@ -113,6 +117,8 @@ impl AetherClient {
&self,
config: &Config,
public_ip: &str,
tls_enabled: bool,
tls_cert_fingerprint: Option<&str>,
) -> anyhow::Result<String> {
let url = format!("{}/api/admin/proxy-nodes/register", self.base_url);
let body = RegisterRequest {
@@ -121,6 +127,8 @@ impl AetherClient {
port: config.listen_port,
region: config.node_region.clone(),
heartbeat_interval: config.heartbeat_interval,
tls_enabled,
tls_cert_fingerprint: tls_cert_fingerprint.map(|s| s.to_string()),
};
info!(

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

@@ -21,6 +21,7 @@ pub async fn run(
node_id: Arc<RwLock<String>>,
config: Arc<Config>,
public_ip: String,
tls_fingerprint: Option<String>,
dynamic: SharedDynamicConfig,
mut shutdown_rx: watch::Receiver<bool>,
) {
@@ -59,7 +60,12 @@ pub async fn run(
old_node_id = %current_node_id,
"node not found, re-registering"
);
match client.register(&config, &public_ip).await {
match client.register(
&config,
&public_ip,
config.enable_tls,
tls_fingerprint.as_deref(),
).await {
Ok(new_id) => {
info!(
old_node_id = %current_node_id,

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

@@ -150,6 +150,14 @@ impl App {
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",
@@ -204,6 +212,7 @@ impl App {
"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 {
@@ -238,6 +247,9 @@ impl App {
timestamp_tolerance: get("timestamp_tolerance").and_then(|v| v.parse().ok()),
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()),
tls_cert: None,
tls_key: None,
}
}