diff --git a/lightway-client/build.rs b/lightway-client/build.rs index 327668de..aa4fbb9b 100644 --- a/lightway-client/build.rs +++ b/lightway-client/build.rs @@ -12,7 +12,13 @@ fn main() { ios: { target_os = "ios" }, tvos: { target_os = "tvos" }, // Backends - desktop: { any(windows, linux, macos) }, + desktop: { + all( + any(windows, linux, macos), + // cross compiling from desktop to mobile, not for a desktop client use case + not(feature="mobile"), + ) + }, mobile: { any(android, ios, tvos) }, // Apple platform apple: { diff --git a/lightway-client/src/lib.rs b/lightway-client/src/lib.rs index 14afe093..4c6797fd 100644 --- a/lightway-client/src/lib.rs +++ b/lightway-client/src/lib.rs @@ -26,12 +26,13 @@ use lightway_app_utils::NetworkChangeMonitor; #[cfg(feature = "postquantum")] use lightway_app_utils::args::KeyShare; use lightway_app_utils::{ - ConnectionTicker, ConnectionTickerState, DplpmtudTimer, EventStream, EventStreamCallback, - PacketCodecFactoryType, TunConfig, args::Cipher, connection_ticker_cb, + ConnectionTicker, ConnectionTickerState, ConnectionTickerTask, DplpmtudTimer, DplpmtudTimerTask, + EventStream, EventStreamCallback, PacketCodecFactoryType, TunConfig, args::Cipher, + connection_ticker_cb, }; use lightway_core::{ - BuilderPredicates, ClientContextBuilder, ClientIpConfig, Connection, ConnectionError, - ConnectionType, Event, EventCallback, IOCallbackResult, InsideIOSendCallbackArg, + BuilderPredicates, ClientConnectionBuilder, ClientContextBuilder, ClientIpConfig, Connection, + ConnectionError, ConnectionType, Event, EventCallback, IOCallbackResult, InsideIOSendCallbackArg, InsideIpConfig, OutsidePacket, State, ipv4_update_destination, ipv4_update_source, }; use tokio::sync::mpsc::UnboundedReceiver; @@ -45,6 +46,8 @@ use crate::keepalive::Config as KeepaliveConfig; use crate::route_manager::{RouteManager, RouteMode}; #[cfg(batch_receive)] use lightway_core::MAX_IO_BATCH_SIZE; +use futures::future::OptionFuture; +use keepalive::KeepaliveResult; pub use lightway_core::{ AuthMethod, DEFAULT_EXPRESSLANE_KEYS_ROTATION_INTERVAL, MAX_INSIDE_MTU, MAX_OUTSIDE_MTU, PluginFactoryError, PluginFactoryList, RootCertificate, Version, @@ -840,7 +843,154 @@ async fn config_reload_task( tracing::info!("config reload task has finished"); } +/// Output of [`establish`]: the live connection plus tasks the caller still needs to spawn. +pub(crate) struct Connect { + pub(crate) conn: Arc>>>, + pub(crate) keepalive: Keepalive, + pub(crate) keepalive_task: OptionFuture>, + pub(crate) keepalive_config: KeepaliveConfig, + pub(crate) event_stream: EventStream, + pub(crate) ticker_task: ConnectionTickerTask, + pub(crate) pmtud_timer_task: DplpmtudTimerTask, +} + +/// Shared connection setup used by both the CLI and mobile paths. +/// +/// Builds the TLS context, establishes the connection, and wires up keepalive. +/// Platform-specific builder tweaks are injected via the two closures: +/// - `customize_ctx` — called after cipher/plugins are set, before `.build()`. +/// - `customize_conn` — called after `.with_auth()` / `.with_event_cb()`, before `.connect()`. +pub(crate) fn establish<'cert, T, FCtx, FConn>( + connection_type: ConnectionType, + outside_io: Arc, + root_ca_cert: RootCertificate<'cert>, + inside_io_opt: Option>>, + cipher: Cipher, + inside_plugins: PluginFactoryList, + outside_plugins: PluginFactoryList, + auth: AuthMethod, + outside_mtu: usize, + server_dn: Option, + enable_pmtud: bool, + keepalive_config: KeepaliveConfig, + customize_ctx: FCtx, + customize_conn: FConn, +) -> Result> +where + T: Default + Send + Sync + 'static, + FCtx: FnOnce( + ClientContextBuilder>, + ) -> ClientContextBuilder>, + FConn: FnOnce( + ClientConnectionBuilder>, + ) -> ClientConnectionBuilder>, +{ + let (event_cb, event_stream) = EventStreamCallback::new(); + let (ticker, ticker_task) = ConnectionTicker::new(); + let state = ConnectionState { + ticker, + ip_config: None, + extended: T::default(), + }; + let (pmtud_timer, pmtud_timer_task) = DplpmtudTimer::new(); + + let ctx = ClientContextBuilder::new( + connection_type, + root_ca_cert, + inside_io_opt, + Arc::new(ClientIpConfigCb), + connection_ticker_cb, + )? + .with_cipher(cipher.into())? + .with_inside_plugins(inside_plugins) + .with_outside_plugins(outside_plugins); + + let conn_builder = customize_ctx(ctx) + .build() + .start_connect(outside_io.into_io_send_callback(), outside_mtu)? + .with_auth(auth) + .with_event_cb(Box::new(event_cb)) + .when(server_dn.is_some(), |b| { + b.with_server_domain_name_validation(&server_dn.expect("checked above")) + }) + .when(connection_type.is_datagram() && enable_pmtud, |b| { + b.with_pmtud_timer(pmtud_timer) + }); + + let conn_builder = customize_conn(conn_builder); + + let conn = Arc::new(Mutex::new(conn_builder.connect(state)?)); + + let (keepalive, keepalive_task) = + Keepalive::new(keepalive_config.clone(), Arc::downgrade(&conn)); + + Ok(Connect { + conn, + keepalive, + keepalive_task, + keepalive_config, + event_stream, + ticker_task, + pmtud_timer_task, + }) +} + +#[cfg(not(feature = "mobile"))] +impl Connect { + pub(crate) fn into_client_connect( + self, + task: JoinHandle>, + inside_io: Arc>, + #[cfg(desktop)] outside_io: Arc, + connected_signal: Option>, + stop_signal: Option>, + network_change_signal: mpsc::Sender<()>, + encoding_request_signal: mpsc::Sender, + ) -> ClientConnection { + ClientConnection { + task, + conn: self.conn, + inside_io, + #[cfg(desktop)] + outside_io, + connected_signal, + stop_signal, + network_change_signal, + encoding_request_signal, + #[cfg(desktop)] + route_manager: None, + #[cfg(desktop)] + dns_manager: None, + } + } +} + +#[cfg(feature = "mobile")] +impl Connect>> { + pub(crate) fn into_client_connect( + self, + outside_io_task: JoinHandle>, + new_outside_io_sender: mpsc::Sender<()>, + join_set: JoinSet<()>, + instance_id: usize, + expresslane_event_rx: Option>, + ) -> ClientConnection { + ClientConnection { + conn: self.conn, + outside_io_task, + new_outside_io_sender, + keepalive: self.keepalive, + keepalive_task: self.keepalive_task, + keepalive_config: self.keepalive_config, + join_set, + instance_id, + expresslane_event_rx, + } + } +} + /// Represents a connection to a server. When dropped, the route table will be removed. +#[cfg(not(feature = "mobile"))] pub struct ClientConnection { task: JoinHandle>, conn: Arc>>>, @@ -857,6 +1007,7 @@ pub struct ClientConnection { dns_manager: Option, } +#[cfg(not(feature = "mobile"))] impl ClientConnection { /// Returns details about the established outside connection. #[cfg(desktop)] @@ -921,6 +1072,19 @@ impl ClientConnection { } } +#[cfg(feature = "mobile")] +pub(crate) struct ClientConnection { + pub(crate) conn: Arc>>>>>, + pub(crate) outside_io_task: JoinHandle>, + pub(crate) new_outside_io_sender: mpsc::Sender<()>, + pub(crate) keepalive: Keepalive, + pub(crate) keepalive_task: OptionFuture>, + pub(crate) keepalive_config: KeepaliveConfig, + pub(crate) join_set: JoinSet<()>, + pub(crate) instance_id: usize, + pub(crate) expresslane_event_rx: Option>, +} + #[tracing::instrument( level = "info", fields(server = server_config.server.to_string(), mode = ?server_config.mode), @@ -990,16 +1154,6 @@ pub async fn connect< } }; - let (event_cb, event_stream) = EventStreamCallback::new(); - - let (ticker, ticker_task) = ConnectionTicker::new(); - let state = ConnectionState { - ticker, - ip_config: None, - extended: Default::default(), - }; - let (pmtud_timer, pmtud_timer_task) = DplpmtudTimer::new(); - #[cfg(feature = "debug")] if config.tls_debug { set_logging_callback(|m: &str| tracing::debug!(target: "ssl_debug", m)); @@ -1017,59 +1171,57 @@ pub async fn connect< None => (None, None, None), }; - let conn_builder = ClientContextBuilder::new( - connection_type, - RootCertificate::PemBuffer(cert_content.as_bytes()), - None, - Arc::new(ClientIpConfigCb), - connection_ticker_cb, - )? - .with_cipher(cipher.into())? - .with_inside_plugins(inside_plugins) - .with_outside_plugins(outside_plugins) - .when(config.enable_expresslane, |b| { - b.with_expresslane(config.expresslane_keys_rotation_interval) - }) - .when(config.expresslane_cb.is_some(), |b| { - b.with_expresslane_cb(config.expresslane_cb.clone().unwrap()) - }) - .when(config.expresslane_metrics.is_some(), |b| { - b.with_expresslane_metrics(config.expresslane_metrics.clone().unwrap()) - }) - .build() - .start_connect( - outside_io.clone().into_io_send_callback(), - config.outside_mtu, - )? - .with_auth(auth) - .with_event_cb(Box::new(event_cb)) - .with_inside_pkt_codec(inside_io_codec) - .when_some(config.pmtud_base_mtu, |b, mtu| b.with_pmtud_base_mtu(mtu)) - .when_some(server_dn, |b, sdn| { - b.with_server_domain_name_validation(&sdn) - }) - .when(connection_type.is_datagram() && config.enable_pmtud, |b| { - b.with_pmtud_timer(pmtud_timer) - }); - - #[cfg(feature = "postquantum")] - let conn_builder = conn_builder.with_pq_crypto(config.keyshare.into()); - - #[cfg(feature = "debug")] - let conn_builder = conn_builder.when_some(config.keylog.clone(), |b, k| { - b.with_key_logger(WiresharkKeyLogger::new(k)) - }); - - let conn = Arc::new(Mutex::new(conn_builder.connect(state)?)); - let keepalive_config = keepalive::Config { interval: config.keepalive_interval, timeout: config.keepalive_timeout, continuous: config.continuous_keepalive, tracer_trigger_timeout: Some(config.tracer_packet_timeout), }; - let (keepalive, keepalive_task) = - Keepalive::new(keepalive_config.clone(), Arc::downgrade(&conn)); + + let Connect { + conn, + keepalive, + keepalive_task, + keepalive_config, + event_stream, + ticker_task, + pmtud_timer_task, + } = establish( + connection_type, + outside_io.clone(), + RootCertificate::PemBuffer(cert_content.as_bytes()), + None, + cipher, + inside_plugins, + outside_plugins, + auth, + config.outside_mtu, + server_dn, + config.enable_pmtud, + keepalive_config, + |ctx| { + ctx.when(config.enable_expresslane, |b| { + b.with_expresslane(config.expresslane_keys_rotation_interval) + }) + .when(config.expresslane_cb.is_some(), |b| { + b.with_expresslane_cb(config.expresslane_cb.clone().unwrap()) + }) + .when(config.expresslane_metrics.is_some(), |b| { + b.with_expresslane_metrics(config.expresslane_metrics.clone().unwrap()) + }) + }, + |conn_builder| { + #[cfg(feature = "postquantum")] + let conn_builder = conn_builder.with_pq_crypto(config.keyshare.into()); + #[cfg(feature = "debug")] + let conn_builder = conn_builder.when_some(config.keylog.clone(), |b, k| { + b.with_key_logger(WiresharkKeyLogger::new(k)) + }); + conn_builder + .with_inside_pkt_codec(inside_io_codec) + .when_some(config.pmtud_base_mtu, |b, mtu| b.with_pmtud_base_mtu(mtu)) + }, + )?; let (connected_tx, connected_rx) = oneshot::channel(); let (disconnected_tx, disconnected_rx) = oneshot::channel(); diff --git a/lightway-client/src/mobile/lightway.rs b/lightway-client/src/mobile/lightway.rs index d494b512..5ba5fa9c 100644 --- a/lightway-client/src/mobile/lightway.rs +++ b/lightway-client/src/mobile/lightway.rs @@ -1,22 +1,19 @@ use crate::config::{Config, ConnectionConfig}; use crate::io::outside::OutsideIO; -use crate::keepalive::{Keepalive, KeepaliveResult}; +use crate::keepalive::Keepalive; use crate::mobile::EventHandlers; use crate::mobile::{DeviceNetworkState, ExpresslaneState}; use crate::{ - ClientIpConfigCb, ClientResult, ConnectionState, inside_io_task, io, + ClientResult, ClientConnection, ConnectionState, Connect, inside_io_task, io, keepalive::Config as KeepaliveConfig, outside_io_task, }; use futures::StreamExt; -use futures::future::{FutureExt, OptionFuture, select_all}; +use futures::future::{FutureExt, select_all}; use futures::stream::{FusedStream, FuturesUnordered}; -use lightway_app_utils::{ - ConnectionTicker, DplpmtudTimer, EventStream, EventStreamCallback, TunConfig, - connection_ticker_cb, -}; +use lightway_app_utils::{EventStream, EventStreamCallback, TunConfig}; use lightway_core::{ - BuilderPredicates, ClientContextBuilder, Connection, ConnectionError, ConnectionType, Event, - EventCallback, IOCallbackResult, InsideIOSendCallback, PluginFactoryList, State, + BuilderPredicates, Connection, ConnectionError, ConnectionType, Event, EventCallback, + IOCallbackResult, InsideIOSendCallback, PluginFactoryList, State, }; use std::collections::HashMap; use std::future::Future; @@ -223,7 +220,7 @@ pub(crate) async fn async_lightway_start( event_stream_handler: event_handler.clone(), external_event_handler: external_event_handler.clone(), }) - .instrument(info_span!("LightwayConnection", instance_id = instance_id)), + .instrument(info_span!("ClientConnection", instance_id = instance_id)), ); (task.abort_handle(), task) }) @@ -244,8 +241,8 @@ pub(crate) async fn async_lightway_start( // Drop the last sender drop(online_signal_sender); - let mut non_preferred_connections: Vec<(usize, LightwayConnection)> = Vec::new(); - let mut pending_online_connections: HashMap = + let mut non_preferred_connections: Vec<(usize, ClientConnection)> = Vec::new(); + let mut pending_online_connections: HashMap = HashMap::with_capacity(server_len); let mut failed_connections = 0usize; let mut connection_error_to_return = None; @@ -254,7 +251,7 @@ pub(crate) async fn async_lightway_start( let active_connection = loop { tokio::select! { // Prioritise management commands over other branches, also make sure we add - // LightwayConnection to HashMap first to make sure when it goes online, + // ClientConnection to HashMap first to make sure when it goes online, // we can remove it from the HashMap and break the loop. biased; _ = futures::future::ready(()), if failed_connections == server_len => { @@ -310,7 +307,7 @@ pub(crate) async fn async_lightway_start( info!("Deferring connection {}", instance_id); non_preferred_connections.push((instance_id, connection)); } else { - warn!(?instance_id, "Cannot find LightwayConnection"); + warn!(?instance_id, "Cannot find ClientConnection"); } }, @@ -358,7 +355,7 @@ pub(crate) async fn async_lightway_start( .collect(), )); - let LightwayConnection { + let ClientConnection { conn, outside_io_task, new_outside_io_sender, @@ -499,7 +496,7 @@ async fn restartable_outside_io_task( } fn first_outside_io_exit( - connections: &mut HashMap, + connections: &mut HashMap, ) -> impl Future, tokio::task::JoinError>)> + '_ { if connections.is_empty() { return futures::future::Either::Left(std::future::pending()); @@ -516,7 +513,7 @@ fn first_outside_io_exit( async fn cleanup_connections( in_progress_connections_abort_handle: Vec, - completed_connections: Vec, + completed_connections: Vec, ) { for conn in in_progress_connections_abort_handle { if !conn.is_finished() { @@ -537,17 +534,6 @@ async fn cleanup_connections( info!("Cleaned up unused connections"); } -struct LightwayConnection { - conn: Arc>>>, - outside_io_task: JoinHandle>, - new_outside_io_sender: MpscSender<()>, - keepalive: Keepalive, - keepalive_task: OptionFuture>, - keepalive_config: KeepaliveConfig, - join_set: JoinSet<()>, - instance_id: usize, - expresslane_event_rx: Option>, -} struct LightwayClientConnectArgs { instance_id: usize, @@ -576,7 +562,7 @@ async fn lightway_client_connect( event_stream_handler, external_event_handler, }: LightwayClientConnectArgs, -) -> uniffi::Result { +) -> uniffi::Result { let mut join_set = JoinSet::new(); // TODO: Should be strong type error @@ -595,62 +581,16 @@ async fn lightway_client_connect( builder.build(&mut outside_plugins).await? }; + // Copy fields before borrowing connect_conf for the CA cert. + let cipher = connect_conf.cipher; + let outside_mtu = connect_conf.outside_mtu; + let mode = connect_conf.mode; let root_ca_cert = connect_conf.load_ca()?; - let inside_io = MobileInsideIo { - mtu: INTERNAL_MTU as usize, - }; let inside_io: Arc> + Send + Sync> = - Arc::new(inside_io); - - let (event_cb, event_stream) = EventStreamCallback::new(); - - let (ticker, ticker_task) = ConnectionTicker::new(); - let state: ConnectionState = ConnectionState { - ticker, - ip_config: None, - extended: None, - }; - let (pmtud_timer, pmtud_timer_task) = DplpmtudTimer::new(); - - let ConnectionConfig { - cipher, - outside_mtu, - mode, - .. - } = connect_conf; - - let conn_builder = ClientContextBuilder::new( - connection_type, - root_ca_cert, - Some(inside_io), - Arc::new(ClientIpConfigCb), - connection_ticker_cb, - )? - .with_cipher(cipher.into())? - .with_inside_plugins(inside_plugins) - .with_outside_plugins(outside_plugins) - .when(connection_type.is_datagram() && enable_expresslane, |b| { - b.with_expresslane(expresslane_keys_rotation_interval) - }) - .build() - .start_connect(outside_io.clone().into_io_send_callback(), outside_mtu)? - .with_auth(auth) - .with_event_cb(Box::new(event_cb)) - .when(server_dn.is_some(), |b| { - b.with_server_domain_name_validation(&server_dn.expect("checked in builder pattern")) - }) - .when(!sni_header.is_empty(), |b| b.with_sni_header(&sni_header)) - .when(connection_type.is_datagram() && ENABLE_PMTUD, |b| { - b.with_pmtud_timer(pmtud_timer) - }); - - #[cfg(feature = "postquantum")] - let conn_builder = conn_builder.when(true, |b| { - b.with_pq_crypto(lightway_app_utils::args::KeyShare::default().into()) - }); - - let conn = Arc::new(Mutex::new(conn_builder.connect(state)?)); + Arc::new(MobileInsideIo { + mtu: INTERNAL_MTU as usize, + }); let keepalive_config = KeepaliveConfig { interval: Duration::new(2, 0), @@ -658,8 +598,43 @@ async fn lightway_client_connect( continuous: enable_keepalive, tracer_trigger_timeout: Some(Duration::from_secs(10)), }; - let (keepalive, keepalive_task) = - Keepalive::new(keepalive_config.clone(), Arc::downgrade(&conn)); + + let Connect { + conn, + keepalive, + keepalive_task, + keepalive_config, + event_stream, + ticker_task, + pmtud_timer_task, + } = crate::establish( + connection_type, + outside_io.clone(), + root_ca_cert, + Some(inside_io), + cipher, + inside_plugins, + outside_plugins, + auth, + outside_mtu, + server_dn, + ENABLE_PMTUD, + keepalive_config, + |ctx| { + ctx.when(connection_type.is_datagram() && enable_expresslane, |b| { + b.with_expresslane(expresslane_keys_rotation_interval) + }) + }, + |conn_builder| { + let conn_builder = + conn_builder.when(!sni_header.is_empty(), |b| b.with_sni_header(&sni_header)); + #[cfg(feature = "postquantum")] + let conn_builder = conn_builder.when(true, |b| { + b.with_pq_crypto(lightway_app_utils::args::KeyShare::default().into()) + }); + conn_builder + }, + )?; let notify_keepalive_reply = Arc::new(Notify::new()); let (expresslane_event_tx, expresslane_event_rx) = if enable_expresslane && !mode.is_tcp() { @@ -700,7 +675,7 @@ async fn lightway_client_connect( .in_current_span(), ); - Ok(LightwayConnection { + Ok(ClientConnection { conn, outside_io_task, new_outside_io_sender,