diff --git a/client/src/connector.rs b/client/src/connector.rs index 4cd783f..f936f76 100644 --- a/client/src/connector.rs +++ b/client/src/connector.rs @@ -19,8 +19,6 @@ struct TlsDialOpts { enable: bool, cfg: Arc, server_name: String, - - write_custom_head: bool, } impl TlsDialOpts { @@ -32,7 +30,6 @@ impl TlsDialOpts { enable: tls.enable, cfg: rustls_cfg, server_name: cfg.tls_server_name().to_string(), - write_custom_head: !tls.disable_custom_tls_first_byte, }) } @@ -40,13 +37,7 @@ impl TlsDialOpts { if !self.enable { return Ok(stream); } - client_enable_tls( - stream, - Arc::clone(&self.cfg), - &self.server_name, - self.write_custom_head, - ) - .await + client_enable_tls(stream, Arc::clone(&self.cfg), &self.server_name).await } } @@ -110,9 +101,10 @@ pub async fn build_connector(cfg: &ClientConfig) -> Result> { let t = &cfg.transport.tls; let session = QuicSession::dial( addr, - cfg.tls_server_name(), + &cfg.tls_server_name(), cfg.transport.quic.keepalive(), cfg.transport.quic.idle_timeout(), + cfg.transport.quic.max_incoming_streams, &t.cert_file, &t.key_file, &t.trusted_ca_file, @@ -128,6 +120,7 @@ pub async fn build_connector(cfg: &ClientConfig) -> Result> { async fn dial_tcp_tls(cfg: &ClientConfig, tls: &TlsDialOpts) -> Result { let stream = TcpStream::connect(cfg.server_endpoint()).await?; + orbien_core::net::enable_nodelay(&stream); tls.maybe_wrap(boxed_stream(stream)).await } @@ -168,6 +161,7 @@ struct TcpConnector { impl Connector for TcpConnector { async fn open(&self) -> Result { let stream = TcpStream::connect(&self.endpoint).await?; + orbien_core::net::enable_nodelay(&stream); self.tls.maybe_wrap(boxed_stream(stream)).await } } diff --git a/client/src/control/session.rs b/client/src/control/session.rs index 80821e3..247e55a 100644 --- a/client/src/control/session.rs +++ b/client/src/control/session.rs @@ -1,51 +1,53 @@ use crate::connector::{build_connector, Connector}; -use crate::proxy::ProxyManager; -use crate::run_id; +use crate::session_id; +use crate::tunnel::TunnelManager; use anyhow::{anyhow, Result}; use orbien_core::auth; use orbien_core::config::ClientConfig; -use orbien_core::msg::{self, Login, Message, NewProxy, NewWorkConn, Ping}; +use orbien_core::msg::{self, Login, Message, NewDataConn, NewTunnel, Ping}; use orbien_core::transport::DynStream; use orbien_core::VERSION; use std::path::Path; +use std::sync::atomic::{AtomicI64, Ordering}; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::io::{AsyncWriteExt, ReadHalf, WriteHalf}; use tokio::sync::Mutex; use tokio::task::JoinSet; -use tokio::time::interval; +use tokio::time::{interval, sleep}; use tokio_util::sync::CancellationToken; #[derive(Debug)] pub enum SessionEnd { - Disconnected { run_id: String }, - Kicked { run_id: String, reason: String }, + Disconnected { session_id: String }, + Kicked { session_id: String, reason: String }, } type CtrlRead = ReadHalf; type CtrlWrite = WriteHalf; -type OnProxyRemote = Arc; +type OnTunnelRemote = Arc; pub struct Control { cfg: ClientConfig, - run_id: String, + session_id: String, reader: Mutex, writer: Mutex, - proxies: ProxyManager, + tunnels: TunnelManager, connector: Arc, cancel: CancellationToken, - work_tasks: Mutex>, - on_proxy_remote: OnProxyRemote, + data_tasks: Mutex>, + on_tunnel_remote: OnTunnelRemote, + last_pong_unix: AtomicI64, } impl Control { pub async fn start( cfg: &ClientConfig, - previous_run_id: String, + previous_session_id: String, config_path: &Path, parent_cancel: CancellationToken, on_connected: impl FnOnce(), - on_proxy_remote: OnProxyRemote, + on_tunnel_remote: OnTunnelRemote, ) -> Result { let session_cancel = parent_cancel.child_token(); let connector = build_connector(cfg).await?; @@ -58,16 +60,16 @@ impl Control { ); let timestamp = now_secs(); - let privilege_key = auth::get_auth_key(&cfg.auth.token, timestamp); + let auth_digest = auth::compute_auth_digest(&cfg.auth.token, timestamp); let login = Login { version: VERSION.into(), hostname: hostname(), os: std::env::consts::OS.into(), arch: std::env::consts::ARCH.into(), user: cfg.user.clone(), - privilege_key, + auth_digest, timestamp, - run_id: previous_run_id, + session_id: previous_session_id, pool_count: cfg.transport.pool_count, }; tracing::info!( @@ -93,25 +95,26 @@ impl Control { return Err(anyhow!("login failed: {}", resp.error)); } - tracing::info!(run_id = %resp.run_id, "login ok"); - if let Err(e) = run_id::save(config_path, &resp.run_id) { - tracing::warn!(error = %e, "failed to persist run_id"); + tracing::info!(session_id = %resp.session_id, "login ok"); + if let Err(e) = session_id::save(config_path, &resp.session_id) { + tracing::warn!(error = %e, "failed to persist session_id"); } let (reader, writer) = tokio::io::split(stream); let ctl = Arc::new(Control { cfg: cfg.clone(), - run_id: resp.run_id.clone(), + session_id: resp.session_id.clone(), reader: Mutex::new(reader), writer: Mutex::new(writer), - proxies: ProxyManager::from_config(cfg)?, + tunnels: TunnelManager::from_config(cfg)?, connector, cancel: session_cancel.clone(), - work_tasks: Mutex::new(JoinSet::new()), - on_proxy_remote, + data_tasks: Mutex::new(JoinSet::new()), + on_tunnel_remote, + last_pong_unix: AtomicI64::new(now_secs()), }); - ctl.register_all_proxies().await?; + ctl.register_all_tunnels().await?; on_connected(); let hb = Arc::clone(&ctl); @@ -123,18 +126,29 @@ impl Control { } }); + let to = Arc::clone(&ctl); + let to_cancel = session_cancel.clone(); + let timeout_watch = tokio::spawn(async move { + tokio::select! { + _ = to_cancel.cancelled() => {} + _ = to.heartbeat_timeout_loop() => {} + } + }); + let result = ctl.clone().reader_loop().await; ctl.shutdown().await; heartbeat.abort(); + timeout_watch.abort(); let _ = heartbeat.await; + let _ = timeout_watch.await; match result { Ok(ReaderEnd::Kicked(reason)) => Ok(SessionEnd::Kicked { - run_id: resp.run_id, + session_id: resp.session_id, reason, }), Ok(ReaderEnd::Closed) => Ok(SessionEnd::Disconnected { - run_id: resp.run_id, + session_id: resp.session_id, }), Err(e) => Err(e), } @@ -146,96 +160,104 @@ impl Control { let mut writer = self.writer.lock().await; let _ = writer.shutdown().await; } - let mut tasks = self.work_tasks.lock().await; + let mut tasks = self.data_tasks.lock().await; tasks.abort_all(); while tasks.join_next().await.is_some() {} } - async fn register_all_proxies(&self) -> Result<()> { - for p in &self.cfg.proxies { - let msg = match p.proxy_type.as_str() { - "tcp" => Message::NewProxy(new_proxy_base( + async fn register_all_tunnels(&self) -> Result<()> { + for p in &self.cfg.tunnels { + let (local_ip, local_port) = p.service_host_port()?; + if p.requires_local_service() && local_port == 0 { + return Err(anyhow!( + "tunnel `{}` requires service = \"host:port\" (local backend)", + p.name + )); + } + if p.remote_port == 0 && matches!(p.protocol.as_str(), "tcp" | "udp") { + return Err(anyhow!( + "tunnel `{}` type {} requires remotePort > 0", + p.name, + p.protocol + )); + } + let msg = match p.protocol.as_str() { + "tcp" => Message::NewTunnel(new_tunnel_base( &p.name, "tcp", p.remote_port as i32, - &p.local_ip, - p.local_port, - &p.transport, - |np| { - np.custom_domains = Vec::new(); - }, - )), - "udp" => Message::NewProxy(new_proxy_base( - &p.name, - "udp", - p.remote_port as i32, - &p.local_ip, - p.local_port, + &local_ip, + local_port, &p.transport, |_| {}, )), - "http" => Message::NewProxy(new_proxy_base( + "udp" => Message::NewTunnel(new_tunnel_base( + &p.name, + "udp", + p.remote_port as i32, + &local_ip, + local_port, + &p.transport, + |_| {}, + )), + "http" => Message::NewTunnel(new_tunnel_base( &p.name, "http", 0, - &p.local_ip, - p.local_port, + &local_ip, + local_port, &p.transport, |np| { - np.custom_domains = p.custom_domains.clone(); - np.subdomain = p.subdomain.clone(); + np.domains = p.domains.clone(); np.locations = p.locations.clone(); - np.http_user = p.http_user.clone(); - np.http_pwd = p.http_password.clone(); + np.basic_auth_user = p.basic_auth_user.clone(); + np.basic_auth_password = p.basic_auth_password.clone(); np.host_header_rewrite = p.host_header_rewrite.clone(); np.route_by_http_user = p.route_by_http_user.clone(); }, )), - "https" => Message::NewProxy(new_proxy_base( + "https" => Message::NewTunnel(new_tunnel_base( &p.name, "https", 0, - &p.local_ip, - p.local_port, + &local_ip, + local_port, &p.transport, |np| { - np.custom_domains = p.custom_domains.clone(); - np.subdomain = p.subdomain.clone(); + np.domains = p.domains.clone(); }, )), other => { - tracing::warn!(name = %p.name, ty = %other, "skip unsupported proxy type"); + tracing::warn!(name = %p.name, protocol = %other, "skip unsupported tunnel protocol"); continue; } }; let mut writer = self.writer.lock().await; msg::write_msg(&mut *writer, &msg).await?; - match p.proxy_type.as_str() { + match p.protocol.as_str() { "tcp" => tracing::info!( name = %p.name, - local = %format!("{}:{}", p.local_ip, p.local_port), + service = %p.service, remote_port = p.remote_port, - "sent NewProxy" + "sent NewTunnel" ), "udp" => tracing::info!( name = %p.name, - local = %format!("{}:{}", p.local_ip, p.local_port), + service = %p.service, remote_port = p.remote_port, - "sent NewProxy udp" + "sent NewTunnel udp" ), "http" => tracing::info!( name = %p.name, - local = %format!("{}:{}", p.local_ip, p.local_port), - domains = ?p.custom_domains, - subdomain = %p.subdomain, - "sent NewProxy http" + service = %p.service, + domains = ?p.domains, + "sent NewTunnel http" ), "https" => tracing::info!( name = %p.name, - local = %format!("{}:{}", p.local_ip, p.local_port), - domains = ?p.custom_domains, - subdomain = %p.subdomain, - "sent NewProxy https (SNI passthrough)" + service = %p.service, + domains = ?p.domains, + "sent NewTunnel https" ), _ => {} } @@ -269,41 +291,39 @@ impl Control { tracing::warn!(reason = %k.reason, "kicked by server — will exit"); return Ok(ReaderEnd::Kicked(k.reason)); } - Message::ReqWorkConn(_) => { + Message::ReqDataConn(_) => { let ctl = Arc::clone(&self); let cancel = self.cancel.clone(); - self.work_tasks.lock().await.spawn(async move { + self.data_tasks.lock().await.spawn(async move { tokio::select! { _ = cancel.cancelled() => {} - res = ctl.handle_req_work_conn() => { + res = ctl.handle_req_data_conn() => { if let Err(e) = res { - tracing::error!(error = %e, "work tunnel failed"); + tracing::error!(error = %e, "data conn failed"); } } } }); } - Message::NewProxyResp(resp) => { + Message::NewTunnelResp(resp) => { if resp.error.is_empty() { - let remote = normalize_remote_addr( - &self.cfg.server_addr, - &resp.remote_addr, - ); + let remote = normalize_remote_addr(&self.cfg.server, &resp.remote_addr); tracing::info!( - name = %resp.proxy_name, + name = %resp.tunnel_name, remote = %remote, - "proxy started" + "tunnel started" ); - (self.on_proxy_remote)(resp.proxy_name.clone(), remote); + (self.on_tunnel_remote)(resp.tunnel_name.clone(), remote); } else { tracing::error!( - name = %resp.proxy_name, + name = %resp.tunnel_name, error = %resp.error, - "proxy start failed" + "tunnel start failed" ); } } Message::Pong(_) => { + self.last_pong_unix.store(now_secs(), Ordering::Relaxed); tracing::trace!("pong"); } other => { @@ -314,9 +334,9 @@ impl Control { } async fn heartbeat_loop(self: Arc) { - let secs = self.cfg.transport.heartbeat_interval; + let secs = self.effective_ping_interval(); if secs <= 0 { - tracing::debug!("app heartbeat disabled (tcpMux / heartbeatInterval<=0)"); + tracing::debug!("app heartbeat disabled"); std::future::pending::<()>().await; return; } @@ -329,7 +349,7 @@ impl Control { tick.tick().await; let timestamp = now_secs(); let ping = Ping { - privilege_key: auth::get_auth_key(&self.cfg.auth.token, timestamp), + auth_digest: auth::compute_auth_digest(&self.cfg.auth.token, timestamp), timestamp, }; let mut writer = self.writer.lock().await; @@ -342,15 +362,64 @@ impl Control { } } - async fn handle_req_work_conn(self: Arc) -> Result<()> { - let mut work = self.connector.open().await?; + async fn heartbeat_timeout_loop(self: Arc) { + let timeout = self.effective_pong_timeout(); + if timeout <= 0 { + std::future::pending::<()>().await; + return; + } + loop { + if self.cancel.is_cancelled() { + break; + } + sleep(Duration::from_secs(1)).await; + let last = self.last_pong_unix.load(Ordering::Relaxed); + let now = now_secs(); + if last > 0 && now.saturating_sub(last) > timeout { + tracing::warn!(timeout_secs = timeout, "heartbeat timeout"); + self.cancel.cancel(); + break; + } + } + } + + fn effective_ping_interval(&self) -> i64 { + let hb = self.cfg.transport.heartbeat_interval; + if hb > 0 { + return hb; + } + if self.cfg.transport.tcp_mux { + let mux_ka = self.cfg.transport.mux_keepalive_secs; + if mux_ka > 0 { + return mux_ka; + } + } + -1 + } + + fn effective_pong_timeout(&self) -> i64 { + let hb_to = self.cfg.transport.heartbeat_timeout; + if hb_to > 0 { + return hb_to; + } + if self.cfg.transport.heartbeat_interval <= 0 && self.cfg.transport.tcp_mux { + let mux_ka = self.cfg.transport.mux_keepalive_secs; + if mux_ka > 0 { + return mux_ka.saturating_mul(3); + } + } + -1 + } + + async fn handle_req_data_conn(self: Arc) -> Result<()> { + let mut data = self.connector.open().await?; let timestamp = now_secs(); msg::write_msg( - &mut work, - &Message::NewWorkConn(NewWorkConn { - run_id: self.run_id.clone(), - privilege_key: auth::get_auth_key(&self.cfg.auth.token, timestamp), + &mut data, + &Message::NewDataConn(NewDataConn { + session_id: self.session_id.clone(), + auth_digest: auth::compute_auth_digest(&self.cfg.auth.token, timestamp), timestamp, }), ) @@ -360,21 +429,21 @@ impl Control { _ = self.cancel.cancelled() => { return Ok(()); } - msg = msg::read_msg(&mut work) => { + msg = msg::read_msg(&mut data) => { match msg? { - Message::StartWorkConn(s) => s, + Message::StartDataConn(s) => s, other => { - return Err(anyhow!("expected StartWorkConn, got {}", other.type_byte())) + return Err(anyhow!("expected StartDataConn, got {}", other.type_byte())) } } } }; if !start.error.is_empty() { - return Err(anyhow!("StartWorkConn error: {}", start.error)); + return Err(anyhow!("StartDataConn error: {}", start.error)); } - self.proxies.handle_work_conn(&start, work).await + self.tunnels.handle_data_conn(&start, data).await } } @@ -406,8 +475,8 @@ fn hostname() -> String { .unwrap_or_else(|| "unknown".into()) } -fn omit_client_mode(mode: &str) -> String { - match mode.trim().to_ascii_lowercase().as_str() { +fn omit_client_side(side: &str) -> String { + match side.trim().to_ascii_lowercase().as_str() { "" | "client" => String::new(), other => other.to_string(), } @@ -420,6 +489,10 @@ fn normalize_remote_addr(server_addr: &str, remote_addr: &str) -> String { } if let Some(port) = remote.strip_prefix(':') { let host = server_addr.trim(); + let host = host.rsplit_once(':').map(|(h, _)| h).unwrap_or(host); + if !host.is_empty() && !port.is_empty() && !host.contains(':') { + return format!("{host}:{port}"); + } if !host.is_empty() && !port.is_empty() { return format!("{host}:{port}"); } @@ -427,32 +500,31 @@ fn normalize_remote_addr(server_addr: &str, remote_addr: &str) -> String { remote.to_string() } -fn new_proxy_base( +fn new_tunnel_base( name: &str, - proxy_type: &str, + protocol: &str, remote_port: i32, local_ip: &str, local_port: u16, - transport: &orbien_core::config::ProxyTransportConfig, - extra: impl FnOnce(&mut NewProxy), -) -> NewProxy { - let mut np = NewProxy { - proxy_name: name.into(), - proxy_type: proxy_type.into(), + transport: &orbien_core::config::TunnelTransportConfig, + extra: impl FnOnce(&mut NewTunnel), +) -> NewTunnel { + let mut np = NewTunnel { + tunnel_name: name.into(), + protocol: protocol.into(), remote_port, local_ip: local_ip.into(), local_port: i32::from(local_port), - custom_domains: Vec::new(), - subdomain: String::new(), + domains: Vec::new(), locations: Vec::new(), - http_user: String::new(), - http_pwd: String::new(), + basic_auth_user: String::new(), + basic_auth_password: String::new(), host_header_rewrite: String::new(), headers: Default::default(), response_headers: Default::default(), route_by_http_user: String::new(), - bandwidth_limit: transport.bandwidth_limit.clone(), - bandwidth_limit_mode: omit_client_mode(&transport.bandwidth_limit_mode), + bandwidth: transport.bandwidth, + bandwidth_limit_side: omit_client_side(&transport.bandwidth_limit_side), }; extra(&mut np); np diff --git a/client/src/handle.rs b/client/src/handle.rs index bc59290..f6525f4 100644 --- a/client/src/handle.rs +++ b/client/src/handle.rs @@ -26,7 +26,7 @@ impl ClientStatus { } #[derive(Debug, Default)] -struct ProxyRemoteState { +struct TunnelRemoteState { gen: u64, by_name: HashMap, } @@ -35,7 +35,7 @@ struct Inner { status: Mutex, last_error: Mutex>, pending_logs: Mutex>, - proxy_remotes: Mutex, + tunnel_remotes: Mutex, cancel: Mutex>, join: Mutex>>, } @@ -58,7 +58,7 @@ impl ClientHandle { status: Mutex::new(ClientStatus::Stopped), last_error: Mutex::new(None), pending_logs: Mutex::new(Vec::new()), - proxy_remotes: Mutex::new(ProxyRemoteState::default()), + tunnel_remotes: Mutex::new(TunnelRemoteState::default()), cancel: Mutex::new(None), join: Mutex::new(None), }), @@ -91,13 +91,13 @@ impl ClientHandle { ) } - pub fn proxy_remotes_if_changed( + pub fn tunnel_remotes_if_changed( &self, since_gen: u64, ) -> Option<(u64, HashMap)> { let g = self .inner - .proxy_remotes + .tunnel_remotes .lock() .unwrap_or_else(|e| e.into_inner()); if g.gen == since_gen { @@ -106,23 +106,23 @@ impl ClientHandle { Some((g.gen, g.by_name.clone())) } - pub fn clear_proxy_remotes(&self) { + pub fn clear_tunnel_remotes(&self) { let mut g = self .inner - .proxy_remotes + .tunnel_remotes .lock() .unwrap_or_else(|e| e.into_inner()); g.by_name.clear(); g.gen = g.gen.wrapping_add(1); } - fn set_proxy_remote(&self, name: String, remote_addr: String) { + fn set_tunnel_remote(&self, name: String, remote_addr: String) { if name.is_empty() { return; } let mut g = self .inner - .proxy_remotes + .tunnel_remotes .lock() .unwrap_or_else(|e| e.into_inner()); if g.by_name.get(&name) == Some(&remote_addr) { @@ -155,11 +155,13 @@ impl ClientHandle { } } - pub async fn run_foreground(self, cfg: ClientConfig, config_path: PathBuf) -> Result<()> { + pub async fn run_foreground(self, mut cfg: ClientConfig, config_path: PathBuf) -> Result<()> { + cfg.prepare_runtime(&config_path); + cfg.validate()?; let cancel = CancellationToken::new(); self.set_status(ClientStatus::Starting); self.set_error(None); - self.clear_proxy_remotes(); + self.clear_tunnel_remotes(); let result = Service::new(cfg, config_path) .run( cancel.clone(), @@ -175,15 +177,15 @@ impl ClientHandle { }, { let h = self.clone(); - Arc::new(move |name, remote| h.set_proxy_remote(name, remote)) + Arc::new(move |name, remote| h.set_tunnel_remote(name, remote)) }, { let h = self.clone(); - Arc::new(move || h.clear_proxy_remotes()) + Arc::new(move || h.clear_tunnel_remotes()) }, ) .await; - self.clear_proxy_remotes(); + self.clear_tunnel_remotes(); self.set_status(ClientStatus::Stopped); if let Err(ref e) = result { self.set_error(Some(e.to_string())); @@ -191,15 +193,17 @@ impl ClientHandle { result } - pub fn start(&self, cfg: ClientConfig, config_path: PathBuf) -> Result<()> { + pub fn start(&self, mut cfg: ClientConfig, config_path: PathBuf) -> Result<()> { if self.status().is_active() { bail!("client already running"); } + cfg.prepare_runtime(&config_path); + cfg.validate()?; let cancel = CancellationToken::new(); *self.inner.cancel.lock().unwrap_or_else(|e| e.into_inner()) = Some(cancel.clone()); self.set_status(ClientStatus::Starting); self.set_error(None); - self.clear_proxy_remotes(); + self.clear_tunnel_remotes(); let handle = self.clone(); let join = tokio::spawn(async move { @@ -213,22 +217,28 @@ impl ClientHandle { h.enqueue_log(line); } }; - let on_proxy_remote: Arc = { + let on_tunnel_remote: Arc = { let h = handle.clone(); - Arc::new(move |name: String, remote: String| h.set_proxy_remote(name, remote)) + Arc::new(move |name: String, remote: String| h.set_tunnel_remote(name, remote)) }; let on_remotes_clear: Arc = { let h = handle.clone(); - Arc::new(move || h.clear_proxy_remotes()) + Arc::new(move || h.clear_tunnel_remotes()) }; let result = Service::new(cfg, config_path) - .run(cancel, on_status, on_log, on_proxy_remote, on_remotes_clear) + .run( + cancel, + on_status, + on_log, + on_tunnel_remote, + on_remotes_clear, + ) .await; if let Err(e) = result { tracing::error!(error = %e, "client service ended with error"); handle.set_error(Some(e.to_string())); } - handle.clear_proxy_remotes(); + handle.clear_tunnel_remotes(); handle.set_status(ClientStatus::Stopped); *handle .inner @@ -274,6 +284,6 @@ impl ClientHandle { } else { self.set_status(ClientStatus::Stopped); } - self.clear_proxy_remotes(); + self.clear_tunnel_remotes(); } } diff --git a/client/src/lib.rs b/client/src/lib.rs index 8e5de67..af68910 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -2,9 +2,9 @@ mod connector; mod control; mod handle; mod plugin; -mod proxy; -mod run_id; mod service; +mod session_id; +mod tunnel; pub use handle::{ClientHandle, ClientStatus}; pub use orbien_core::config::{resolve_client_config_path, ClientConfig}; diff --git a/client/src/main.rs b/client/src/main.rs index f7f9d91..5f5873f 100644 --- a/client/src/main.rs +++ b/client/src/main.rs @@ -6,7 +6,7 @@ use tracing_subscriber::EnvFilter; #[derive(Parser, Debug)] #[command( name = "orbien", - about = "orbien client — TCP tunnel over TCP/QUIC", + about = "orbien client", after_help = "Config:\n \ orbien # try ./orbien.toml, then ./conf/orbien.toml\n \ orbien -c conf/orbien.toml # explicit path" @@ -30,11 +30,9 @@ async fn main() -> Result<()> { tracing::info!( server = %cfg.server_endpoint(), protocol = %cfg.transport.protocol, - proxies = cfg.proxies.len(), + tunnels = cfg.tunnels.len(), "starting orbien" ); - ClientHandle::new() - .run_foreground(cfg, config_path) - .await + ClientHandle::new().run_foreground(cfg, config_path).await } diff --git a/client/src/plugin/mod.rs b/client/src/plugin/mod.rs index 5999070..dc3e735 100644 --- a/client/src/plugin/mod.rs +++ b/client/src/plugin/mod.rs @@ -1,4 +1,4 @@ -mod https2http; +mod tls_term; use anyhow::{bail, Result}; use async_trait::async_trait; @@ -29,8 +29,8 @@ pub trait Plugin: Send + Sync { pub fn create(ctx: PluginContext, cfg: &PluginConfig) -> Result> { match cfg.plugin_type.as_str() { - "https2http" => { - let p = https2http::Https2HttpPlugin::new(ctx, cfg)?; + "tls-term" => { + let p = tls_term::TlsTermPlugin::new(ctx, cfg)?; Ok(Arc::new(p)) } other => bail!("unknown client plugin type: {other}"), diff --git a/client/src/plugin/https2http.rs b/client/src/plugin/tls_term.rs similarity index 87% rename from client/src/plugin/https2http.rs rename to client/src/plugin/tls_term.rs index 0be4b46..a5c9660 100644 --- a/client/src/plugin/https2http.rs +++ b/client/src/plugin/tls_term.rs @@ -8,18 +8,18 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; use tokio_rustls::TlsAcceptor; -pub struct Https2HttpPlugin { +pub struct TlsTermPlugin { local_addr: String, host_header_rewrite: String, request_headers: Vec<(String, String)>, acceptor: TlsAcceptor, } -impl Https2HttpPlugin { +impl TlsTermPlugin { pub fn new(ctx: PluginContext, cfg: &PluginConfig) -> Result { - let local_addr = cfg.local_addr.trim().to_string(); + let local_addr = cfg.service.trim().to_string(); if local_addr.is_empty() { - bail!("https2http requires plugin.localAddr (e.g. \"127.0.0.1:80\")"); + bail!("tls-term requires plugin.service (e.g. \"127.0.0.1:80\")"); } let cn = if ctx.cert_common_name.is_empty() { @@ -27,7 +27,7 @@ impl Https2HttpPlugin { } else { ctx.cert_common_name.clone() }; - let tls_cfg = load_or_generate_https_server_config(&cfg.crt_path, &cfg.key_path, &cn)?; + let tls_cfg = load_or_generate_https_server_config(&cfg.cert_file, &cfg.key_file, &cn)?; let acceptor = TlsAcceptor::from(tls_cfg); let request_headers: Vec<(String, String)> = cfg @@ -38,10 +38,10 @@ impl Https2HttpPlugin { .collect(); tracing::info!( - proxy = %ctx.name, + tunnel = %ctx.name, %local_addr, rewrite = %cfg.host_header_rewrite, - "plugin https2http ready (TLS terminates on agent)" + "plugin tls-term ready (TLS terminates on agent)" ); Ok(Self { @@ -54,9 +54,9 @@ impl Https2HttpPlugin { } #[async_trait] -impl Plugin for Https2HttpPlugin { +impl Plugin for TlsTermPlugin { fn name(&self) -> &str { - "https2http" + "tls-term" } async fn handle(&self, conn: ConnectionInfo) -> Result<()> { @@ -64,11 +64,12 @@ impl Plugin for Https2HttpPlugin { .acceptor .accept(conn.stream) .await - .map_err(|e| anyhow!("https2http TLS accept failed: {e}"))?; + .map_err(|e| anyhow!("tls-term TLS accept failed: {e}"))?; let mut local = TcpStream::connect(&self.local_addr) .await - .map_err(|e| anyhow!("https2http dial {}: {e}", self.local_addr))?; + .map_err(|e| anyhow!("tls-term dial {}: {e}", self.local_addr))?; + orbien_core::net::enable_nodelay(&local); let (mut tls_r, mut tls_w) = tokio::io::split(tls); let mut head = read_http_request_head(&mut tls_r).await?; @@ -80,7 +81,7 @@ impl Plugin for Https2HttpPlugin { tracing::debug!( local = %self.local_addr, src = %format!("{}:{}", conn.src_addr, conn.src_port), - "https2http joining decrypted <-> local HTTP" + "tls-term joining decrypted <-> local HTTP" ); let (mut local_r, mut local_w) = tokio::io::split(local); diff --git a/client/src/proxy/mod.rs b/client/src/proxy/mod.rs deleted file mode 100644 index 575ac15..0000000 --- a/client/src/proxy/mod.rs +++ /dev/null @@ -1,4 +0,0 @@ -mod manager; -mod udp; - -pub use manager::ProxyManager; diff --git a/client/src/service.rs b/client/src/service.rs index 5828660..870ae12 100644 --- a/client/src/service.rs +++ b/client/src/service.rs @@ -1,7 +1,7 @@ use crate::control::{Control, SessionEnd}; use crate::handle::ClientStatus; -use crate::run_id; -use anyhow::{anyhow, Result}; +use crate::session_id; +use anyhow::Result; use orbien_core::config::ClientConfig; use std::path::PathBuf; use std::sync::Arc; @@ -30,12 +30,12 @@ impl Service { cancel: CancellationToken, mut on_status: impl FnMut(ClientStatus), mut on_log: impl FnMut(String), - on_proxy_remote: Arc, + on_tunnel_remote: Arc, on_remotes_clear: Arc, ) -> Result<()> { - let mut run_id = run_id::load(&self.config_path); - if !run_id.is_empty() { - tracing::info!(%run_id, "restored persisted run_id"); + let mut session_id = session_id::load(&self.config_path); + if !session_id.is_empty() { + tracing::info!(%session_id, "restored persisted session_id"); } let mut first_attempt = true; @@ -57,14 +57,14 @@ impl Service { let end = Control::start( &self.cfg, - run_id.clone(), + session_id.clone(), &self.config_path, cancel.clone(), || { on_status(ClientStatus::Running); on_log("INFO connected to server".into()); }, - Arc::clone(&on_proxy_remote), + Arc::clone(&on_tunnel_remote), ) .await; @@ -72,23 +72,23 @@ impl Service { match end { Ok(SessionEnd::Kicked { - run_id: rid, + session_id: rid, reason, }) => { - tracing::error!( - run_id = %rid, + tracing::warn!( + session_id = %rid, %reason, "kicked by server — stopping (no reconnect)" ); - on_log(format!("ERROR kicked by server: {reason}")); - return Err(anyhow!("kicked by server: {reason}")); + on_log(format!("WARN kicked by server: {reason}")); + return Ok(()); } - Ok(SessionEnd::Disconnected { run_id: rid }) => { + Ok(SessionEnd::Disconnected { session_id: rid }) => { if cancel.is_cancelled() { - tracing::info!(run_id = %rid, "session ended after cancel"); + tracing::info!(session_id = %rid, "session ended after cancel"); return Ok(()); } - run_id = rid; + session_id = rid; on_log("WARN disconnected from server".into()); on_status(ClientStatus::Reconnecting); backoff_secs = RECONNECT_BASE_SECS; diff --git a/client/src/run_id.rs b/client/src/session_id.rs similarity index 62% rename from client/src/run_id.rs rename to client/src/session_id.rs index 3e8f66a..f6ce7bb 100644 --- a/client/src/run_id.rs +++ b/client/src/session_id.rs @@ -5,27 +5,29 @@ pub fn path_for(config_path: &Path) -> PathBuf { let ext = p .extension() .and_then(|e| e.to_str()) - .map(|e| format!("{e}.run_id")) - .unwrap_or_else(|| "run_id".into()); + .map(|e| format!("{e}.session_id")) + .unwrap_or_else(|| "session_id".into()); p.set_extension(ext); p } -pub fn load(config_path: &Path) -> String { - let path = path_for(config_path); - std::fs::read_to_string(&path) +fn read_valid_id(path: &Path) -> Option { + std::fs::read_to_string(path) .ok() .map(|s| s.trim().to_string()) .filter(|s| { !s.is_empty() && s.len() <= 64 && s.chars().all(|c| c.is_ascii_hexdigit() || c == '-') }) - .unwrap_or_default() } -pub fn save(config_path: &Path, run_id: &str) -> std::io::Result<()> { +pub fn load(config_path: &Path) -> String { + read_valid_id(&path_for(config_path)).unwrap_or_default() +} + +pub fn save(config_path: &Path, session_id: &str) -> std::io::Result<()> { let path = path_for(config_path); if let Some(parent) = path.parent() { std::fs::create_dir_all(parent)?; } - std::fs::write(path, run_id) + std::fs::write(path, session_id) } diff --git a/client/src/proxy/manager.rs b/client/src/tunnel/manager.rs similarity index 52% rename from client/src/proxy/manager.rs rename to client/src/tunnel/manager.rs index 561dcb8..2416566 100644 --- a/client/src/proxy/manager.rs +++ b/client/src/tunnel/manager.rs @@ -1,12 +1,12 @@ use super::udp::run_udp_session; use crate::plugin::{self, ConnectionInfo, Plugin, PluginContext}; use anyhow::{anyhow, Result}; -use orbien_core::config::{ClientConfig, ProxyConfig}; +use orbien_core::config::{ClientConfig, TunnelConfig}; use orbien_core::io; -use orbien_core::limit::{self, maybe_limit, BandwidthLimitMode, BandwidthLimiter}; -use orbien_core::msg::StartWorkConn; +use orbien_core::limit::{self, maybe_limit, BandwidthLimitSide, BandwidthLimiter}; +use orbien_core::msg::StartDataConn; use orbien_core::net::{ - addrs_from_start_work, build_proxy_protocol_header, parse_proxy_protocol_version, + addrs_from_start_data_conn, build_proxy_protocol_header, parse_proxy_protocol_version, }; use orbien_core::transport::DynStream; use std::collections::HashMap; @@ -15,41 +15,42 @@ use tokio::io::AsyncWriteExt; use tokio::net::TcpStream; use tokio::sync::{oneshot, Mutex}; -struct ProxyEntry { - cfg: ProxyConfig, +struct TunnelEntry { + cfg: TunnelConfig, limiter: Option>, plugin: Option>, proxy_protocol: Option<&'static str>, udp_cancel: Mutex>>, } -pub struct ProxyManager { - by_name: HashMap, +pub struct TunnelManager { + by_name: HashMap, udp_packet_size: usize, } -impl ProxyManager { +impl TunnelManager { pub fn from_config(cfg: &ClientConfig) -> Result { let mut by_name = HashMap::new(); - for p in &cfg.proxies { - let limiter = limit::limiter_if_mode( - &p.transport.bandwidth_limit, - &p.transport.bandwidth_limit_mode, - BandwidthLimitMode::Client, + for p in &cfg.tunnels { + let limiter = limit::limiter_if_side( + p.transport.bandwidth, + &p.transport.bandwidth_limit_side, + BandwidthLimitSide::Client, ) .unwrap_or_else(|e| { tracing::warn!( - proxy = %p.name, + tunnel = %p.name, error = %e, - "invalid bandwidthLimit; ignoring" + "invalid bandwidth; ignoring" ); None }); if let Some(ref l) = limiter { tracing::info!( - proxy = %p.name, + tunnel = %p.name, + mbps = p.transport.bandwidth, bytes_per_sec = l.bytes_per_sec(), - mode = "client", + side = "client", "bandwidth limit enabled" ); } @@ -58,13 +59,7 @@ impl ProxyManager { if pc.plugin_type.is_empty() { None } else { - let cn = p.custom_domains.first().cloned().unwrap_or_else(|| { - if p.subdomain.is_empty() { - "localhost".into() - } else { - p.subdomain.clone() - } - }); + let cn = pick_cert_common_name(&p.domains, &p.name); let ctx = PluginContext { name: p.name.clone(), cert_common_name: cn, @@ -78,15 +73,15 @@ impl ProxyManager { let proxy_protocol = parse_proxy_protocol_version(&p.transport.proxy_protocol_version)?; if let Some(ver) = proxy_protocol { tracing::info!( - proxy = %p.name, + tunnel = %p.name, version = ver, - "proxy protocol enabled (client writes PP to local)" + "PROXY Protocol enabled (client writes header to local)" ); } by_name.insert( p.name.clone(), - ProxyEntry { + TunnelEntry { cfg: p.clone(), limiter, plugin, @@ -101,40 +96,40 @@ impl ProxyManager { }) } - pub async fn handle_work_conn(&self, start: &StartWorkConn, work: DynStream) -> Result<()> { + pub async fn handle_data_conn(&self, start: &StartDataConn, data: DynStream) -> Result<()> { let entry = self .by_name - .get(&start.proxy_name) - .ok_or_else(|| anyhow!("unknown proxy: {}", start.proxy_name))?; + .get(&start.tunnel_name) + .ok_or_else(|| anyhow!("unknown tunnel: {}", start.tunnel_name))?; - match entry.cfg.proxy_type.as_str() { - "udp" => self.handle_udp(entry, work).await, - "tcp" | "http" | "https" => self.handle_stream_proxy(entry, start, work).await, + match entry.cfg.protocol.as_str() { + "udp" => self.handle_udp(entry, data).await, + "tcp" | "http" | "https" => self.handle_stream(entry, start, data).await, other => Err(anyhow!( - "unsupported proxy type on work conn: {} ({})", + "unsupported tunnel protocol on data conn: {} ({})", other, entry.cfg.name )), } } - async fn handle_stream_proxy( + async fn handle_stream( &self, - entry: &ProxyEntry, - start: &StartWorkConn, - work: DynStream, + entry: &TunnelEntry, + start: &StartDataConn, + data: DynStream, ) -> Result<()> { - let work = maybe_limit(work, entry.limiter.clone()); + let data = maybe_limit(data, entry.limiter.clone()); if let Some(ref plugin) = entry.plugin { tracing::debug!( - proxy = %entry.cfg.name, + tunnel = %entry.cfg.name, plugin = plugin.name(), "handle by plugin" ); return plugin .handle(ConnectionInfo { - stream: work, + stream: data, src_addr: start.src_addr.clone(), src_port: start.src_port, dst_addr: start.dst_addr.clone(), @@ -143,56 +138,70 @@ impl ProxyManager { .await; } - let local_addr = format!("{}:{}", entry.cfg.local_ip, entry.cfg.local_port); - let mut local = TcpStream::connect(&local_addr).await.map_err(|e| { + let (svc_host, svc_port) = entry.cfg.service_host_port()?; + let local_addr = entry.cfg.service.trim().to_string(); + if local_addr.is_empty() || svc_port == 0 { + return Err(anyhow!( + "tunnel {} has empty/invalid service (local backend)", + entry.cfg.name + )); + } + let local = TcpStream::connect(&local_addr).await.map_err(|e| { anyhow!( - "dial local {} for proxy {}: {}", + "dial local {} for tunnel {}: {}", local_addr, entry.cfg.name, e ) })?; + orbien_core::net::enable_nodelay(&local); + let mut local = local; if let Some(ver) = entry.proxy_protocol { - if let Some((src, dst)) = addrs_from_start_work( + if let Some((src, dst)) = addrs_from_start_data_conn( &start.src_addr, start.src_port, &start.dst_addr, start.dst_port, - entry.cfg.local_port, + svc_port, ) { let hdr = build_proxy_protocol_header(src, dst, ver)?; local.write_all(&hdr).await?; tracing::debug!( - proxy = %entry.cfg.name, + tunnel = %entry.cfg.name, version = ver, %src, %dst, - "wrote proxy protocol header to local" + "wrote PROXY Protocol header to local" ); } else { tracing::debug!( - proxy = %entry.cfg.name, - "proxy protocol configured but StartWorkConn src empty; skip" + tunnel = %entry.cfg.name, + "PROXY Protocol configured but StartDataConn src empty; skip" ); } } tracing::debug!( - proxy = %entry.cfg.name, + tunnel = %entry.cfg.name, %local_addr, + host = %svc_host, limited = entry.limiter.is_some(), - "joining work <-> local" + "joining data <-> local" ); - let _ = io::join(work, local).await; + if let Err(e) = io::join(data, local).await { + tracing::debug!(error = %e, "join ended"); + } Ok(()) } - async fn handle_udp(&self, entry: &ProxyEntry, work: DynStream) -> Result<()> { - let local_addr: std::net::SocketAddr = - format!("{}:{}", entry.cfg.local_ip, entry.cfg.local_port) - .parse() - .map_err(|e| anyhow!("invalid local udp addr: {e}"))?; + async fn handle_udp(&self, entry: &TunnelEntry, data: DynStream) -> Result<()> { + let local_addr: std::net::SocketAddr = entry + .cfg + .service + .trim() + .parse() + .map_err(|e| anyhow!("invalid local udp service {}: {e}", entry.cfg.service))?; let (cancel_tx, cancel_rx) = oneshot::channel(); { @@ -204,15 +213,15 @@ impl ProxyManager { } tracing::info!( - proxy = %entry.cfg.name, + tunnel = %entry.cfg.name, %local_addr, - "udp work conn; starting forwarder" + "udp data conn; starting forwarder" ); - let work = maybe_limit(work, entry.limiter.clone()); + let data = maybe_limit(data, entry.limiter.clone()); run_udp_session( - work, + data, local_addr, self.udp_packet_size, entry.proxy_protocol.map(|s| s.to_string()), @@ -221,3 +230,21 @@ impl ProxyManager { .await } } +fn pick_cert_common_name(domains: &[String], tunnel_name: &str) -> String { + for d in domains { + let d = d.trim(); + if d.is_empty() { + continue; + } + if d.contains('.') && !d.contains('*') { + return d.to_ascii_lowercase(); + } + } + for d in domains { + let d = d.trim(); + if !d.is_empty() { + return d.to_ascii_lowercase(); + } + } + tunnel_name.trim().to_ascii_lowercase() +} diff --git a/client/src/tunnel/mod.rs b/client/src/tunnel/mod.rs new file mode 100644 index 0000000..1d53e92 --- /dev/null +++ b/client/src/tunnel/mod.rs @@ -0,0 +1,4 @@ +mod manager; +mod udp; + +pub use manager::TunnelManager; diff --git a/client/src/proxy/udp.rs b/client/src/tunnel/udp.rs similarity index 68% rename from client/src/proxy/udp.rs rename to client/src/tunnel/udp.rs index ac313cb..a4aece1 100644 --- a/client/src/proxy/udp.rs +++ b/client/src/tunnel/udp.rs @@ -1,30 +1,30 @@ use anyhow::{anyhow, Result}; use orbien_core::msg::{self, Message}; use orbien_core::transport::DynStream; -use orbien_core::udp::{forwarder, spawn_work_ping, CHANNEL_CAP}; +use orbien_core::udp::{forwarder, spawn_data_ping, CHANNEL_CAP}; use std::net::SocketAddr; use tokio::io::AsyncRead; use tokio::sync::{mpsc, oneshot}; pub async fn run_udp_session( - work: DynStream, + data: DynStream, local_addr: SocketAddr, packet_size: usize, proxy_protocol_version: Option, mut cancel_rx: oneshot::Receiver<()>, ) -> Result<()> { - let (reader, mut writer) = tokio::io::split(work); + let (reader, mut writer) = tokio::io::split(data); - let (to_work_tx, mut to_work_rx) = mpsc::channel::(CHANNEL_CAP); - let (from_work_tx, from_work_rx) = mpsc::channel(CHANNEL_CAP); + let (to_data_tx, mut to_data_rx) = mpsc::channel::(CHANNEL_CAP); + let (from_data_tx, from_data_rx) = mpsc::channel(CHANNEL_CAP); - let ping = spawn_work_ping(to_work_tx.clone()); + let ping = spawn_data_ping(to_data_tx.clone()); let forward = { - let tx = to_work_tx; + let tx = to_data_tx; tokio::spawn(async move { forwarder( local_addr, - from_work_rx, + from_data_rx, tx, packet_size, proxy_protocol_version, @@ -34,10 +34,10 @@ pub async fn run_udp_session( }; let (fail_tx, mut fail_rx) = mpsc::channel::(1); - let from_work_tx_r = from_work_tx; + let from_data_tx_r = from_data_tx; let fail_r = fail_tx; let reader_task = tokio::spawn(async move { - work_reader(reader, from_work_tx_r, fail_r).await; + data_reader(reader, from_data_tx_r, fail_r).await; }); let result = loop { @@ -52,11 +52,11 @@ pub async fn run_udp_session( None => Ok(()), }; } - out = to_work_rx.recv() => { + out = to_data_rx.recv() => { match out { Some(m) => { if let Err(e) = msg::write_msg(&mut writer, &m).await { - break Err(anyhow!("udp work write: {e}")); + break Err(anyhow!("udp data write: {e}")); } } None => break Ok(()), @@ -74,23 +74,23 @@ pub async fn run_udp_session( result } -async fn work_reader( +async fn data_reader( mut reader: R, - from_work_tx: mpsc::Sender, + from_data_tx: mpsc::Sender, fail_tx: mpsc::Sender, ) { loop { match msg::read_msg(&mut reader).await { Ok(Message::UdpPacket(pkt)) => { - tracing::trace!(len = pkt.content.len(), "udp packet from work"); - let _ = from_work_tx.try_send(pkt); + tracing::trace!(len = pkt.content.len(), "udp packet from data"); + let _ = from_data_tx.try_send(pkt); } Ok(Message::Ping(_)) => {} Ok(other) => { - tracing::debug!(ty = other.type_byte(), "udp work unexpected message"); + tracing::debug!(ty = other.type_byte(), "udp data unexpected message"); } Err(e) => { - let _ = fail_tx.send(anyhow!("udp work read: {e}")).await; + let _ = fail_tx.send(anyhow!("udp data read: {e}")).await; return; } }