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

Merge remote-tracking branch 'origin/pr/535'

Browse Source
This commit is contained in:
fawney19
2026-05-23 13:00:18 +08:00
parent ab048b8a03 9dd545353c
commit 74f7348529
34 changed files with 1173 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
apps/aether-gateway/Cargo.toml
View File

@@ -31,7 +31,7 @@ aether-task-runtime.workspace = true
aether-usage-runtime.workspace = true
aether-video-tasks-core.workspace = true
aether-wallet.workspace = true
aes-gcm = "0.10"
aes-gcm.workspace = true
async-stream.workspace = true
async-trait.workspace = true
axum = { version = "0.8", features = ["ws"] }

1
apps/aether-gateway/src/api/backend/public.rs
View File

@@ -26,7 +26,6 @@ pub(crate) fn mount_public_support_routes(router: Router<AppState>) -> Router<Ap
.route("/api/capabilities/model/{*model_path}", get(proxy_request))
.route("/install/{*install_path}", get(proxy_request))
.route("/install-tunnel/{*install_path}", get(proxy_request))
.route("/install-proxy/{*install_path}", get(proxy_request))
.route("/i/{*install_path}", get(proxy_request))
.route("/", get(proxy_request))
}

1
apps/aether-gateway/src/control/route/public_support.rs
View File

