starred/orbien
1
0
Fork 0
mirror of https://github.com/orbien-org/orbien.git synced 2026-09-22 00:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Refactor the client code and redesign a more concise configuration.

Browse Source
This commit is contained in:
lxien
2026-08-16 08:46:18 +08:00
parent faa308cb42
commit beaf8ee4c5
13 changed files with 382 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/connector.rs
+5 -11
View File
@@ -19,8 +19,6 @@ struct TlsDialOpts {
enable: bool,
cfg: Arc<RustlsClientConfig>,
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<Arc<dyn Connector>> {
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<Arc<dyn Connector>> {
async fn dial_tcp_tls(cfg: &ClientConfig, tls: &TlsDialOpts) -> Result<DynStream> {
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<DynStream> {
let stream = TcpStream::connect(&self.endpoint).await?;
orbien_core::net::enable_nodelay(&stream);
self.tls.maybe_wrap(boxed_stream(stream)).await
}
}
client/src/control/session.rs
+186 -114
View File
@@ -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<DynStream>;
type CtrlWrite = WriteHalf<DynStream>;
type OnProxyRemote = Arc<dyn Fn(String, String) + Send + Sync>;
type OnTunnelRemote = Arc<dyn Fn(String, String) + Send + Sync>;
pub struct Control {
cfg: ClientConfig,
run_id: String,
session_id: String,
reader: Mutex<CtrlRead>,
writer: Mutex<CtrlWrite>,
proxies: ProxyManager,
tunnels: TunnelManager,
connector: Arc<dyn Connector>,
cancel: CancellationToken,
work_tasks: Mutex<JoinSet<()>>,
on_proxy_remote: OnProxyRemote,
data_tasks: Mutex<JoinSet<()>>,
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<SessionEnd> {
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<Self>) {
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<Self>) -> Result<()> {
let mut work = self.connector.open().await?;
async fn heartbeat_timeout_loop(self: Arc<Self>) {
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<Self>) -> 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
client/src/handle.rs
+32 -22
View File
@@ -26,7 +26,7 @@ impl ClientStatus {
}
#[derive(Debug, Default)]
struct ProxyRemoteState {
struct TunnelRemoteState {
gen: u64,
by_name: HashMap<String, String>,
}
@@ -35,7 +35,7 @@ struct Inner {
status: Mutex<ClientStatus>,
last_error: Mutex<Option<String>>,
pending_logs: Mutex<Vec<String>>,
proxy_remotes: Mutex<ProxyRemoteState>,
tunnel_remotes: Mutex<TunnelRemoteState>,
cancel: Mutex<Option<CancellationToken>>,
join: Mutex<Option<JoinHandle<()>>>,
}
@@ -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<String, String>)> {
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<dyn Fn(String, String) + Send + Sync> = {
let on_tunnel_remote: Arc<dyn Fn(String, String) + Send + Sync> = {
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<dyn Fn() + Send + Sync> = {
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();
}
}
client/src/lib.rs
+2 -2
View File
@@ -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};
client/src/main.rs
+3 -5
View File
@@ -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
}
client/src/plugin/mod.rs
+3 -3
View File
@@ -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<Arc<dyn Plugin>> {
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}"),
client/src/plugin/https2http.rs → client/src/plugin/tls_term.rs
+13 -12
View File
@@ -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<Self> {
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);
client/src/proxy/mod.rs
-4
View File
@@ -1,4 +0,0 @@
mod manager;
mod udp;
pub use manager::ProxyManager;
client/src/service.rs
+16 -16
View File
@@ -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<dyn Fn(String, String) + Send + Sync>,
on_tunnel_remote: Arc<dyn Fn(String, String) + Send + Sync>,
on_remotes_clear: Arc<dyn Fn() + Send + Sync>,
) -> 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;
client/src/run_id.rs → client/src/session_id.rs
+10 -8
View File
@@ -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<String> {
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)
}
client/src/proxy/manager.rs → client/src/tunnel/manager.rs
+90 -63
View File
@@ -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<Arc<BandwidthLimiter>>,
plugin: Option<Arc<dyn Plugin>>,
proxy_protocol: Option<&'static str>,
udp_cancel: Mutex<Option<oneshot::Sender<()>>>,
}
pub struct ProxyManager {
by_name: HashMap<String, ProxyEntry>,
pub struct TunnelManager {
by_name: HashMap<String, TunnelEntry>,
udp_packet_size: usize,
}
impl ProxyManager {
impl TunnelManager {
pub fn from_config(cfg: &ClientConfig) -> Result<Self> {
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()
}
client/src/tunnel/mod.rs
+4
View File
@@ -0,0 +1,4 @@
mod manager;
mod udp;
pub use manager::TunnelManager;
client/src/proxy/udp.rs → client/src/tunnel/udp.rs
+18 -18
View File
@@ -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<String>,
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::<Message>(CHANNEL_CAP);
let (from_work_tx, from_work_rx) = mpsc::channel(CHANNEL_CAP);
let (to_data_tx, mut to_data_rx) = mpsc::channel::<Message>(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::<anyhow::Error>(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<R: AsyncRead + Unpin + Send + 'static>(
async fn data_reader<R: AsyncRead + Unpin + Send + 'static>(
mut reader: R,
from_work_tx: mpsc::Sender<orbien_core::msg::UdpPacket>,
from_data_tx: mpsc::Sender<orbien_core::msg::UdpPacket>,
fail_tx: mpsc::Sender<anyhow::Error>,
) {
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;
}
}