@@ -797,7 +797,6 @@ pub(super) fn classify_public_support_route(
} else if method == http::Method::GET
&& (has_single_segment_after_prefix(normalized_path, "/install/")
|| has_single_segment_after_prefix(normalized_path, "/install-tunnel/")
|| has_single_segment_after_prefix(normalized_path, "/install-proxy/")
|| has_single_segment_after_prefix(normalized_path, "/i/"))
{
Some(classified(

49
apps/aether-gateway/src/handlers/admin/system/proxy_nodes.rs
View File

@@ -63,6 +63,10 @@ struct ProxyNodeRegisterRequest {
proxy_version: Option<String>,
#[serde(default)]
tunnel_mode: Option<bool>,
#[serde(default)]
tunnel_security: Option<String>,
#[serde(default)]
tunnel_encryption_key: Option<String>,
}
#[derive(Debug, Deserialize)]
@@ -349,9 +353,21 @@ pub(crate) async fn maybe_build_local_admin_proxy_nodes_response(
Ok(mutation) => mutation,
Err(response) => return Ok(Some(response)),
};
let tunnel_encryption_key = mutation
.proxy_metadata
.as_ref()
.and_then(|metadata| metadata.pointer("/tunnel_security/encryption_key"))
.and_then(|value| value.as_str())
.map(str::to_string);
let Some(node) = state.register_proxy_node(&mutation).await? else {
return Ok(Some(build_admin_proxy_nodes_data_unavailable_response()));
};
if let Some(key) = tunnel_encryption_key {
state
.app()
.tunnel
.register_secure_tunnel_key(node.id.clone(), key);
}
return Ok(Some(
Json(json!({
"node_id": node.id,
@@ -1420,12 +1436,43 @@ fn validate_register_request(
}
validate_optional_object(input.hardware_info.as_ref(), "hardware_info")?;
validate_optional_object(input.proxy_metadata.as_ref(), "proxy_metadata")?;
let tunnel_security =
normalize_optional_string(input.tunnel_security.as_deref(), "tunnel_security", 64)?;
let tunnel_encryption_key = normalize_optional_string(
input.tunnel_encryption_key.as_deref(),
"tunnel_encryption_key",
128,
)?;
let registered_by = request_context
.decision()
.and_then(|decision| decision.admin_principal.as_ref())
.map(|principal| principal.user_id.clone());
let mut proxy_metadata = input.proxy_metadata;
if tunnel_security.as_deref()
== Some(aether_contracts::tunnel_security::TUNNEL_SECURITY_NON_TLS_REQUIRED)
{
let key = tunnel_encryption_key.as_deref().ok_or_else(|| {
bad_request_response(
"tunnel_encryption_key is required when tunnel_security=non_tls_required",
)
})?;
aether_contracts::tunnel_security::decode_psk(key)
.map_err(|err| bad_request_response(err.to_string()))?;
let mut metadata = proxy_metadata
.and_then(|value| value.as_object().cloned())
.unwrap_or_default();
metadata.insert(
"tunnel_security".to_string(),
json!({
"mode": aether_contracts::tunnel_security::TUNNEL_SECURITY_NON_TLS_REQUIRED,
"encryption_key": key,
}),
);
proxy_metadata = Some(Value::Object(metadata));
}
Ok(
aether_data::repository::proxy_nodes::ProxyNodeRegistrationMutation {
name,
@@ -1438,7 +1485,7 @@ fn validate_register_request(
avg_latency_ms: input.avg_latency_ms,
hardware_info: input.hardware_info,
estimated_max_concurrency: input.estimated_max_concurrency,
proxy_metadata: input.proxy_metadata,
proxy_metadata,
proxy_version: normalize_optional_string(
input.proxy_version.as_deref(),
"proxy_version",

44
apps/aether-gateway/src/handlers/public/support/install.rs
View File

@@ -16,7 +16,7 @@ const INSTALL_SESSION_TTL_SECS: u64 = 15 * 60;
const INSTALL_SESSION_KEY_PREFIX: &str = "install:session:";
const TUNNEL_INSTALL_SESSION_KEY_PREFIX: &str = "tunnel-install:session:";
const TUNNEL_INSTALL_UNIX_SCRIPT_URL: &str =
"https://raw.githubusercontent.com/fawney19/Aether/main/apps/aether-tunnel/install.sh";
"https://raw.githubusercontent.com/fawney19/Aether/refs/heads/main/apps/aether-tunnel/install.sh";
const TUNNEL_INSTALL_POWERSHELL_SCRIPT_URL: &str =
"https://raw.githubusercontent.com/fawney19/Aether/main/apps/aether-tunnel/install.ps1";
@@ -59,6 +59,8 @@ struct StoredTunnelInstallSession {
aether_url: String,
management_token: String,
node_name: String,
tunnel_security: String,
tunnel_encryption_key: String,
expires_at_unix_secs: u64,
}
@@ -93,8 +95,7 @@ fn install_code_from_path(request_path: &str) -> Option<(String, bool)> {
fn tunnel_install_code_from_path(request_path: &str) -> Option<(String, bool)> {
let raw = request_path
.strip_prefix("/install-tunnel/")
.or_else(|| request_path.strip_prefix("/install-proxy/"))?
.strip_prefix("/install-tunnel/")?
.trim()
.trim_matches('/');
if raw.is_empty() || raw.contains('/') {
@@ -122,6 +123,17 @@ fn generate_install_code() -> String {
.collect()
}
fn generate_tunnel_encryption_key() -> String {
use base64::Engine;
let first = uuid::Uuid::new_v4();
let second = uuid::Uuid::new_v4();
let mut key = [0_u8; 32];
key[..16].copy_from_slice(first.as_bytes());
key[16..].copy_from_slice(second.as_bytes());
base64::engine::general_purpose::STANDARD.encode(key)
}
fn unix_secs_now() -> u64 {
chrono::Utc::now().timestamp().max(0) as u64
}
@@ -172,6 +184,8 @@ set -eu
export AETHER_TUNNEL_AETHER_URL={aether_url}
export AETHER_TUNNEL_MANAGEMENT_TOKEN={management_token}
export AETHER_TUNNEL_NODE_NAME={node_name}
export AETHER_TUNNEL_SECURITY={tunnel_security}
export AETHER_TUNNEL_ENCRYPTION_KEY={tunnel_encryption_key}
if command -v curl >/dev/null 2>&1; then
curl -fsSL {script_url} | sh
@@ -185,6 +199,8 @@ fi
aether_url = shell_single_quote(&session.aether_url),
management_token = shell_single_quote(&session.management_token),
node_name = shell_single_quote(&session.node_name),
tunnel_security = shell_single_quote(&session.tunnel_security),
tunnel_encryption_key = shell_single_quote(&session.tunnel_encryption_key),
script_url = shell_single_quote(TUNNEL_INSTALL_UNIX_SCRIPT_URL),
)
}
@@ -195,11 +211,15 @@ fn build_tunnel_powershell_script(session: &StoredTunnelInstallSession) -> Strin
$env:AETHER_TUNNEL_AETHER_URL = {aether_url}
$env:AETHER_TUNNEL_MANAGEMENT_TOKEN = {management_token}
$env:AETHER_TUNNEL_NODE_NAME = {node_name}
$env:AETHER_TUNNEL_SECURITY = {tunnel_security}
$env:AETHER_TUNNEL_ENCRYPTION_KEY = {tunnel_encryption_key}
irm {script_url} | iex
"###,
aether_url = powershell_single_quote(&session.aether_url),
management_token = powershell_single_quote(&session.management_token),
node_name = powershell_single_quote(&session.node_name),
tunnel_security = powershell_single_quote(&session.tunnel_security),
tunnel_encryption_key = powershell_single_quote(&session.tunnel_encryption_key),
script_url = powershell_single_quote(TUNNEL_INSTALL_POWERSHELL_SCRIPT_URL),
)
}
@@ -664,6 +684,8 @@ pub(crate) async fn build_proxy_node_install_session_response(
aether_url: base_url_from_request(headers, request_context),
management_token,
node_name,
tunnel_security: "non_tls_required".to_string(),
tunnel_encryption_key: generate_tunnel_encryption_key(),
expires_at_unix_secs,
};
let serialized = match serde_json::to_string(&session) {
@@ -712,9 +734,7 @@ pub(super) async fn maybe_build_local_install_response(
if decision.route_family.as_deref() != Some("install") {
return None;
}
if request_context.request_path.starts_with("/install-tunnel/")
|| request_context.request_path.starts_with("/install-proxy/")
{
if request_context.request_path.starts_with("/install-tunnel/") {
return Some(maybe_build_local_tunnel_install_response(state, request_context).await);
}
let Some((code, wants_powershell)) = install_code_from_path(&request_context.request_path)
@@ -893,6 +913,8 @@ mod tests {
aether_url: "https://aether.example".to_string(),
management_token: "ae-test-token".to_string(),
node_name: "jp-proxy-01".to_string(),
tunnel_security: "non_tls_required".to_string(),
tunnel_encryption_key: "base64-32-bytes".to_string(),
expires_at_unix_secs: u64::MAX,
}
}
@@ -907,10 +929,6 @@ mod tests {
tunnel_install_code_from_path("/install-tunnel/abc123.ps1"),
Some(("abc123".to_string(), true))
);
assert_eq!(
tunnel_install_code_from_path("/install-proxy/abc123"),
Some(("abc123".to_string(), false))
);
assert_eq!(tunnel_install_code_from_path("/install-tunnel/a/b"), None);
}
@@ -921,8 +939,10 @@ mod tests {
assert!(script.contains("export AETHER_TUNNEL_AETHER_URL='https://aether.example'"));
assert!(script.contains("export AETHER_TUNNEL_MANAGEMENT_TOKEN='ae-test-token'"));
assert!(script.contains("export AETHER_TUNNEL_NODE_NAME='jp-proxy-01'"));
assert!(script.contains("export AETHER_TUNNEL_SECURITY='non_tls_required'"));
assert!(script.contains("export AETHER_TUNNEL_ENCRYPTION_KEY='base64-32-bytes'"));
assert!(script.contains(
"https://raw.githubusercontent.com/fawney19/Aether/main/apps/aether-tunnel/install.sh"
"https://raw.githubusercontent.com/fawney19/Aether/refs/heads/main/apps/aether-tunnel/install.sh"
));
assert!(!script.contains("aether-rust-pioneer"));
assert!(!script.contains("[[servers]]"));
@@ -935,6 +955,8 @@ mod tests {
assert!(script.contains("$env:AETHER_TUNNEL_AETHER_URL = 'https://aether.example'"));
assert!(script.contains("$env:AETHER_TUNNEL_MANAGEMENT_TOKEN = 'ae-test-token'"));
assert!(script.contains("$env:AETHER_TUNNEL_NODE_NAME = 'jp-proxy-01'"));
assert!(script.contains("$env:AETHER_TUNNEL_SECURITY = 'non_tls_required'"));
assert!(script.contains("$env:AETHER_TUNNEL_ENCRYPTION_KEY = 'base64-32-bytes'"));
assert!(script.contains(
"https://raw.githubusercontent.com/fawney19/Aether/main/apps/aether-tunnel/install.ps1"
));

1
apps/aether-gateway/src/router.rs
View File

@@ -90,7 +90,6 @@ fn frontend_path_bypasses_static(path: &str) -> bool {
|| path.starts_with("/.well-known/")
|| path.starts_with("/install/")
|| path.starts_with("/install-tunnel/")
|| path.starts_with("/install-proxy/")
|| path.starts_with("/i/")
}

79
apps/aether-gateway/src/tunnel/embedded/mod.rs
View File

@@ -17,6 +17,7 @@ use axum::http::HeaderMap;
use axum::response::{IntoResponse, Json};
use axum::routing::{get, post};
use axum::Router;
use dashmap::DashMap;
use tracing::warn;
use crate::{data::GatewayDataState, middleware};
@@ -34,6 +35,7 @@ pub struct AppState {
data: Arc<GatewayDataState>,
request_gate: Option<Arc<ConcurrencyGate>>,
distributed_request_gate: Option<Arc<RuntimeSemaphore>>,
secure_tunnel_keys: Arc<DashMap<String, String>>,
}
#[derive(Debug)]
@@ -55,9 +57,48 @@ impl AppState {
data: Arc::new(GatewayDataState::disabled()),
request_gate: None,
distributed_request_gate: None,
secure_tunnel_keys: Arc::new(DashMap::new()),
}
}
pub(crate) fn register_secure_tunnel_key(
&self,
node_id: impl Into<String>,
key: impl Into<String>,
) {
self.secure_tunnel_keys.insert(node_id.into(), key.into());
}
pub(crate) fn secure_tunnel_key(&self, node_id: &str) -> Option<String> {
self.secure_tunnel_keys
.get(node_id)
.map(|entry| entry.value().clone())
}
async fn secure_tunnel_key_for_node(&self, node_id: &str) -> Option<String> {
if let Some(key) = self.secure_tunnel_key(node_id) {
return Some(key);
}
let key = self
.data
.find_proxy_node(node_id)
.await
.ok()
.flatten()
.and_then(|node| {
node.proxy_metadata.and_then(|metadata| {
metadata
.pointer("/tunnel_security/encryption_key")
.and_then(|value| value.as_str())
.map(str::to_string)
})
});
if let Some(key) = key.as_ref() {
self.register_secure_tunnel_key(node_id.to_string(), key.clone());
}
key
}
pub(crate) fn with_data(mut self, data: Arc<GatewayDataState>) -> Self {
self.data = data;
self
@@ -212,11 +253,47 @@ pub async fn ws_proxy(
let max_streams = resolve_proxy_max_streams(&headers, state.max_streams);
let protocol_version = resolve_proxy_protocol_version(&headers);
let tunnel_security = headers
.get(aether_contracts::tunnel_security::TUNNEL_SECURITY_HEADER)
.and_then(|value| value.to_str().ok())
.map(str::trim)
.filter(|value| !value.is_empty())
.map(str::to_string);
let security_session = headers
.get(aether_contracts::tunnel_security::TUNNEL_SECURITY_SESSION_HEADER)
.and_then(|value| value.to_str().ok())
.map(str::trim)
.filter(|value| !value.is_empty())
.map(str::to_string);
if node_id.is_empty() {
warn!("proxy connection rejected: missing X-Node-ID header");
return axum::http::StatusCode::BAD_REQUEST.into_response();
}
let stored_security_key = state.secure_tunnel_key_for_node(&node_id).await;
let (security_key, security_session) = match tunnel_security.as_deref() {
Some(aether_contracts::tunnel_security::TUNNEL_SECURITY_NON_TLS_REQUIRED) => {
match stored_security_key {
Some(key) => {
let Some(session) = security_session else {
warn!(node_id = %node_id, "secure tunnel requested without a security session");
return axum::http::StatusCode::BAD_REQUEST.into_response();
};
(Some(key), session)
}
None => {
warn!(node_id = %node_id, "secure tunnel requested but no PSK is registered");
return axum::http::StatusCode::UNAUTHORIZED.into_response();
}
}
}
Some(_) => return axum::http::StatusCode::BAD_REQUEST.into_response(),
None if stored_security_key.is_some() => {
warn!(node_id = %node_id, "proxy connection rejected: stored secure tunnel key requires encrypted frames");
return axum::http::StatusCode::UNAUTHORIZED.into_response();
}
None => (None, String::new()),
};
let request_permit = match state.try_acquire_request_permit().await {
Ok(permit) => permit,
@@ -253,6 +330,8 @@ pub async fn ws_proxy(
node_name,
max_streams,
protocol_version,
security_key,
security_session,
state.proxy_conn_cfg,
)
.await

66
apps/aether-gateway/src/tunnel/embedded/proxy_conn.rs
View File

@@ -13,6 +13,8 @@ use tracing::{debug, info, warn};
use super::hub::{ConnConfig, HubRouter, ProxyConn, SendStatus};
use super::protocol;
use aether_contracts::tunnel::Frame;
use aether_contracts::tunnel_security::{SecureFrameCodec, TunnelSecurityRole};
/// Maximum single frame size: 64 MB
const MAX_FRAME_SIZE: usize = 64 * 1024 * 1024;
@@ -24,6 +26,8 @@ pub async fn handle_proxy_connection(
node_name: String,
max_streams: usize,
protocol_version: u8,
security_key: Option<String>,
security_session: String,
cfg: ConnConfig,
) {
let conn_id = hub.alloc_conn_id();
@@ -31,6 +35,18 @@ pub async fn handle_proxy_connection(
let (tx, mut rx) = bounded_queue::<Message>(cfg.outbound_queue_capacity);
let (close_tx, mut close_rx) = watch::channel(false);
let security = match security_key.as_deref() {
Some(key) => {
match SecureFrameCodec::new(key, &security_session, TunnelSecurityRole::Server) {
Ok(codec) => Some(Arc::new(codec)),
Err(error) => {
warn!(conn_id, node_id = %node_id, error = %error, "secure tunnel codec initialization failed");
return;
}
}
}
None => None,
};
let conn = Arc::new(ProxyConn::new(
conn_id,
@@ -46,6 +62,7 @@ pub async fn handle_proxy_connection(
let writer_conn_id = conn_id;
let writer_conn = conn.clone();
let writer_security = security.clone();
let writer = tokio::spawn(async move {
let mut frames_sent: u64 = 0;
loop {
@@ -58,6 +75,13 @@ pub async fn handle_proxy_connection(
_ => 0,
};
let send_started_at = std::time::Instant::now();
let msg = match encrypt_message(msg, writer_security.as_deref()) {
Ok(msg) => msg,
Err(error) => {
warn!(conn_id = writer_conn_id, error = %error, "failed to encrypt outbound proxy frame");
break;
}
};
let send_result = tokio::time::timeout(
Duration::from_secs(15),
ws_tx.send(msg),
@@ -194,7 +218,7 @@ pub async fn handle_proxy_connection(
let reader_hub = hub.clone();
let reader_conn = conn.clone();
let reader = tokio::spawn(async move {
run_proxy_reader(ws_rx, reader_hub, reader_conn, cfg.idle_timeout).await;
run_proxy_reader(ws_rx, reader_hub, reader_conn, cfg.idle_timeout, security).await;
});
let _ = reader.await;
@@ -223,6 +247,7 @@ async fn run_proxy_reader(
hub: Arc<HubRouter>,
conn: Arc<ProxyConn>,
idle_timeout: Duration,
security: Option<Arc<SecureFrameCodec>>,
) {
let idle_enabled = !idle_timeout.is_zero();
let mut oversized_count = 0u32;
@@ -245,7 +270,14 @@ async fn run_proxy_reader(
match msg {
Some(Ok(Message::Binary(data))) => {
frames_received += 1;
let mut data = data.to_vec();
let mut data = match decrypt_message(data, security.as_deref()) {
Ok(data) => data,
Err(error) => {
warn!(conn_id = conn.id, error = %error, "failed to decrypt secure proxy frame");
conn.request_close();
break;
}
};
if data.len() > MAX_FRAME_SIZE {
oversized_count += 1;
warn!(
@@ -294,3 +326,33 @@ async fn run_proxy_reader(
}
}
}
fn encrypt_message(
msg: Message,
security: Option<&SecureFrameCodec>,
) -> Result<Message, aether_contracts::tunnel_security::TunnelSecurityError> {
let Some(codec) = security else {
return Ok(msg);
};
match msg {
Message::Binary(data) => {
let frame = Frame::decode(bytes::Bytes::from(data.to_vec()))
.map_err(|_| aether_contracts::tunnel_security::TunnelSecurityError::Encrypt)?;
Ok(Message::Binary(codec.encrypt_frame(frame)?))
}
other => Ok(other),
}
}
fn decrypt_message(
data: bytes::Bytes,
security: Option<&SecureFrameCodec>,
) -> Result<Vec<u8>, aether_contracts::tunnel_security::TunnelSecurityError> {
let Some(codec) = security else {
return Ok(data.to_vec());
};
let frame = Frame::decode(data)
.map_err(|_| aether_contracts::tunnel_security::TunnelSecurityError::Decrypt)?;
let frame = codec.decrypt_frame(frame)?;
Ok(frame.encode().to_vec())
}

8
apps/aether-gateway/src/tunnel/mod.rs
View File

@@ -455,6 +455,14 @@ impl EmbeddedTunnelState {
self.inner.clone()
}
pub(crate) fn register_secure_tunnel_key(
&self,
node_id: impl Into<String>,
key: impl Into<String>,
) {
self.inner.register_secure_tunnel_key(node_id, key);
}
pub(crate) fn has_local_proxy(&self, node_id: &str) -> bool {
self.inner.hub.has_local_proxy(node_id)
}