diff --git a/easytier-core/Cargo.toml b/easytier-core/Cargo.toml index c67d412a..939a99fe 100644 --- a/easytier-core/Cargo.toml +++ b/easytier-core/Cargo.toml @@ -132,9 +132,9 @@ proxy-packet = [ "smoltcp/std", "smoltcp/proto-ipv4", "smoltcp/proto-ipv4-fragmentation", - "smoltcp/fragmentation-buffer-size-8192", + "smoltcp/fragmentation-buffer-size-65536", "smoltcp/assembler-max-segment-count-16", - "smoltcp/reassembly-buffer-size-8192", + "smoltcp/reassembly-buffer-size-65536", "smoltcp/reassembly-buffer-count-16", ] proxy-smoltcp-stack = [ diff --git a/easytier-core/src/foundation/mod.rs b/easytier-core/src/foundation/mod.rs index 967a6ad8..effe2897 100644 --- a/easytier-core/src/foundation/mod.rs +++ b/easytier-core/src/foundation/mod.rs @@ -3,7 +3,11 @@ //! Everything in `foundation` may be used by any layer, and nothing here may //! depend on a domain Module. See `CONTEXT.md` "Module layers". -#[cfg(any(feature = "proxy-smoltcp-stack", test))] +#[cfg(any( + feature = "proxy-smoltcp-stack", + test, + all(feature = "management-rpc", target_os = "wasi") +))] pub(crate) mod operation_broker; pub mod stats; pub(crate) mod task; diff --git a/easytier-core/src/gateway/dataplane/stack.rs b/easytier-core/src/gateway/dataplane/stack.rs index 81928fed..5f3aea6a 100644 --- a/easytier-core/src/gateway/dataplane/stack.rs +++ b/easytier-core/src/gateway/dataplane/stack.rs @@ -96,8 +96,9 @@ impl SmoltcpPlane { tcp_rx_size: 1024 * 128, tcp_tx_size: 1024 * 128, udp_rx_size: 1024 * 128, + udp_tx_size: 1024 * 128, udp_rx_meta_size: 128, - ..Default::default() + udp_tx_meta_size: 128, }), ), ); diff --git a/easytier-core/src/gateway/dataplane/tests.rs b/easytier-core/src/gateway/dataplane/tests.rs index fcc5c2b7..fabea048 100644 --- a/easytier-core/src/gateway/dataplane/tests.rs +++ b/easytier-core/src/gateway/dataplane/tests.rs @@ -526,6 +526,30 @@ async fn data_plane_udp_pingpong() { stop_data_plane_pair(&a, &b).await; } +#[tokio::test] +async fn data_plane_udp_carries_maximum_ipv4_payload() { + let (a, b) = setup_data_plane_pair().await; + let timeout = Duration::from_secs(10); + let socket_a = a.gateway.data_plane_udp_bind(0, timeout).await.unwrap(); + let socket_b = b.gateway.data_plane_udp_bind(0, timeout).await.unwrap(); + let addr_a = SocketAddr::new(a.ip.address().into(), socket_a.local_addr().port()); + let addr_b = SocketAddr::new(b.ip.address().into(), socket_b.local_addr().port()); + + socket_b.send_to(b"warmup", addr_a).await.unwrap(); + let payload = vec![0x5a; 65_507]; + socket_a.send_to(&payload, addr_b).await.unwrap(); + let mut received = vec![0; payload.len()]; + let (len, from) = tokio::time::timeout(timeout, socket_b.recv_from(&mut received)) + .await + .expect("receive maximum UDP payload timed out") + .unwrap(); + assert_eq!(from, addr_a); + assert_eq!(len, payload.len()); + assert_eq!(received, payload); + + stop_data_plane_pair(&a, &b).await; +} + #[tokio::test] async fn udp_socket_drop_releases_every_destination_flow() { let (a, b) = setup_data_plane_pair().await; diff --git a/easytier-core/src/gateway/port_forward.rs b/easytier-core/src/gateway/port_forward.rs index e58792b3..84bb9840 100644 --- a/easytier-core/src/gateway/port_forward.rs +++ b/easytier-core/src/gateway/port_forward.rs @@ -12,7 +12,11 @@ use std::{ use crossbeam::atomic::AtomicCell; use dashmap::DashMap; use quanta::Instant; -use tokio::{select, sync::Mutex, task::JoinSet}; +use tokio::{ + select, + sync::{Mutex, OwnedMutexGuard, OwnedSemaphorePermit, Semaphore}, + task::JoinSet, +}; use tokio_util::{ sync::{CancellationToken, DropGuard}, task::AbortOnDropHandle, @@ -36,6 +40,12 @@ use crate::{ }, }; +const MAX_UDP_PAYLOAD_SIZE: usize = 65_507; +// A remote flow owns roughly 320 KiB of smoltcp and response buffers. +// Keep the per-instance worst case near 80 MiB instead of allowing a +// source-port flood to grow the WASM heap without bound. +const MAX_ACTIVE_UDP_CLIENTS: usize = 256; + #[derive(Debug, Eq, PartialEq, Hash, Clone)] struct UdpClientKey { client_addr: SocketAddr, @@ -75,6 +85,12 @@ where { flow: Arc>, last_active: AtomicCell, + _slot: Arc, +} + +struct UdpClientReservation { + slot: OwnedSemaphorePermit, + admission: OwnedMutexGuard<()>, } pub(crate) struct PortForwardAdapter @@ -92,6 +108,8 @@ where cancel_tokens: Arc>, udp_clients: Arc>>>, udp_response_tasks: Arc>>, + udp_client_admission: Arc>, + udp_client_slots: Arc, consumer_lease: Mutex>, } @@ -118,6 +136,8 @@ where cancel_tokens: Arc::new(DashMap::new()), udp_clients: Arc::new(DashMap::new()), udp_response_tasks: Arc::new(DashMap::new()), + udp_client_admission: Arc::new(Mutex::new(())), + udp_client_slots: Arc::new(Semaphore::new(MAX_ACTIVE_UDP_CLIENTS)), consumer_lease: Mutex::new(None), }) } @@ -281,14 +301,16 @@ where let socket_context = self.socket_context.clone(); let udp_clients = self.udp_clients.clone(); let response_tasks = self.udp_response_tasks.clone(); + let client_admission = self.udp_client_admission.clone(); + let client_slots = self.udp_client_slots.clone(); self.tasks.lock().unwrap().spawn(async move { let adapter = UdpFlowFactory { data_plane, host, socket_context, }; + let mut buf = vec![0u8; MAX_UDP_PAYLOAD_SIZE]; loop { - let mut buf = vec![0u8; 8192]; let (len, client_addr) = select! { biased; _ = cancel.cancelled() => break, @@ -318,20 +340,40 @@ where continue; } }; + let Some(reservation) = reserve_udp_client_slot( + &cancel, + &client_admission, + &client_slots, + &udp_clients, + &response_tasks, + ) + .await + else { + tracing::trace!( + ?client_addr, + "UDP port-forward client limit remains full" + ); + continue; + }; + let UdpClientReservation { slot, admission } = reservation; + let slot = Arc::new(slot); let client = Arc::new(UdpClientInfo { flow: flow.clone(), last_active: AtomicCell::new(Instant::now()), + _slot: slot.clone(), }); udp_clients.insert(key.clone(), client.clone()); let inbound = socket.clone(); let response_flow = flow.clone(); let response_client = client_addr; + let response_slot = slot; response_tasks.insert( key.clone(), AbortOnDropHandle::new(tokio::spawn(async move { + let _slot = response_slot; + let mut buf = vec![0u8; MAX_UDP_PAYLOAD_SIZE]; loop { - let mut buf = vec![0u8; 8192]; match response_flow.recv_from(&mut buf).await { Ok((len, remote)) => { tracing::trace!( @@ -355,6 +397,7 @@ where } })), ); + drop(admission); client } }; @@ -404,6 +447,45 @@ where } } +async fn reserve_udp_client_slot( + cancel: &CancellationToken, + admission: &Arc>, + slots: &Arc, + clients: &DashMap>>, + response_tasks: &DashMap>, +) -> Option +where + H: VirtualUdpSocketFactory, +{ + // Keep eviction, permit handoff, and publication in one critical section. + let admission = select! { + biased; + _ = cancel.cancelled() => return None, + admission = admission.clone().lock_owned() => admission, + }; + loop { + if let Ok(slot) = slots.clone().try_acquire_owned() { + return Some(UdpClientReservation { slot, admission }); + } + + let oldest = clients + .iter() + .min_by_key(|entry| entry.value().last_active.load()) + .map(|entry| entry.key().clone())?; + let Some(evicted) = clients.remove(&oldest) else { + continue; + }; + drop(response_tasks.remove(&oldest)); + drop(evicted); + let slot = select! { + biased; + _ = cancel.cancelled() => return None, + slot = slots.clone().acquire_owned() => slot.ok()?, + }; + return Some(UdpClientReservation { slot, admission }); + } +} + struct UdpFlowFactory where H: VirtualTcpSocketFactory + VirtualTcpListenerFactory + VirtualUdpSocketFactory, @@ -451,3 +533,195 @@ where Err(error) => tracing::error!(?error, ?dst_addr, "port-forward connection failed"), } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::host::testkit::{TestHost, TestUdpSocket}; + + fn udp_client_key(port: u16) -> UdpClientKey { + UdpClientKey { + client_addr: SocketAddr::from(([127, 0, 0, 1], port)), + forward: PortForwardConfig { + bind_addr: "127.0.0.1:5202".parse().unwrap(), + dst_addr: "10.144.0.20:5201".parse().unwrap(), + proto: "udp".to_owned(), + }, + } + } + + fn udp_client(slots: &Arc, last_active: Instant) -> Arc> { + udp_client_with_slot( + Arc::new(slots.clone().try_acquire_owned().unwrap()), + last_active, + ) + } + + fn udp_client_with_slot( + slot: Arc, + last_active: Instant, + ) -> Arc> { + let flow: Arc> = Arc::new(PortForwardUdpFlow::Host(Arc::new( + TestUdpSocket("127.0.0.1:20002".parse().unwrap()), + ))); + Arc::new(UdpClientInfo { + flow, + last_active: AtomicCell::new(last_active), + _slot: slot, + }) + } + + fn pending_response_task(slot: Arc) -> AbortOnDropHandle<()> { + AbortOnDropHandle::new(tokio::spawn(async move { + let _slot = slot; + std::future::pending().await + })) + } + + #[tokio::test] + async fn udp_client_capacity_evicts_least_recently_active_flow() { + let slots = Arc::new(Semaphore::new(2)); + let admission = Arc::new(Mutex::new(())); + let clients: DashMap>> = DashMap::new(); + let response_tasks: DashMap> = DashMap::new(); + let oldest = udp_client_key(40001); + let newest = udp_client_key(40002); + let now = Instant::now(); + let oldest_client = udp_client(&slots, now - Duration::from_secs(2)); + let newest_client = udp_client(&slots, now - Duration::from_secs(1)); + response_tasks.insert( + oldest.clone(), + pending_response_task(oldest_client._slot.clone()), + ); + response_tasks.insert( + newest.clone(), + pending_response_task(newest_client._slot.clone()), + ); + clients.insert(oldest.clone(), oldest_client); + clients.insert(newest.clone(), newest_client); + assert_eq!(slots.available_permits(), 0); + + let replacement = reserve_udp_client_slot( + &CancellationToken::new(), + &admission, + &slots, + &clients, + &response_tasks, + ) + .await + .unwrap(); + + assert!(!clients.contains_key(&oldest)); + assert!(!response_tasks.contains_key(&oldest)); + assert!(clients.contains_key(&newest)); + assert!(response_tasks.contains_key(&newest)); + assert_eq!(slots.available_permits(), 0); + + drop(replacement); + response_tasks.clear(); + clients.clear(); + tokio::task::yield_now().await; + assert_eq!(slots.available_permits(), 2); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn udp_client_admission_covers_client_and_response_task_publication() { + let slots = Arc::new(Semaphore::new(1)); + let admission = Arc::new(Mutex::new(())); + let clients: Arc>>> = + Arc::new(DashMap::new()); + let response_tasks: Arc>> = + Arc::new(DashMap::new()); + let oldest = udp_client_key(40001); + let oldest_client = udp_client(&slots, Instant::now()); + response_tasks.insert( + oldest.clone(), + pending_response_task(oldest_client._slot.clone()), + ); + clients.insert(oldest.clone(), oldest_client.clone()); + + let reserve = || { + let admission = admission.clone(); + let slots = slots.clone(); + let clients = clients.clone(); + let response_tasks = response_tasks.clone(); + tokio::spawn(async move { + reserve_udp_client_slot( + &CancellationToken::new(), + &admission, + &slots, + &clients, + &response_tasks, + ) + .await + }) + }; + let first = reserve(); + tokio::time::timeout(Duration::from_secs(1), async { + while clients.contains_key(&oldest) { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + + let mut second = reserve(); + assert!( + tokio::time::timeout(Duration::from_millis(50), &mut second) + .await + .is_err() + ); + + drop(oldest_client); + let first_slot = tokio::time::timeout(Duration::from_secs(1), first) + .await + .unwrap() + .unwrap() + .unwrap(); + + assert!( + tokio::time::timeout(Duration::from_millis(50), &mut second) + .await + .is_err() + ); + + let UdpClientReservation { + slot, + admission: admission_guard, + } = first_slot; + let replacement = udp_client_key(40002); + let replacement_slot = Arc::new(slot); + let replacement_client = udp_client_with_slot(replacement_slot.clone(), Instant::now()); + clients.insert(replacement.clone(), replacement_client.clone()); + + assert!( + tokio::time::timeout(Duration::from_millis(50), &mut second) + .await + .is_err() + ); + assert!(clients.contains_key(&replacement)); + + response_tasks.insert(replacement.clone(), pending_response_task(replacement_slot)); + drop(admission_guard); + + tokio::time::timeout(Duration::from_secs(1), async { + while clients.contains_key(&replacement) { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(!response_tasks.contains_key(&replacement)); + + drop(replacement_client); + let second_slot = tokio::time::timeout(Duration::from_secs(1), second) + .await + .unwrap() + .unwrap() + .unwrap(); + + drop(second_slot); + tokio::task::yield_now().await; + assert_eq!(slots.available_permits(), 1); + } +} diff --git a/easytier-core/src/host/management.rs b/easytier-core/src/host/management.rs new file mode 100644 index 00000000..dac29498 --- /dev/null +++ b/easytier-core/src/host/management.rs @@ -0,0 +1,41 @@ +use std::{io, sync::Arc, task::Poll}; + +use super::socket::{HostOperationId, HostSocketRuntime}; + +/// Mechanical process-management I/O delegated to the embedding host. +pub trait HostManagementIo: Send + Sync + 'static { + fn submit_call(&self, operation: HostOperationId, request: &[u8]) -> io::Result<()>; + + fn take_call(&self, operation: HostOperationId) -> Poll>>; + + fn cancel_operation(&self, operation: HostOperationId) -> io::Result<()>; +} + +#[derive(Clone)] +pub struct HostManagementClient +where + I: HostManagementIo, +{ + runtime: HostSocketRuntime, + io: Arc, +} + +impl HostManagementClient +where + I: HostManagementIo, +{ + pub fn new(runtime: HostSocketRuntime, io: Arc) -> Self { + Self { runtime, io } + } + + pub async fn call(&self, request: &[u8]) -> io::Result> { + self.runtime + .run_operation( + self.io.clone(), + |io, operation| io.submit_call(operation, request), + HostManagementIo::take_call, + |io, operation| io.cancel_operation(operation), + ) + .await + } +} diff --git a/easytier-core/src/host/mod.rs b/easytier-core/src/host/mod.rs index 8dd46e1a..4c955685 100644 --- a/easytier-core/src/host/mod.rs +++ b/easytier-core/src/host/mod.rs @@ -9,6 +9,8 @@ pub mod dns; pub mod environment; +#[cfg(feature = "management")] +pub mod management; pub mod packet; pub mod socket; #[cfg(test)] diff --git a/easytier-core/src/management/forwarded_rpc.rs b/easytier-core/src/management/forwarded_rpc.rs new file mode 100644 index 00000000..cd3c5bcd --- /dev/null +++ b/easytier-core/src/management/forwarded_rpc.rs @@ -0,0 +1,152 @@ +use std::marker::PhantomData; + +use async_trait::async_trait; +use bytes::Bytes; + +use crate::{ + proto::{ + api::{ + config::ConfigRpcDescriptor, + instance::{ConnectorManageRpcDescriptor, PeerManageRpcDescriptor}, + }, + rpc_types::{ + controller::BaseController, + descriptor::{MethodDescriptor, ServiceDescriptor}, + error, + handler::Handler, + }, + }, + rpc::service_registry::ServiceRegistry, +}; + +#[async_trait] +pub(crate) trait ManagementRpcForwarder: Clone + Send + Sync + 'static { + async fn forward(&self, full_method_name: String, input: Bytes) -> error::Result; +} + +struct ForwardedManagementHandler { + forwarder: F, + _descriptor: PhantomData D>, +} + +impl ForwardedManagementHandler { + fn new(forwarder: F) -> Self { + Self { + forwarder, + _descriptor: PhantomData, + } + } +} + +impl Clone for ForwardedManagementHandler +where + F: Clone, +{ + fn clone(&self) -> Self { + Self::new(self.forwarder.clone()) + } +} + +#[async_trait] +impl Handler for ForwardedManagementHandler +where + F: ManagementRpcForwarder, + D: ServiceDescriptor + Default + 'static, +{ + type Descriptor = D; + type Controller = BaseController; + + async fn call( + &self, + _: Self::Controller, + method: D::Method, + input: Bytes, + ) -> error::Result { + let descriptor = D::default(); + let full_method_name = if descriptor.package().is_empty() { + format!("{}.{}", descriptor.proto_name(), method.proto_name()) + } else { + format!( + "{}.{}.{}", + descriptor.package(), + descriptor.proto_name(), + method.proto_name() + ) + }; + self.forwarder.forward(full_method_name, input).await + } +} + +/// Registers the instance-management services supported by the bound WASI RPC ABI. +pub(crate) fn register_forwarded_instance_management_rpc( + forwarder: F, + registry: &ServiceRegistry, +) where + F: ManagementRpcForwarder, +{ + registry.register( + ForwardedManagementHandler::::new(forwarder.clone()), + "", + ); + registry.register( + ForwardedManagementHandler::::new(forwarder.clone()), + "", + ); + registry.register( + ForwardedManagementHandler::::new(forwarder), + "", + ); +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + + use super::*; + + #[derive(Clone, Default)] + struct RecordingForwarder { + calls: Arc>>, + } + + #[async_trait] + impl ManagementRpcForwarder for RecordingForwarder { + async fn forward(&self, full_method_name: String, input: Bytes) -> error::Result { + self.calls + .lock() + .unwrap() + .push((full_method_name, input.clone())); + Ok(input) + } + } + + #[tokio::test] + async fn registers_and_forwards_bound_instance_management_services() { + let forwarder = RecordingForwarder::default(); + let registry = ServiceRegistry::new(); + register_forwarded_instance_management_rpc(forwarder.clone(), ®istry); + + let methods = [ + "api.instance.PeerManageRpc.ListPeer", + "api.instance.ConnectorManageRpc.ListConnector", + "api.config.ConfigRpc.GetConfig", + ]; + for (index, full_method_name) in methods.into_iter().enumerate() { + let descriptor = registry + .resolve_method("", full_method_name) + .unwrap_or_else(|| panic!("missing forwarded method: {full_method_name}")); + let request = Bytes::from(vec![index as u8]); + let response = registry + .call_method(descriptor, BaseController::default(), request.clone()) + .await + .unwrap(); + assert_eq!(response, request); + } + + let calls = forwarder.calls.lock().unwrap(); + assert_eq!(calls.len(), methods.len()); + for ((actual, _), expected) in calls.iter().zip(methods) { + assert_eq!(actual, expected); + } + } +} diff --git a/easytier-core/src/management/full/mod.rs b/easytier-core/src/management/full/mod.rs index 837bb3f8..b5114e0d 100644 --- a/easytier-core/src/management/full/mod.rs +++ b/easytier-core/src/management/full/mod.rs @@ -40,6 +40,8 @@ pub use process_rpc::{ ConfigFileStorage, InstanceMutationHooks, InstanceMutationResult, ProcessManagement, ProcessManagementRpc, UnsupportedConfigFileStorage, }; +#[cfg(target_os = "wasi")] +pub(crate) use web_client::WebClientBackend; pub use web_client::{ConfigServerEndpoint, WebClient, WebClientConfig}; pub use super::instance_rpc::full::call_instance_json_rpc; diff --git a/easytier-core/src/management/full/web_client.rs b/easytier-core/src/management/full/web_client.rs index dd5b3c58..0bdf58e4 100644 --- a/easytier-core/src/management/full/web_client.rs +++ b/easytier-core/src/management/full/web_client.rs @@ -3,6 +3,7 @@ use std::sync::{ atomic::{AtomicBool, Ordering}, }; +use async_trait::async_trait; use easytier_proto::{ rpc_types::controller::BaseController, web::{ @@ -93,24 +94,31 @@ pub struct WebClientConfig { pub secure_mode: bool, } -struct WebClientController +#[async_trait] +pub(crate) trait WebClientBackend: Send + Sync + 'static { + fn register(&self, registry: &ServiceRegistry); + + async fn instance_ids(&self) -> anyhow::Result>; +} + +struct NativeWebClientBackend where F: InstanceFactory, { - config: WebClientConfig, instances: Arc>, hooks: Arc, storage: Arc, logger: Arc, } -impl WebClientController +#[async_trait] +impl WebClientBackend for NativeWebClientBackend where F: InstanceFactory, CreateContext = ()>, F::Error: std::fmt::Debug + std::fmt::Display + Send + Sync + 'static, H: CoreInstanceHost, { - fn register_management_entry(&self, registry: &ServiceRegistry) { + fn register(&self, registry: &ServiceRegistry) { register_management_rpc( self.instances.clone(), registry, @@ -119,17 +127,24 @@ where self.logger.clone(), ); } + + async fn instance_ids(&self) -> anyhow::Result> { + Ok(self.instances.instance_ids()) + } +} + +struct WebClientController { + config: WebClientConfig, + backend: Arc, } /// Portable config-server client. Hosts only supply identity and adapters. -pub struct WebClient -where - F: InstanceFactory, -{ - _controller: Arc>, +pub struct WebClient { + _controller: Arc, _tasks: AbortOnDropHandle<()>, - _manager_guard: DaemonGuard, + _manager_guard: Option, connected: Arc, + _factory: std::marker::PhantomData, } impl WebClient @@ -147,15 +162,37 @@ where logger: Arc, ) -> Self { let manager_guard = instances.register_daemon(); - let controller = Arc::new(WebClientController { - config, + let backend = Arc::new(NativeWebClientBackend { instances, hooks, storage, logger, }); + Self::start(connector, config, backend, Some(manager_guard)) + } +} + +#[cfg(target_os = "wasi")] +impl WebClient<()> { + pub(crate) fn with_backend( + connector: T, + config: WebClientConfig, + backend: Arc, + ) -> Self { + Self::start(connector, config, backend, None) + } +} + +impl WebClient { + fn start( + connector: T, + config: WebClientConfig, + backend: Arc, + manager_guard: Option, + ) -> Self { + let controller = Arc::new(WebClientController { config, backend }); let connected = Arc::new(AtomicBool::new(false)); - let tasks = AbortOnDropHandle::new(tokio::spawn(Self::routine( + let tasks = AbortOnDropHandle::new(tokio::spawn(web_client_routine( controller.clone(), connected.clone(), Box::new(connector), @@ -166,93 +203,7 @@ where _tasks: tasks, _manager_guard: manager_guard, connected, - } - } - - async fn routine( - controller: Arc>, - connected: Arc, - connector: Box, - ) { - loop { - let connection = match connect_config_server(connector.as_ref(), CONNECT_TIMEOUT).await - { - Ok(connection) => connection, - Err(error) => { - tracing::warn!(%error, "failed to connect to config server; retrying"); - time::sleep(RETRY_INTERVAL).await; - continue; - } - }; - - connected.store(true, Ordering::Release); - tracing::info!(?connection, "connected to config server"); - let mut session = WebClientSession::new(connection, controller.clone()); - let support_encryption = - match time::timeout(FEATURE_TIMEOUT, session.get_feature()).await { - Ok(Ok(feature)) => feature.support_encryption, - Ok(Err(error)) => { - tracing::warn!(%error, "GetFeature RPC failed; using legacy tunnel"); - false - } - Err(_) => { - tracing::warn!("GetFeature RPC timed out; using legacy tunnel"); - false - } - }; - - if support_encryption && web_security::web_secure_tunnel_supported() { - drop(session); - let connection = match connect_config_server(connector.as_ref(), CONNECT_TIMEOUT) - .await - { - Ok(connection) => connection, - Err(error) => { - connected.store(false, Ordering::Release); - tracing::warn!(%error, "failed to reconnect secure config-server tunnel"); - time::sleep(RETRY_INTERVAL).await; - continue; - } - }; - let connection = match web_security::upgrade_client_tunnel(connection).await { - Ok(connection) => connection, - Err(error) => { - connected.store(false, Ordering::Release); - tracing::warn!(%error, "config-server secure handshake failed"); - time::sleep(RETRY_INTERVAL).await; - continue; - } - }; - let mut session = WebClientSession::new(connection, controller.clone()); - session.start_heartbeat().await; - session.wait().await; - connected.store(false, Ordering::Release); - continue; - } - - if support_encryption { - if controller.config.secure_mode { - connected.store(false, Ordering::Release); - tracing::warn!( - "secure mode requires web secure-tunnel support in the local build" - ); - time::sleep(RETRY_INTERVAL).await; - continue; - } - tracing::warn!( - "server supports encryption but the local build is using a legacy tunnel" - ); - } - if controller.config.secure_mode { - connected.store(false, Ordering::Release); - tracing::warn!("secure mode requires config-server encryption support"); - time::sleep(RETRY_INTERVAL).await; - continue; - } - - session.start_heartbeat().await; - session.wait().await; - connected.store(false, Ordering::Release); + _factory: std::marker::PhantomData, } } @@ -261,26 +212,100 @@ where } } -struct WebClientSession -where - F: InstanceFactory, -{ +async fn web_client_routine( + controller: Arc, + connected: Arc, + connector: Box, +) { + loop { + let connection = match connect_config_server(connector.as_ref(), CONNECT_TIMEOUT).await { + Ok(connection) => connection, + Err(error) => { + tracing::warn!(%error, "failed to connect to config server; retrying"); + time::sleep(RETRY_INTERVAL).await; + continue; + } + }; + + connected.store(true, Ordering::Release); + tracing::info!(?connection, "connected to config server"); + let mut session = WebClientSession::new(connection, controller.clone()); + let support_encryption = match time::timeout(FEATURE_TIMEOUT, session.get_feature()).await { + Ok(Ok(feature)) => feature.support_encryption, + Ok(Err(error)) => { + tracing::warn!(%error, "GetFeature RPC failed; using legacy tunnel"); + false + } + Err(_) => { + tracing::warn!("GetFeature RPC timed out; using legacy tunnel"); + false + } + }; + + if support_encryption && web_security::web_secure_tunnel_supported() { + drop(session); + let connection = match connect_config_server(connector.as_ref(), CONNECT_TIMEOUT).await + { + Ok(connection) => connection, + Err(error) => { + connected.store(false, Ordering::Release); + tracing::warn!(%error, "failed to reconnect secure config-server tunnel"); + time::sleep(RETRY_INTERVAL).await; + continue; + } + }; + let connection = match web_security::upgrade_client_tunnel(connection).await { + Ok(connection) => connection, + Err(error) => { + connected.store(false, Ordering::Release); + tracing::warn!(%error, "config-server secure handshake failed"); + time::sleep(RETRY_INTERVAL).await; + continue; + } + }; + let mut session = WebClientSession::new(connection, controller.clone()); + session.start_heartbeat().await; + session.wait().await; + connected.store(false, Ordering::Release); + continue; + } + + if support_encryption { + if controller.config.secure_mode { + connected.store(false, Ordering::Release); + tracing::warn!("secure mode requires web secure-tunnel support in the local build"); + time::sleep(RETRY_INTERVAL).await; + continue; + } + tracing::warn!( + "server supports encryption but the local build is using a legacy tunnel" + ); + } + if controller.config.secure_mode { + connected.store(false, Ordering::Release); + tracing::warn!("secure mode requires config-server encryption support"); + time::sleep(RETRY_INTERVAL).await; + continue; + } + + session.start_heartbeat().await; + session.wait().await; + connected.store(false, Ordering::Release); + } +} + +struct WebClientSession { rpc: BidirectRpcManager, - controller: Arc>, + controller: Arc, heartbeat_started: AtomicBool, tasks: Mutex>, } -impl WebClientSession -where - F: InstanceFactory, CreateContext = ()>, - F::Error: std::fmt::Debug + std::fmt::Display + Send + Sync + 'static, - H: CoreInstanceHost, -{ - fn new(tunnel: Box, controller: Arc>) -> Self { +impl WebClientSession { + fn new(tunnel: Box, controller: Arc) -> Self { let rpc = BidirectRpcManager::new(); rpc.run_with_tunnel(tunnel); - controller.register_management_entry(rpc.rpc_server().registry()); + controller.backend.register(rpc.rpc_server().registry()); Self { rpc, controller, @@ -299,7 +324,7 @@ where fn heartbeat_routine( rpc: &BidirectRpcManager, - controller: Weak>, + controller: Weak, tasks: &mut JoinSet<()>, ) { let controller = controller.upgrade().expect("web client controller"); @@ -321,6 +346,13 @@ where let Some(controller) = controller.upgrade() else { break; }; + let running_network_instances = match controller.backend.instance_ids().await { + Ok(instance_ids) => instance_ids.into_iter().map(Into::into).collect(), + Err(error) => { + tracing::error!(%error, "failed to list config-server instances"); + break; + } + }; let request = HeartbeatRequest { machine_id: Some(machine_id.into()), inst_id: Some(session_id.into()), @@ -330,12 +362,7 @@ where report_time: chrono::Local::now().to_rfc3339(), device_os: Some(device_os.clone()), support_config_source: true, - running_network_instances: controller - .instances - .instance_ids() - .into_iter() - .map(Into::into) - .collect(), + running_network_instances, }; match client.heartbeat(BaseController::default(), request).await { diff --git a/easytier-core/src/management/instance_rpc/full.rs b/easytier-core/src/management/instance_rpc/full.rs index eb9bc7e7..8b398d2d 100644 --- a/easytier-core/src/management/instance_rpc/full.rs +++ b/easytier-core/src/management/instance_rpc/full.rs @@ -35,7 +35,7 @@ use crate::{ peers::credential_manager::{CredentialCreateOptions, CredentialInfo as CoreCredentialInfo}, }; -use super::InstanceManagementRpc; +use super::{InstanceManagementRpc, ReadOnlyInstanceResolver, ResolvedInstanceManagementRpc}; use crate::management::{ full::{apply_config_patch, packet_proxy}, resolve_instance, @@ -347,7 +347,7 @@ where _: BaseController, _: GetGlobalPeerMapRequest, ) -> rpc_types::error::Result { - let instance = resolve_instance(&self.manager, None).map_err(|error| { + let instance = resolve_instance(self.manager(), None).map_err(|error| { if error.to_string().contains("please specify the instance ID") { anyhow::anyhow!( "PeerCenter management RPC cannot select an instance automatically when \ @@ -371,10 +371,9 @@ where } #[async_trait::async_trait] -impl ConfigRpc for InstanceManagementRpc +impl ConfigRpc for ResolvedInstanceManagementRpc where - F: InstanceFactory>, - H: CoreInstanceHost, + R: ReadOnlyInstanceResolver, { type Controller = BaseController; @@ -401,6 +400,7 @@ where .ok_or_else(|| anyhow::anyhow!("shared TOML configuration is not available"))?; Ok(GetConfigResponse { config: Some(network_config_from_toml(&config)), + toml_config: config.dump(), }) } } diff --git a/easytier-core/src/management/instance_rpc/mod.rs b/easytier-core/src/management/instance_rpc/mod.rs index b7c99112..0e7e6485 100644 --- a/easytier-core/src/management/instance_rpc/mod.rs +++ b/easytier-core/src/management/instance_rpc/mod.rs @@ -33,15 +33,26 @@ pub(super) mod full; pub(super) mod packet_proxy; mod projection; -/// One process-level implementation for Instance-targeted management RPC. -pub struct InstanceManagementRpc +#[doc(hidden)] +pub trait ReadOnlyInstanceResolver: Clone + Send + Sync + 'static { + type Host: CoreInstanceHost; + + fn resolve( + &self, + identifier: Option<&easytier_proto::api::instance::InstanceIdentifier>, + ) -> rpc_types::error::Result>>; +} + +/// Resolver used by the public process-level management RPC type. +#[doc(hidden)] +pub struct ManagerInstanceResolver where F: InstanceFactory, { manager: Arc>, } -impl Clone for InstanceManagementRpc +impl Clone for ManagerInstanceResolver where F: InstanceFactory, { @@ -52,23 +63,146 @@ where } } -impl InstanceManagementRpc +impl ReadOnlyInstanceResolver for ManagerInstanceResolver +where + F: InstanceFactory>, + H: CoreInstanceHost, +{ + type Host = H; + + fn resolve( + &self, + identifier: Option<&easytier_proto::api::instance::InstanceIdentifier>, + ) -> rpc_types::error::Result>> { + resolve_instance(&self.manager, identifier).map_err(Into::into) + } +} + +#[cfg(target_os = "wasi")] +pub(super) struct BoundInstanceResolver +where + H: CoreInstanceHost, +{ + instance: Arc>, +} + +#[cfg(target_os = "wasi")] +impl Clone for BoundInstanceResolver +where + H: CoreInstanceHost, +{ + fn clone(&self) -> Self { + Self { + instance: self.instance.clone(), + } + } +} + +#[cfg(target_os = "wasi")] +impl ReadOnlyInstanceResolver for BoundInstanceResolver +where + H: CoreInstanceHost, +{ + type Host = H; + + fn resolve( + &self, + identifier: Option<&easytier_proto::api::instance::InstanceIdentifier>, + ) -> rpc_types::error::Result>> { + validate_bound_identifier( + identifier, + self.instance.instance_id(), + self.instance.instance_name(), + )?; + Ok(self.instance.clone()) + } +} + +/// Instance-targeted management RPC backed by a selector resolver. +#[doc(hidden)] +pub struct ResolvedInstanceManagementRpc { + resolver: R, +} + +impl Clone for ResolvedInstanceManagementRpc +where + R: Clone, +{ + fn clone(&self) -> Self { + Self { + resolver: self.resolver.clone(), + } + } +} + +impl ResolvedInstanceManagementRpc +where + R: ReadOnlyInstanceResolver, +{ + fn instance( + &self, + identifier: Option<&easytier_proto::api::instance::InstanceIdentifier>, + ) -> rpc_types::error::Result>> { + self.resolver.resolve(identifier) + } +} + +/// One process-level implementation for Instance-targeted management RPC. +pub type InstanceManagementRpc = ResolvedInstanceManagementRpc>; + +impl ResolvedInstanceManagementRpc> where F: InstanceFactory>, H: CoreInstanceHost, { pub fn new(manager: Arc>) -> Self { - Self { manager } + Self { + resolver: ManagerInstanceResolver { manager }, + } } - fn instance( - &self, - identifier: Option<&easytier_proto::api::instance::InstanceIdentifier>, - ) -> rpc_types::error::Result>> { - resolve_instance(&self.manager, identifier).map_err(Into::into) + #[cfg(feature = "management")] + pub(super) fn manager(&self) -> &Arc> { + &self.resolver.manager } } +#[cfg(target_os = "wasi")] +pub(super) fn bound_rpc( + instance: Arc>, +) -> ResolvedInstanceManagementRpc> +where + H: CoreInstanceHost, +{ + ResolvedInstanceManagementRpc { + resolver: BoundInstanceResolver { instance }, + } +} + +#[cfg(any(test, target_os = "wasi"))] +fn validate_bound_identifier( + identifier: Option<&easytier_proto::api::instance::InstanceIdentifier>, + instance_id: uuid::Uuid, + instance_name: &str, +) -> anyhow::Result<()> { + use easytier_proto::api::instance::instance_identifier::Selector; + + let matches = match identifier.and_then(|identifier| identifier.selector.as_ref()) { + None + | Some(Selector::InstanceSelector( + easytier_proto::api::instance::instance_identifier::InstanceSelector { name: None }, + )) => true, + Some(Selector::Id(selected)) => uuid::Uuid::from(*selected) == instance_id, + Some(Selector::InstanceSelector(selector)) => { + selector.name.as_deref() == Some(instance_name) + } + }; + if !matches { + anyhow::bail!("instance selector does not match the bound WASM instance"); + } + Ok(()) +} + fn foreign_network_info_to_api(info: ForeignNetworkEntryInfo) -> ForeignNetworkEntryPb { ForeignNetworkEntryPb { network_secret_digest: info.network_secret_digest, @@ -129,10 +263,9 @@ fn connector_snapshots_to_api(snapshots: Vec) -> Vec PeerManageRpc for InstanceManagementRpc +impl PeerManageRpc for ResolvedInstanceManagementRpc where - F: InstanceFactory>, - H: CoreInstanceHost, + R: ReadOnlyInstanceResolver, { type Controller = BaseController; @@ -306,10 +439,9 @@ where } #[async_trait::async_trait] -impl ConnectorManageRpc for InstanceManagementRpc +impl ConnectorManageRpc for ResolvedInstanceManagementRpc where - F: InstanceFactory>, - H: CoreInstanceHost, + R: ReadOnlyInstanceResolver, { type Controller = BaseController; @@ -325,3 +457,44 @@ where }) } } + +#[cfg(test)] +mod tests { + use easytier_proto::{ + api::instance::{ + InstanceIdentifier, + instance_identifier::{InstanceSelector, Selector}, + }, + common::Uuid as UuidPb, + }; + + use super::validate_bound_identifier; + + #[test] + fn bound_identifier_accepts_only_the_current_instance() { + let instance_id = uuid::Uuid::new_v4(); + let by_id = |id| InstanceIdentifier { + selector: Some(Selector::Id(UuidPb::from(id))), + }; + let by_name = |name: &str| InstanceIdentifier { + selector: Some(Selector::InstanceSelector(InstanceSelector { + name: Some(name.to_owned()), + })), + }; + + assert!(validate_bound_identifier(None, instance_id, "current").is_ok()); + assert!( + validate_bound_identifier(Some(&by_id(instance_id)), instance_id, "current").is_ok() + ); + assert!( + validate_bound_identifier(Some(&by_name("current")), instance_id, "current").is_ok() + ); + assert!( + validate_bound_identifier(Some(&by_id(uuid::Uuid::new_v4())), instance_id, "current") + .is_err() + ); + assert!( + validate_bound_identifier(Some(&by_name("other")), instance_id, "current").is_err() + ); + } +} diff --git a/easytier-core/src/management/mod.rs b/easytier-core/src/management/mod.rs index 9cad5205..f0f5e7d2 100644 --- a/easytier-core/src/management/mod.rs +++ b/easytier-core/src/management/mod.rs @@ -1,5 +1,7 @@ //! Process-level management over the canonical Instance collection. +#[cfg(all(feature = "management", any(test, target_os = "wasi")))] +mod forwarded_rpc; #[cfg(feature = "management")] mod full; mod instance_rpc; @@ -13,11 +15,19 @@ use crate::{ instance::{CoreInstance, CoreInstanceHost, manager::InstanceFactory}, rpc::service_registry::ServiceRegistry, }; +#[cfg(all(feature = "management", target_os = "wasi"))] +use easytier_proto::api::config::ConfigRpcServer; use easytier_proto::api::instance::{ConnectorManageRpcServer, PeerManageRpcServer}; pub use crate::instance::manager::{ ConfigFileControl, ConfigFilePermission, DaemonGuard, InstanceManager, ProcessRuntimeProvider, }; +#[cfg(all(feature = "management", target_os = "wasi"))] +pub(crate) use forwarded_rpc::{ + ManagementRpcForwarder, register_forwarded_instance_management_rpc, +}; +#[cfg(all(feature = "management", target_os = "wasi"))] +pub(crate) use full::WebClientBackend; #[cfg(feature = "management")] pub use full::remote_client; #[cfg(feature = "management")] @@ -50,3 +60,17 @@ pub fn register_read_only_management_rpc( registry.register(PeerManageRpcServer::new(rpc.clone()), ""); registry.register(ConnectorManageRpcServer::new(rpc), ""); } + +#[cfg(target_os = "wasi")] +pub(crate) fn register_bound_management_rpc( + instance: Arc>, + registry: &ServiceRegistry, +) where + H: CoreInstanceHost, +{ + let rpc = instance_rpc::bound_rpc(instance); + registry.register(PeerManageRpcServer::new(rpc.clone()), ""); + registry.register(ConnectorManageRpcServer::new(rpc.clone()), ""); + #[cfg(feature = "management")] + registry.register(ConfigRpcServer::new(rpc), ""); +} diff --git a/easytier-core/src/rpc/client.rs b/easytier-core/src/rpc/client.rs index 52d03abf..f35b51b8 100644 --- a/easytier-core/src/rpc/client.rs +++ b/easytier-core/src/rpc/client.rs @@ -18,17 +18,15 @@ use crate::{ time::timeout, }, proto::{ - common::{ - CompressionAlgoPb, RpcCompressionInfo, RpcDescriptor, RpcPacket, RpcRequest, - RpcResponse, - }, + common::{RpcCompressionInfo, RpcDescriptor, RpcPacket, RpcRequest, RpcResponse}, rpc_types::controller::Controller, rpc_types::descriptor::MethodDescriptor, rpc_types::error::{Error, Result}, rpc_types::{__rt::RpcClientFactory, descriptor::ServiceDescriptor, handler::Handler}, }, rpc::packet::{ - BuildRpcPacketArgs, PacketMerger, build_rpc_packet, compress_packet, decompress_packet, + BuildRpcPacketArgs, PacketMerger, accepted_compression_algo, build_rpc_packet, + compress_packet, decompress_packet, }, tunnel::{ Tunnel, TunnelError, ZCPacketStream, @@ -339,7 +337,7 @@ impl Client { trace_id: ctrl.trace_id(), compression_info: RpcCompressionInfo { algo: c_algo.into(), - accepted_algo: CompressionAlgoPb::Zstd.into(), + accepted_algo: accepted_compression_algo().into(), }, }); let timeout_dur = std::time::Duration::from_millis(ctrl.timeout_ms() as u64); diff --git a/easytier-core/src/rpc/dispatch.rs b/easytier-core/src/rpc/dispatch.rs new file mode 100644 index 00000000..b6c6b998 --- /dev/null +++ b/easytier-core/src/rpc/dispatch.rs @@ -0,0 +1,50 @@ +//! Transport-neutral dispatch for one decoded RPC request. + +use bytes::Bytes; +use std::time::Duration; + +use crate::{ + foundation::time::timeout, + proto::{ + common::{RpcDescriptor, RpcRequest, TunnelInfo}, + rpc_types::{ + controller::{BaseController, Controller as _}, + error::Result, + }, + }, + rpc::service_registry::ServiceRegistry, +}; + +pub(crate) async fn dispatch_request( + registry: &ServiceRegistry, + descriptor: RpcDescriptor, + request: RpcRequest, + tunnel_info: Option, +) -> Result { + dispatch_payload( + registry, + descriptor, + Bytes::from(request.request), + Some(Duration::from_millis(request.timeout_ms as u64)), + tunnel_info, + ) + .await +} + +pub(crate) async fn dispatch_payload( + registry: &ServiceRegistry, + descriptor: RpcDescriptor, + raw_request: Bytes, + timeout_duration: Option, + tunnel_info: Option, +) -> Result { + let mut controller = BaseController::default(); + controller.set_raw_input(raw_request.clone()); + controller.set_tunnel_info(tunnel_info); + let call = registry.call_method(descriptor, controller.clone(), raw_request); + let response = match timeout_duration { + Some(duration) => timeout(duration, call).await??, + None => call.await?, + }; + Ok(controller.get_raw_output().unwrap_or(response)) +} diff --git a/easytier-core/src/rpc/mod.rs b/easytier-core/src/rpc/mod.rs index 77baf687..3b35f3e6 100644 --- a/easytier-core/src/rpc/mod.rs +++ b/easytier-core/src/rpc/mod.rs @@ -4,6 +4,9 @@ pub type RpcController = crate::proto::rpc_types::controller::BaseController; pub mod bidirect; pub mod client; +pub(crate) mod dispatch; +#[cfg(all(feature = "management-rpc", any(test, target_os = "wasi")))] +pub(crate) mod operation; pub mod packet; pub mod server; pub mod service_registry; diff --git a/easytier-core/src/rpc/operation.rs b/easytier-core/src/rpc/operation.rs new file mode 100644 index 00000000..efbe55a7 --- /dev/null +++ b/easytier-core/src/rpc/operation.rs @@ -0,0 +1,430 @@ +//! Asynchronous local RPC operations backed by the shared operation broker. + +use std::sync::{Arc, Mutex}; + +use prost::Message; + +use crate::{ + foundation::operation_broker::{ + AccessError as BrokerAccessError, AdmissionError, OperationBroker, OperationId, + }, + proto::{ + common::{DirectRpcRequest, RpcResponse}, + rpc_types::error, + }, + rpc::{dispatch::dispatch_payload, service_registry::ServiceRegistry}, +}; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[repr(transparent)] +pub(crate) struct RpcOperationId(OperationId); + +impl RpcOperationId { + #[cfg(target_os = "wasi")] + pub(crate) fn from_raw(value: u64) -> Option { + OperationId::from_raw(value).map(Self) + } + + #[cfg(target_os = "wasi")] + pub(crate) fn get(self) -> u64 { + self.0.get() + } +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum RpcSubmitError { + #[error("invalid RPC request protobuf: {0}")] + Decode(#[from] prost::DecodeError), + #[error("RPC full method name is required")] + MissingMethod, + #[error("too many outstanding RPC operations")] + AtCapacity, + #[error("RPC operation ID space is exhausted")] + IdExhausted, + #[error("RPC submission requires an active Tokio runtime")] + ExecutorUnavailable, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)] +pub(crate) enum RpcAccessError { + #[error("unknown RPC operation")] + Missing, + #[error("RPC response is pending")] + Pending, +} + +struct RpcOperationState { + broker: OperationBroker<(), Vec, ()>, +} + +pub(crate) struct RpcOperationSession { + registry: Arc, + state: Arc>, + max_response_len: usize, +} + +impl RpcOperationSession { + pub(crate) fn new( + registry: Arc, + max_operations: usize, + max_response_len: usize, + ) -> Self { + Self { + registry, + state: Arc::new(Mutex::new(RpcOperationState { + broker: OperationBroker::new(max_operations), + })), + max_response_len, + } + } + + pub(crate) fn submit_encoded( + &self, + encoded_request: &[u8], + ) -> Result { + let request = DirectRpcRequest::decode(encoded_request)?; + if request.full_method_name.is_empty() { + return Err(RpcSubmitError::MissingMethod); + } + let executor = tokio::runtime::Handle::try_current() + .map_err(|_| RpcSubmitError::ExecutorUnavailable)?; + let admission = self + .state + .lock() + .unwrap_or_else(|error| error.into_inner()) + .broker + .admit((), ()) + .map_err(|error| match error { + AdmissionError::AtCapacity => RpcSubmitError::AtCapacity, + AdmissionError::IdExhausted => RpcSubmitError::IdExhausted, + })?; + let operation_id = RpcOperationId(admission.id); + let cancellation = admission.cancellation; + let registry = self.registry.clone(); + let state = self.state.clone(); + let max_response_len = self.max_response_len; + + executor.spawn(async move { + let started = std::time::Instant::now(); + let response = tokio::select! { + _ = cancellation.cancelled() => None, + result = async { + let descriptor = registry + .resolve_method("", &request.full_method_name) + .ok_or_else(|| { + error::Error::InvalidServiceKey( + request.full_method_name.clone(), + String::new(), + ) + })?; + dispatch_payload( + registry.as_ref(), + descriptor, + request.request.into(), + request + .timeout_ms + .map(std::time::Duration::from_millis), + None, + ) + .await + } => { + Some(encode_response( + result, + started.elapsed().as_micros() as u64, + max_response_len, + )) + } + }; + let mut state = state.lock().unwrap_or_else(|error| error.into_inner()); + state + .broker + .complete_with(operation_id.0, |_, _| response.unwrap_or_default()); + }); + Ok(operation_id) + } + + pub(crate) fn response_len( + &self, + operation_id: RpcOperationId, + ) -> Result { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + state.broker.drain(usize::MAX, |_| ()); + state + .broker + .with_drained(operation_id.0, |_, _, response| response.len()) + .map_err(map_access_error) + } + + pub(crate) fn take_response_with( + &self, + operation_id: RpcOperationId, + take: impl FnOnce(&[u8]) -> Option, + ) -> Result, RpcAccessError> { + let mut state = self.state.lock().unwrap_or_else(|error| error.into_inner()); + state.broker.drain(usize::MAX, |_| ()); + state + .broker + .take_with(operation_id.0, |response| take(response)) + .map(|taken| taken.map(|taken| taken.value)) + .map_err(map_access_error) + } + + pub(crate) fn free(&self, operation_id: RpcOperationId) -> bool { + self.state + .lock() + .unwrap_or_else(|error| error.into_inner()) + .broker + .free(operation_id.0) + .is_some() + } + + #[cfg(target_os = "wasi")] + pub(crate) fn discard_all(&self) { + self.state + .lock() + .unwrap_or_else(|error| error.into_inner()) + .broker + .discard_all(); + } +} + +fn map_access_error(error: BrokerAccessError) -> RpcAccessError { + match error { + BrokerAccessError::Missing => RpcAccessError::Missing, + BrokerAccessError::NotDrained => RpcAccessError::Pending, + } +} + +fn encode_response( + result: error::Result, + runtime_us: u64, + max_response_len: usize, +) -> Vec { + let mut response = RpcResponse::default(); + match result { + Ok(bytes) => response.response = bytes.into(), + Err(error) => response.error = Some((&error).into()), + } + response.runtime_us = runtime_us; + let encoded = response.encode_to_vec(); + if encoded.len() <= max_response_len { + return encoded; + } + + let error = error::Error::ExecutionError(anyhow::anyhow!( + "RPC response length {} exceeds ABI limit {}", + encoded.len(), + max_response_len + )); + RpcResponse { + error: Some((&error).into()), + ..Default::default() + } + .encode_to_vec() +} + +#[cfg(test)] +mod tests { + use std::any::TypeId; + + use bytes::Bytes; + use easytier_proto::rpc_types::{ + controller::BaseController, + descriptor::{MethodDescriptor, ServiceDescriptor}, + handler::Handler, + }; + + use super::*; + + #[derive(Clone, Copy, Debug)] + enum TestMethod { + Echo, + } + + impl TryFrom for TestMethod { + type Error = (); + + fn try_from(index: u8) -> Result { + match index { + 0 => Ok(Self::Echo), + _ => Err(()), + } + } + } + + impl MethodDescriptor for TestMethod { + fn name(&self) -> &'static str { + "Echo" + } + + fn proto_name(&self) -> &'static str { + "Echo" + } + + fn input_type(&self) -> TypeId { + TypeId::of::>() + } + + fn input_proto_type(&self) -> &'static str { + "bytes" + } + + fn output_type(&self) -> TypeId { + TypeId::of::>() + } + + fn output_proto_type(&self) -> &'static str { + "bytes" + } + + fn index(&self) -> u8 { + 0 + } + } + + #[derive(Clone, Debug, Default)] + struct TestService; + + impl ServiceDescriptor for TestService { + type Method = TestMethod; + + fn name(&self) -> &'static str { + "EchoService" + } + + fn proto_name(&self) -> &'static str { + "test" + } + + fn methods(&self) -> &'static [Self::Method] { + &[TestMethod::Echo] + } + } + + #[derive(Clone)] + struct TestHandler; + + #[async_trait::async_trait] + impl Handler for TestHandler { + type Descriptor = TestService; + type Controller = BaseController; + + async fn call( + &self, + _: Self::Controller, + _: TestMethod, + input: Bytes, + ) -> error::Result { + if input == b"pending"[..] { + std::future::pending().await + } + Ok(input) + } + } + + fn request(full_method_name: &str, payload: &[u8]) -> Vec { + DirectRpcRequest { + full_method_name: full_method_name.to_owned(), + request: payload.to_vec(), + timeout_ms: None, + } + .encode_to_vec() + } + + fn session(max_operations: usize) -> RpcOperationSession { + let registry = Arc::new(ServiceRegistry::new()); + registry.register(TestHandler, ""); + RpcOperationSession::new(registry, max_operations, 1024) + } + + async fn wait_ready(session: &RpcOperationSession, operation: RpcOperationId) -> usize { + for _ in 0..16 { + match session.response_len(operation) { + Ok(length) => return length, + Err(RpcAccessError::Pending) => tokio::task::yield_now().await, + Err(error) => panic!("unexpected RPC access error: {error}"), + } + } + panic!("RPC operation did not complete"); + } + + #[tokio::test] + async fn dispatches_and_consumes_a_protobuf_rpc_response_once() { + let session = session(4); + let operation = session + .submit_encoded(&request("test.Echo", b"hello")) + .unwrap(); + let expected_len = wait_ready(&session, operation).await; + assert!( + session + .take_response_with(operation, |_| None::<()>) + .unwrap() + .is_none() + ); + assert_eq!(session.response_len(operation), Ok(expected_len)); + let mut encoded_response = Vec::new(); + let taken = session + .take_response_with(operation, |response| { + encoded_response.extend_from_slice(response); + Some(response.len()) + }) + .unwrap(); + + assert_eq!(taken, Some(expected_len)); + let response = RpcResponse::decode(encoded_response.as_slice()).unwrap(); + assert_eq!(response.response, b"hello"); + assert!(response.error.is_none()); + assert_eq!( + session.response_len(operation), + Err(RpcAccessError::Missing) + ); + } + + #[tokio::test] + async fn rpc_dispatch_errors_stay_inside_the_response_envelope() { + let session = session(4); + let operation = session + .submit_encoded(&request("test.Missing", b"request")) + .unwrap(); + wait_ready(&session, operation).await; + let mut encoded_response = Vec::new(); + session + .take_response_with(operation, |response| { + encoded_response.extend_from_slice(response); + Some(()) + }) + .unwrap(); + + let response = RpcResponse::decode(encoded_response.as_slice()).unwrap(); + assert!(response.error.is_some()); + } + + #[tokio::test] + async fn freeing_a_pending_operation_releases_capacity_after_cancellation() { + let session = session(1); + let operation = session + .submit_encoded(&request("test.Echo", b"pending")) + .unwrap(); + assert!(session.free(operation)); + tokio::task::yield_now().await; + tokio::task::yield_now().await; + + session + .submit_encoded(&request("test.Echo", b"next")) + .unwrap(); + } + + #[tokio::test] + async fn rejects_a_request_without_a_method() { + let encoded = DirectRpcRequest { + request: b"hello".to_vec(), + ..Default::default() + } + .encode_to_vec(); + + assert!(matches!( + session(1).submit_encoded(&encoded), + Err(RpcSubmitError::MissingMethod) + )); + } +} diff --git a/easytier-core/src/rpc/packet.rs b/easytier-core/src/rpc/packet.rs index 6e06019c..5649e806 100644 --- a/easytier-core/src/rpc/packet.rs +++ b/easytier-core/src/rpc/packet.rs @@ -13,6 +13,14 @@ use super::RpcTransactId; const RPC_PACKET_UDP_PAYLOAD_BUDGET: usize = 1300; +pub(crate) fn accepted_compression_algo() -> CompressionAlgoPb { + if CompressorAlgo::ZstdDefault.is_available() { + CompressionAlgoPb::Zstd + } else { + CompressionAlgoPb::None + } +} + pub async fn compress_packet( accepted_compression_algo: CompressionAlgoPb, content: &[u8], @@ -273,10 +281,12 @@ pub fn build_rpc_packet(args: BuildRpcPacketArgs<'_>) -> Vec { mod tests { use crate::proto::common::CompressionAlgoPb; - use super::compress_packet; + use super::{accepted_compression_algo, compress_packet}; #[tokio::test] async fn compression_negotiation_falls_back_when_zstd_is_unavailable() { + assert_eq!(accepted_compression_algo(), CompressionAlgoPb::None); + let (content, algorithm) = compress_packet(CompressionAlgoPb::Zstd, b"rpc body") .await .unwrap(); diff --git a/easytier-core/src/rpc/server.rs b/easytier-core/src/rpc/server.rs index dc365952..c93f7827 100644 --- a/easytier-core/src/rpc/server.rs +++ b/easytier-core/src/rpc/server.rs @@ -16,14 +16,10 @@ use crate::{ foundation::{ stats::{ArcRpcMetrics, RpcMetricLabels, RpcMetricsProvider}, task::reap_joinset_background, - time::timeout, }, proto::{ - common::{ - self, CompressionAlgoPb, RpcCompressionInfo, RpcPacket, RpcRequest, RpcResponse, - TunnelInfo, - }, - rpc_types::{controller::Controller, error::Result}, + common::{self, RpcCompressionInfo, RpcPacket, RpcRequest, RpcResponse, TunnelInfo}, + rpc_types::error::Result, }, rpc::packet::BuildRpcPacketArgs, tunnel::{ @@ -34,8 +30,12 @@ use crate::{ }; use super::{ - RpcController, Transport, - packet::{PacketMerger, build_rpc_packet, compress_packet, decompress_packet}, + Transport, + dispatch::dispatch_request, + packet::{ + PacketMerger, accepted_compression_algo, build_rpc_packet, compress_packet, + decompress_packet, + }, service_registry::ServiceRegistry, }; @@ -209,6 +209,7 @@ impl Server { async fn handle_rpc_request( packet: RpcPacket, + descriptor: common::RpcDescriptor, reg: Arc, tunnel_info: Option, ) -> Result { @@ -222,21 +223,7 @@ impl Server { packet.body }; let rpc_request = RpcRequest::decode(Bytes::from(body))?; - let timeout_duration = std::time::Duration::from_millis(rpc_request.timeout_ms as u64); - let mut ctrl = RpcController::default(); - let raw_req = Bytes::from(rpc_request.request); - ctrl.set_raw_input(raw_req.clone()); - ctrl.set_tunnel_info(tunnel_info); - let ret = timeout( - timeout_duration, - reg.call_method(packet.descriptor.unwrap(), ctrl.clone(), raw_req), - ) - .await??; - if let Some(raw_output) = ctrl.get_raw_output() { - Ok(raw_output) - } else { - Ok(ret) - } + dispatch_request(reg.as_ref(), descriptor, rpc_request, tunnel_info).await } async fn handle_rpc( @@ -250,7 +237,10 @@ impl Server { let to_peer = packet.to_peer; let transaction_id = packet.transaction_id; let trace_id = packet.trace_id; - let desc = packet.descriptor.clone().unwrap(); + let Some(desc) = packet.descriptor.clone() else { + tracing::warn!("received RPC request without a descriptor"); + return; + }; let method_name = reg.get_method_name(&desc).unwrap_or("".to_owned()); let labels = RpcMetricLabels { network_name: desc.domain_name.clone(), @@ -268,7 +258,7 @@ impl Server { let now = std::time::Instant::now(); let compression_info = packet.compression_info; - let resp_bytes = Self::handle_rpc_request(packet, reg, tunnel_info).await; + let resp_bytes = Self::handle_rpc_request(packet, desc.clone(), reg, tunnel_info).await; match &resp_bytes { Ok(r) => { @@ -307,7 +297,7 @@ impl Server { trace_id, compression_info: RpcCompressionInfo { algo: algo.into(), - accepted_algo: CompressionAlgoPb::Zstd.into(), + accepted_algo: accepted_compression_algo().into(), }, }); for packet in packets { diff --git a/easytier-core/src/rpc/service_registry.rs b/easytier-core/src/rpc/service_registry.rs index 14305282..60b47a83 100644 --- a/easytier-core/src/rpc/service_registry.rs +++ b/easytier-core/src/rpc/service_registry.rs @@ -4,6 +4,8 @@ use dashmap::DashMap; use crate::proto::common::RpcDescriptor; use crate::proto::rpc_types; +#[cfg(all(feature = "management-rpc", any(test, target_os = "wasi")))] +use crate::proto::rpc_types::descriptor::MethodDescriptor; use crate::proto::rpc_types::descriptor::ServiceDescriptor; use crate::proto::rpc_types::handler::{Handler, HandlerExt}; @@ -29,12 +31,35 @@ impl From<&RpcDescriptor> for ServiceKey { #[derive(Clone)] struct ServiceEntry { service: Arc>>, + #[cfg(all(feature = "management-rpc", any(test, target_os = "wasi")))] + methods: Arc<[(String, u8)]>, } impl ServiceEntry { fn new>(h: H) -> Self { + #[cfg(all(feature = "management-rpc", any(test, target_os = "wasi")))] + let descriptor = h.service_descriptor(); + #[cfg(all(feature = "management-rpc", any(test, target_os = "wasi")))] + let service_name = match descriptor.package() { + "" => descriptor.proto_name().to_owned(), + package => format!("{package}.{}", descriptor.proto_name()), + }; + #[cfg(all(feature = "management-rpc", any(test, target_os = "wasi")))] + let methods = descriptor + .methods() + .iter() + .map(|method| { + ( + format!("{service_name}.{}", method.proto_name()), + method.index(), + ) + }) + .collect::>() + .into(); Self { service: Arc::new(Box::new(h)), + #[cfg(all(feature = "management-rpc", any(test, target_os = "wasi")))] + methods, } } @@ -92,6 +117,30 @@ impl ServiceRegistry { Some(method_name) } + #[cfg(all(feature = "management-rpc", any(test, target_os = "wasi")))] + pub(crate) fn resolve_method( + &self, + domain_name: &str, + full_method_name: &str, + ) -> Option { + self.table.iter().find_map(|entry| { + if entry.key().domain_name != domain_name { + return None; + } + let method_index = entry + .value() + .methods + .iter() + .find_map(|(name, index)| (name == full_method_name).then_some(*index))?; + Some(RpcDescriptor { + domain_name: domain_name.to_owned(), + proto_name: entry.key().proto_name.clone(), + service_name: entry.key().service_name.clone(), + method_index: method_index.into(), + }) + }) + } + pub fn unregister>( &self, h: H, diff --git a/easytier-core/src/wasi/abi.rs b/easytier-core/src/wasi/abi.rs index fe7ea972..1f6a3557 100644 --- a/easytier-core/src/wasi/abi.rs +++ b/easytier-core/src/wasi/abi.rs @@ -29,9 +29,21 @@ pub const HOST_CRYPTO_AUTH_FAILED: i32 = -10; /// Version of the JSON document accepted by `easytier_instance_create`. pub const CORE_INSTANCE_CONFIG_VERSION: u32 = 14; +/// Version of the JSON document accepted by `easytier_web_client_create`. +#[cfg(feature = "management")] +pub const WEB_CLIENT_CONFIG_VERSION: u32 = 1; + /// Version of the public data-plane guest export contract. pub const DATA_PLANE_ABI_VERSION: u32 = 3; +/// Version of the protobuf RPC guest export contract. +#[cfg(feature = "management-rpc")] +pub const RPC_ABI_VERSION: u32 = 2; + +/// `easytier_rpc_response_take` has not completed yet. +#[cfg(feature = "management-rpc")] +pub const RPC_STATUS_PENDING: i32 = -6; + /// The guest exposes an instance-scoped data-plane operation broker. pub const DATA_PLANE_CAPABILITY: u64 = 1 << 0; /// The guest data plane supports TCP streams and listeners. @@ -68,6 +80,38 @@ pub const GUEST_EXPORTS: &[&str] = &[ "easytier_instance_error_copy", ]; +/// Guest exports present when process-level WebClient management is enabled. +#[cfg(feature = "management")] +pub const WEB_CLIENT_GUEST_EXPORTS: &[&str] = &[ + "easytier_web_client_create", + "easytier_web_client_drive", + "easytier_web_client_notify_completions", + "easytier_web_client_next_deadline_millis", + "easytier_web_client_is_connected", + "easytier_web_client_drop", +]; + +/// Guest exports present when protobuf management RPC is enabled. +/// +/// `easytier_rpc_request_submit` accepts a serialized +/// `common.DirectRpcRequest`. Its `full_method_name` uses the canonical +/// protobuf reflection name, and `request` contains the serialized protobuf +/// input. An absent `timeout_ms` means the dispatcher has no timeout. +/// `easytier_rpc_response_take` returns a serialized `common.RpcResponse`, +/// including any method error. +/// Submit writes an opaque big-endian `u64` operation ID. A response take +/// returns [`RPC_STATUS_PENDING`] while running; `(output, capacity) == (0, +/// 0)` probes the size, and an undersized buffer returns the required size +/// without consuming the response. Free cancels pending work and discards +/// any retained result. +#[cfg(feature = "management-rpc")] +pub const RPC_GUEST_EXPORTS: &[&str] = &[ + "easytier_rpc_abi_version", + "easytier_rpc_request_submit", + "easytier_rpc_response_take", + "easytier_rpc_operation_free", +]; + /// Guest exports present when the core is built with the smoltcp data plane. #[cfg(feature = "proxy-smoltcp-stack")] pub const DATA_PLANE_GUEST_EXPORTS: &[&str] = &[ diff --git a/easytier-core/src/wasi/adapter/management.rs b/easytier-core/src/wasi/adapter/management.rs new file mode 100644 index 00000000..a031659e --- /dev/null +++ b/easytier-core/src/wasi/adapter/management.rs @@ -0,0 +1,66 @@ +use std::{io, task::Poll}; + +use crate::{ + host::{management::HostManagementIo, socket::HostOperationId}, + wasi::{ + imports::{HOST_PENDING, cancel_operation, start_management_call, take_management_call}, + wire::common::{host_error, status}, + }, +}; + +const MAX_MANAGEMENT_RESULT_LEN: usize = 16 * 1024 * 1024; + +#[derive(Clone, Default)] +pub struct WasiHostManagementIo; + +impl HostManagementIo for WasiHostManagementIo { + fn submit_call(&self, operation: HostOperationId, request: &[u8]) -> io::Result<()> { + let length = u32::try_from(request.len()).map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidInput, + "management request is too long", + ) + })?; + status("start_management_call", unsafe { + start_management_call(operation.0, request.as_ptr() as u32, length) + }) + } + + fn take_call(&self, operation: HostOperationId) -> Poll>> { + let required = unsafe { take_management_call(operation.0, 0, 0) }; + if required == HOST_PENDING { + return Poll::Pending; + } + if required <= 0 { + return Poll::Ready(Err(host_error("take_management_call", required))); + } + let required = usize::try_from(required).expect("positive i32 fits usize"); + if required > MAX_MANAGEMENT_RESULT_LEN { + let _ = unsafe { cancel_operation(operation.0) }; + return Poll::Ready(Err(io::Error::new( + io::ErrorKind::InvalidData, + "host management response is too long", + ))); + } + let mut response = vec![0; required]; + let copied = unsafe { + take_management_call( + operation.0, + response.as_mut_ptr() as u32, + u32::try_from(required).expect("management result limit fits u32"), + ) + }; + if copied != i32::try_from(required).expect("management result length fits i32") { + let _ = unsafe { cancel_operation(operation.0) }; + return Poll::Ready(Err(io::Error::new( + io::ErrorKind::InvalidData, + "host management response length changed", + ))); + } + Poll::Ready(Ok(response)) + } + + fn cancel_operation(&self, operation: HostOperationId) -> io::Result<()> { + status("cancel_operation", unsafe { cancel_operation(operation.0) }) + } +} diff --git a/easytier-core/src/wasi/adapter/mod.rs b/easytier-core/src/wasi/adapter/mod.rs index 63e2bfdf..1fd692c9 100644 --- a/easytier-core/src/wasi/adapter/mod.rs +++ b/easytier-core/src/wasi/adapter/mod.rs @@ -3,5 +3,7 @@ pub mod dns; pub mod environment; pub mod event; +#[cfg(feature = "management")] +pub mod management; pub mod packet; pub mod socket; diff --git a/easytier-core/src/wasi/imports.rs b/easytier-core/src/wasi/imports.rs index c4ce7a00..11aad863 100644 --- a/easytier-core/src/wasi/imports.rs +++ b/easytier-core/src/wasi/imports.rs @@ -168,6 +168,14 @@ unsafe extern "C" { /// Copies the resolved local socket address into the fixed-size `result` buffer. pub(crate) fn take_local_addr_for_remote(operation: u64, result: u32, result_len: u32) -> i32; + /// Starts one process-level WebClient management call after copying its request. + #[cfg(feature = "management")] + pub(crate) fn start_management_call(operation: u64, request: u32, request_len: u32) -> i32; + + /// Probes or copies the process-level management response for `operation`. + #[cfg(feature = "management")] + pub(crate) fn take_management_call(operation: u64, result: u32, result_capacity: u32) -> i32; + /// Attempts to deliver one raw IP packet to a host packet sink. /// /// On success the host owns a complete copy. [`HOST_WOULD_BLOCK`] leaves diff --git a/easytier-core/src/wasi/mod.rs b/easytier-core/src/wasi/mod.rs index 4377237c..0b80dce4 100644 --- a/easytier-core/src/wasi/mod.rs +++ b/easytier-core/src/wasi/mod.rs @@ -20,5 +20,7 @@ pub(crate) mod runtime_driver; pub(crate) mod schema; #[cfg(any(test, target_os = "wasi"))] pub(crate) mod time; +#[cfg(all(target_os = "wasi", feature = "management"))] +pub(crate) mod web_client; #[cfg(any(test, target_os = "wasi"))] pub(crate) mod wire; diff --git a/easytier-core/src/wasi/runtime.rs b/easytier-core/src/wasi/runtime.rs index 9a1badcc..573fde3b 100644 --- a/easytier-core/src/wasi/runtime.rs +++ b/easytier-core/src/wasi/runtime.rs @@ -1,9 +1,41 @@ //! Runtime implementation and lifecycle exports for a WASI core instance. use crate::{ - config::toml::TomlConfig, connectivity::connector_host::HostConnectorEnvironmentSnapshot, + config::toml::TomlConfig, + connectivity::connector_host::HostConnectorEnvironmentSnapshot, + gateway::dhcp::{DhcpIpv4ApplyOutcome, DhcpIpv4Host}, + instance::{CorePacketPlane, InstanceRuntimeHost}, }; +struct WasiInstanceRuntimeHost; + +#[async_trait::async_trait] +impl DhcpIpv4Host for WasiInstanceRuntimeHost { + fn take_interface_closed(&self) -> bool { + false + } + + async fn apply_dhcp_ipv4( + &self, + _previous: Option, + next: Option, + ) -> DhcpIpv4ApplyOutcome { + DhcpIpv4ApplyOutcome::applied(next) + } +} + +#[async_trait::async_trait] +impl InstanceRuntimeHost for WasiInstanceRuntimeHost { + async fn prepare( + &self, + _packet_plane: std::sync::Arc, + ) -> anyhow::Result>> { + Ok(Some(std::sync::Arc::new(WasiInstanceRuntimeHost))) + } + + async fn shutdown(&self) {} +} + pub(super) type WasiCore = crate::instance::CoreInstance< crate::connectivity::connector_host::ConnectorHost< crate::wasi::adapter::socket::backend::WasiHostSocketBackend, @@ -63,6 +95,7 @@ pub(super) fn new_wasi_core_runtime( packet_sink, )); let mut adapters = CoreHostAdapters::new(host, dns, packet_sink, process_runtime); + adapters.instance_runtime = Arc::new(WasiInstanceRuntimeHost); adapters.events = Arc::new(WasiHostEventSink::new(event_sink)); let core = CoreInstance::from_toml(config, adapters)?; @@ -98,9 +131,17 @@ mod abi { #[cfg(feature = "proxy-smoltcp-stack")] mod data_plane; + #[cfg(feature = "management-rpc")] + mod rpc; + #[cfg(feature = "management")] + mod web_client; const MAX_CREATE_CONFIG_LEN: usize = 16 * 1024 * 1024; const MAX_GUEST_BUFFER_LEN: usize = MAX_CREATE_CONFIG_LEN; + #[cfg(feature = "management-rpc")] + const MAX_RPC_MESSAGE_LEN: usize = 16 * 1024 * 1024; + #[cfg(feature = "management-rpc")] + const MAX_RPC_OPERATIONS: usize = 256; const INVALID_HANDLE: i32 = -1; const INVALID_STATE: i32 = -2; const INVALID_INPUT: i32 = -3; @@ -135,6 +176,8 @@ mod abi { struct WasiContext { factory: WasiInstanceFactory, instances: RefCell>>, + #[cfg(feature = "management")] + web_client: RefCell>, abi: RefCell, } @@ -146,6 +189,8 @@ mod abi { Self { factory, instances: RefCell::new(BTreeMap::new()), + #[cfg(feature = "management")] + web_client: RefCell::new(None), abi: RefCell::new(WasiAbiState::default()), } } @@ -171,6 +216,8 @@ mod abi { domain: u64, core: WasiCoreRuntime, execution: Mutex, + #[cfg(feature = "management-rpc")] + rpc_operations: crate::rpc::operation::RpcOperationSession, _protected_tcp_port_leases: Vec, } @@ -223,6 +270,19 @@ mod abi { context.event_sink, )? }; + #[cfg(feature = "management-rpc")] + let rpc_operations = { + let registry = Arc::new(crate::rpc::service_registry::ServiceRegistry::new()); + crate::management::register_bound_management_rpc( + core.core().clone(), + registry.as_ref(), + ); + crate::rpc::operation::RpcOperationSession::new( + registry, + MAX_RPC_OPERATIONS, + MAX_RPC_MESSAGE_LEN, + ) + }; Ok(Arc::new(WasiInstance { instance_id, @@ -235,6 +295,8 @@ mod abi { start_task: None, stop_task: None, }), + #[cfg(feature = "management-rpc")] + rpc_operations, _protected_tcp_port_leases: protected_tcp_ports, })) } @@ -749,6 +811,8 @@ mod abi { return INVALID_STATE; }; let domain = instance.domain; + #[cfg(feature = "management-rpc")] + instance.rpc_operations.discard_all(); { let _domain = enter_domain(domain); drop(instance); diff --git a/easytier-core/src/wasi/runtime/abi/rpc.rs b/easytier-core/src/wasi/runtime/abi/rpc.rs new file mode 100644 index 00000000..e5c83784 --- /dev/null +++ b/easytier-core/src/wasi/runtime/abi/rpc.rs @@ -0,0 +1,214 @@ +//! Protobuf RPC guest exports bound to one WASI core instance. + +use crate::{ + rpc::operation::{RpcAccessError, RpcOperationId, RpcSubmitError}, + wasi::abi::{RPC_ABI_VERSION, RPC_STATUS_PENDING}, +}; + +use super::{ + ASYNC_ERROR, BUSY, INVALID_INPUT, MAX_RPC_MESSAGE_LEN, WasiInstance, read_guest_buffer, + set_instance_error, with_abi_state, with_abi_state_mut, with_instance, +}; + +const OPERATION_ID_LEN: usize = 8; + +impl WasiInstance { + fn submit_rpc(&self, encoded_request: &[u8]) -> Result { + let execution = self.execution.lock().unwrap(); + let _domain = crate::foundation::time::enter_domain(self.domain); + let _runtime = execution.runtime.enter(); + self.rpc_operations.submit_encoded(encoded_request) + } +} + +fn operation_id(raw: u64) -> anyhow::Result { + RpcOperationId::from_raw(raw).ok_or_else(|| anyhow::anyhow!("invalid RPC operation ID")) +} + +fn validate_output(pointer: u32, capacity: usize) -> anyhow::Result<()> { + if pointer == 0 { + anyhow::bail!("guest RPC output buffer pointer is zero"); + } + with_abi_state(|state| { + let buffer = state + .buffers + .get(&pointer) + .ok_or_else(|| anyhow::anyhow!("unknown guest buffer: {pointer}"))?; + if capacity > buffer.len() { + anyhow::bail!( + "guest RPC output capacity {capacity} exceeds allocation {}", + buffer.len() + ); + } + Ok(()) + }) +} + +fn write_output(pointer: u32, bytes: &[u8]) -> anyhow::Result<()> { + with_abi_state_mut(|state| { + let buffer = state + .buffers + .get_mut(&pointer) + .ok_or_else(|| anyhow::anyhow!("unknown guest buffer: {pointer}"))?; + if bytes.len() > buffer.len() { + anyhow::bail!( + "RPC response requires {} bytes, allocation has {}", + bytes.len(), + buffer.len() + ); + } + buffer[..bytes.len()].copy_from_slice(bytes); + Ok(()) + }) +} + +fn submit_status(error: &RpcSubmitError) -> i32 { + match error { + RpcSubmitError::AtCapacity | RpcSubmitError::IdExhausted => BUSY, + RpcSubmitError::ExecutorUnavailable => ASYNC_ERROR, + RpcSubmitError::Decode(_) | RpcSubmitError::MissingMethod => INVALID_INPUT, + } +} + +#[unsafe(no_mangle)] +pub extern "C" fn easytier_rpc_abi_version() -> u32 { + RPC_ABI_VERSION +} + +/// Submits one serialized `common.DirectRpcRequest`. +/// +/// The full protobuf method name is mandatory. On success, writes the opaque +/// operation ID as one big-endian `u64` to `output_operation`. +#[unsafe(no_mangle)] +pub extern "C" fn easytier_rpc_request_submit( + handle: u64, + request_pointer: u32, + request_length: u32, + output_operation: u32, +) -> i32 { + with_instance(handle, |instance| { + if let Err(error) = validate_output(output_operation, OPERATION_ID_LEN) { + set_instance_error(handle, error); + return Ok(INVALID_INPUT); + } + let request = match read_guest_buffer(request_pointer, request_length, MAX_RPC_MESSAGE_LEN) + { + Ok(request) => request, + Err(error) => { + set_instance_error(handle, error); + return Ok(INVALID_INPUT); + } + }; + let operation = match instance.submit_rpc(&request) { + Ok(operation) => operation, + Err(error) => { + let status = submit_status(&error); + set_instance_error(handle, error); + return Ok(status); + } + }; + if let Err(error) = write_output(output_operation, &operation.get().to_be_bytes()) { + instance.rpc_operations.free(operation); + set_instance_error(handle, error); + return Ok(INVALID_INPUT); + } + Ok(0) + }) +} + +/// Probes or consumes one serialized `common.RpcResponse`. +/// +/// Returns [`RPC_STATUS_PENDING`] while the operation is running. Passing +/// `(output, capacity) == (0, 0)` returns the required byte length without +/// consuming the response. If `capacity` is too small, the required length is +/// returned and the response remains available. Otherwise the response is +/// copied, consumed, and its byte length is returned. +#[unsafe(no_mangle)] +pub extern "C" fn easytier_rpc_response_take( + handle: u64, + operation: u64, + output: u32, + capacity: u32, +) -> i32 { + with_instance(handle, |instance| { + let operation = match operation_id(operation) { + Ok(operation) => operation, + Err(error) => { + set_instance_error(handle, error); + return Ok(INVALID_INPUT); + } + }; + let required = match instance.rpc_operations.response_len(operation) { + Ok(required) => required, + Err(RpcAccessError::Pending) => return Ok(RPC_STATUS_PENDING), + Err(error) => { + set_instance_error(handle, error); + return Ok(INVALID_INPUT); + } + }; + let required_i32 = match i32::try_from(required) { + Ok(required) => required, + Err(_) => { + set_instance_error(handle, "RPC response length exceeds i32"); + return Ok(ASYNC_ERROR); + } + }; + if output == 0 && capacity == 0 { + return Ok(required_i32); + } + let capacity = usize::try_from(capacity).expect("u32 fits usize on wasm32"); + if let Err(error) = validate_output(output, capacity) { + set_instance_error(handle, error); + return Ok(INVALID_INPUT); + } + if capacity < required { + return Ok(required_i32); + } + + let mut write_error = None; + let taken = instance + .rpc_operations + .take_response_with(operation, |response| { + if let Err(error) = write_output(output, response) { + write_error = Some(error); + return None; + } + Some(()) + }); + if let Some(error) = write_error { + set_instance_error(handle, error); + return Ok(INVALID_INPUT); + } + match taken { + Ok(Some(())) => Ok(required_i32), + Ok(None) => { + set_instance_error(handle, "RPC response could not be consumed"); + Ok(ASYNC_ERROR) + } + Err(RpcAccessError::Pending) => Ok(RPC_STATUS_PENDING), + Err(error) => { + set_instance_error(handle, error); + Ok(INVALID_INPUT) + } + } + }) +} + +/// Cancels and discards a pending operation, or discards an untaken response. +#[unsafe(no_mangle)] +pub extern "C" fn easytier_rpc_operation_free(handle: u64, operation: u64) -> i32 { + with_instance(handle, |instance| { + let operation = match operation_id(operation) { + Ok(operation) => operation, + Err(error) => { + set_instance_error(handle, error); + return Ok(INVALID_INPUT); + } + }; + if !instance.rpc_operations.free(operation) { + set_instance_error(handle, "unknown RPC operation"); + return Ok(INVALID_INPUT); + } + Ok(0) + }) +} diff --git a/easytier-core/src/wasi/runtime/abi/web_client.rs b/easytier-core/src/wasi/runtime/abi/web_client.rs new file mode 100644 index 00000000..157d9436 --- /dev/null +++ b/easytier-core/src/wasi/runtime/abi/web_client.rs @@ -0,0 +1,121 @@ +use crate::{ + foundation::time::{clear_domain, enter_domain}, + wasi::{ + schema::WasiWebClientCreateConfig, + web_client::{WEB_CLIENT_DOMAIN, WasiWebClientRuntime}, + }, +}; + +use super::{ + ASYNC_ERROR, CONTEXT, INVALID_INPUT, INVALID_STATE, MAX_CREATE_CONFIG_LEN, read_guest_buffer, + set_abi_error, +}; + +fn with_web_client(operation: impl FnOnce(&WasiWebClientRuntime) -> i32) -> i32 { + CONTEXT.with(|context| { + let web_client = context.web_client.borrow(); + match web_client.as_ref() { + Some(web_client) => operation(web_client), + None => { + set_abi_error("WebClient is not running"); + INVALID_STATE + } + } + }) +} + +#[unsafe(no_mangle)] +pub extern "C" fn easytier_web_client_create(config_pointer: u32, config_length: u32) -> i32 { + let encoded = match read_guest_buffer(config_pointer, config_length, MAX_CREATE_CONFIG_LEN) { + Ok(encoded) => encoded, + Err(error) => { + set_abi_error(error); + return INVALID_INPUT; + } + }; + let config: WasiWebClientCreateConfig = match serde_json::from_slice(&encoded) { + Ok(config) => config, + Err(error) => { + set_abi_error(error); + return INVALID_INPUT; + } + }; + if let Err(error) = config.validate() { + set_abi_error(error); + return INVALID_INPUT; + } + let process_runtime = CONTEXT.with(|context| { + if context.web_client.borrow().is_some() { + return None; + } + Some(context.factory.process_runtime.clone()) + }); + let Some(process_runtime) = process_runtime else { + set_abi_error("WebClient is already running"); + return INVALID_STATE; + }; + let web_client = match WasiWebClientRuntime::new(config, process_runtime) { + Ok(web_client) => web_client, + Err(error) => { + set_abi_error(error); + return ASYNC_ERROR; + } + }; + CONTEXT.with(|context| { + *context.web_client.borrow_mut() = Some(web_client); + }); + 0 +} + +#[unsafe(no_mangle)] +pub extern "C" fn easytier_web_client_drive() -> i32 { + with_web_client(|web_client| { + web_client.drive(); + 0 + }) +} + +#[unsafe(no_mangle)] +pub extern "C" fn easytier_web_client_notify_completions() -> i32 { + with_web_client(|web_client| { + web_client.notify_host_completions(); + 0 + }) +} + +#[unsafe(no_mangle)] +pub extern "C" fn easytier_web_client_next_deadline_millis() -> i64 { + let mut result = i64::from(INVALID_STATE); + let status = with_web_client(|web_client| { + result = web_client + .next_wait_millis() + .map(|millis| i64::try_from(millis).unwrap_or(i64::MAX)) + .unwrap_or(i64::MAX); + 0 + }); + if status == 0 { + result + } else { + i64::from(status) + } +} + +#[unsafe(no_mangle)] +pub extern "C" fn easytier_web_client_is_connected() -> i32 { + with_web_client(|web_client| i32::from(web_client.is_connected())) +} + +#[unsafe(no_mangle)] +pub extern "C" fn easytier_web_client_drop() -> i32 { + let web_client = CONTEXT.with(|context| context.web_client.borrow_mut().take()); + let Some(web_client) = web_client else { + set_abi_error("WebClient is not running"); + return INVALID_STATE; + }; + { + let _domain = enter_domain(WEB_CLIENT_DOMAIN); + drop(web_client); + } + clear_domain(WEB_CLIENT_DOMAIN); + 0 +} diff --git a/easytier-core/src/wasi/schema.rs b/easytier-core/src/wasi/schema.rs index b9fa8329..40a7153b 100644 --- a/easytier-core/src/wasi/schema.rs +++ b/easytier-core/src/wasi/schema.rs @@ -8,6 +8,8 @@ use crate::{ pub(crate) const WASI_CORE_INSTANCE_CONFIG_VERSION: u32 = crate::wasi::abi::CORE_INSTANCE_CONFIG_VERSION; +#[cfg(all(target_os = "wasi", feature = "management"))] +pub(crate) const WASI_WEB_CLIENT_CONFIG_VERSION: u32 = crate::wasi::abi::WEB_CLIENT_CONFIG_VERSION; /// Versioned payload accepted by host-driven instance frontends. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -32,3 +34,29 @@ impl WasiCoreInstanceCreateConfig { TomlConfig::new_from_str_with_source("WASI create config", &self.config) } } + +#[cfg(all(target_os = "wasi", feature = "management"))] +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct WasiWebClientCreateConfig { + pub version: u32, + pub endpoint: String, + pub machine_id: String, + pub hostname: String, + pub secure_mode: bool, + pub os_type: String, + pub environment: HostConnectorEnvironmentSnapshot, +} + +#[cfg(all(target_os = "wasi", feature = "management"))] +impl WasiWebClientCreateConfig { + pub fn validate(&self) -> anyhow::Result<()> { + if self.version != WASI_WEB_CLIENT_CONFIG_VERSION { + anyhow::bail!( + "unsupported host WebClient config version: {}", + self.version + ); + } + uuid::Uuid::parse_str(&self.machine_id)?; + Ok(()) + } +} diff --git a/easytier-core/src/wasi/web_client.rs b/easytier-core/src/wasi/web_client.rs new file mode 100644 index 00000000..e061073c --- /dev/null +++ b/easytier-core/src/wasi/web_client.rs @@ -0,0 +1,474 @@ +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use bytes::Bytes; +use prost::Message; +use tokio::runtime::Builder; +use url::Url; + +use crate::{ + config::{api_input::NetworkConfigExt, toml::ConfigLoader}, + connectivity::{ + connector_host::{ConnectorHost, new_connector_host}, + manual::{ + ManualConnectorOptions, ManualTunnelConnector, discovery::ManualEndpointDiscoveryConfig, + }, + protocol::{CoreClientProtocolConfig, CoreClientProtocolUpgrader, raw::TunnelDialer}, + }, + foundation::time::{enter_domain, next_deadline_millis}, + host::{dns::HostDnsResolver, management::HostManagementClient, socket::HostSocketRuntime}, + management::{ + ConfigServerEndpoint, ManagementRpcForwarder, WebClient, WebClientBackend, WebClientConfig, + config_source_from_rpc, register_forwarded_instance_management_rpc, + }, + process_runtime::CoreProcessRuntime, + proto::{ + api::manage::{ + ListNetworkInstanceRequest, NetworkConfig, NetworkingMethod, RunNetworkInstanceRequest, + ValidateConfigRequest, ValidateConfigResponse, WebClientService, + WebClientServiceClient, WebClientServiceDescriptor, WebClientServiceMethodDescriptor, + }, + common::{DirectRpcRequest, HostManagementRequest, RpcResponse}, + rpc_types::{ + controller::BaseController, + descriptor::{MethodDescriptor, ServiceDescriptor}, + error, + handler::Handler, + }, + web::DeviceOsInfo, + }, + rpc::service_registry::ServiceRegistry, + socket::IpVersion, + tunnel::Tunnel, + wasi::{ + adapter::{ + dns::WasiHostDnsIo, environment::WasiHostConnectorEnvironmentIo, + management::WasiHostManagementIo, socket::backend::WasiHostSocketBackend, + }, + runtime_driver::{RuntimeDriveOutcome, RuntimeDriver}, + schema::WasiWebClientCreateConfig, + }, +}; + +pub(super) const WEB_CLIENT_DOMAIN: u64 = u64::MAX; + +type WasiConnectorHost = ConnectorHost; + +fn supports_hosted_tunnel_url(value: &str) -> bool { + Url::parse(value).is_ok_and(|url| matches!(url.scheme(), "tcp" | "udp")) +} + +fn hosted_network_config(config: &NetworkConfig) -> NetworkConfig { + let public_server_url = config + .public_server_url + .as_ref() + .filter(|url| supports_hosted_tunnel_url(url)) + .cloned(); + let peers = config + .peers + .iter() + .filter(|peer| supports_hosted_tunnel_url(&peer.uri)) + .cloned() + .collect::>(); + let networking_method = + match NetworkingMethod::try_from(config.networking_method.unwrap_or_default()) { + Ok(NetworkingMethod::PublicServer) + if public_server_url.is_none() && peers.is_empty() => + { + NetworkingMethod::Standalone + } + Ok(method) => method, + Err(_) => NetworkingMethod::Standalone, + }; + + NetworkConfig { + instance_id: config.instance_id.clone(), + dhcp: config.dhcp, + virtual_ipv4: config.virtual_ipv4.clone(), + network_length: config.network_length, + hostname: config.hostname.clone(), + network_name: config.network_name.clone(), + network_secret: config.network_secret.clone(), + networking_method: Some(networking_method as i32), + public_server_url, + peer_urls: config + .peer_urls + .iter() + .filter(|url| supports_hosted_tunnel_url(url)) + .cloned() + .collect(), + proxy_cidrs: config.proxy_cidrs.clone(), + listener_urls: config + .listener_urls + .iter() + .filter(|url| supports_hosted_tunnel_url(url)) + .cloned() + .collect(), + latency_first: config.latency_first, + disable_ipv6: config.disable_ipv6, + disable_p2p: config.disable_p2p, + no_tun: config.no_tun, + relay_all_peer_rpc: config.relay_all_peer_rpc, + enable_relay_network_whitelist: config.enable_relay_network_whitelist, + relay_network_whitelist: config.relay_network_whitelist.clone(), + disable_encryption: config.disable_encryption, + disable_udp_hole_punching: config.disable_udp_hole_punching, + mtu: config.mtu, + enable_private_mode: config.enable_private_mode, + disable_sym_hole_punching: config.disable_sym_hole_punching, + p2p_only: config.p2p_only, + disable_tcp_hole_punching: config.disable_tcp_hole_punching, + secure_mode: config.secure_mode.clone(), + acl: config.acl.clone(), + port_forwards: config.port_forwards.clone(), + lazy_p2p: config.lazy_p2p, + need_p2p: config.need_p2p, + instance_recv_bps_limit: config.instance_recv_bps_limit, + disable_upnp: config.disable_upnp, + disable_relay_data: config.disable_relay_data, + enable_udp_broadcast_relay: config.enable_udp_broadcast_relay, + peers, + ..Default::default() + } +} + +struct WasiConfigServerConnector { + url: Url, + connector: ManualTunnelConnector, +} + +#[async_trait] +impl TunnelDialer for WasiConfigServerConnector { + async fn connect(&self) -> anyhow::Result> { + self.connector + .connect(self.url.clone(), IpVersion::Both) + .await + } + + fn remote_url(&self) -> Url { + self.url.clone() + } +} + +#[derive(Clone)] +struct HostManagementHandler { + client: HostManagementClient, +} + +impl HostManagementHandler { + async fn forward( + &self, + full_method_name: String, + request: Bytes, + prepared_config: Option, + prepared_instance_id: Option, + ) -> error::Result { + let request = HostManagementRequest { + rpc: Some(DirectRpcRequest { + full_method_name, + request: request.into(), + timeout_ms: None, + }), + prepared_config, + prepared_instance_id: prepared_instance_id.map(Into::into), + }; + let response = self + .client + .call(&request.encode_to_vec()) + .await + .map_err(|error| error::Error::ExecutionError(error.into()))?; + let response = RpcResponse::decode(response.as_slice())?; + if let Some(error) = response.error { + return Err((&error).into()); + } + Ok(response.response.into()) + } +} + +#[async_trait] +impl ManagementRpcForwarder for HostManagementHandler { + async fn forward(&self, full_method_name: String, input: Bytes) -> error::Result { + HostManagementHandler::forward(self, full_method_name, input, None, None).await + } +} + +#[async_trait] +impl Handler for HostManagementHandler { + type Descriptor = WebClientServiceDescriptor; + type Controller = BaseController; + + async fn call( + &self, + _: Self::Controller, + method: WebClientServiceMethodDescriptor, + input: Bytes, + ) -> error::Result { + let full_method_name = format!( + "{}.{}.{}", + WebClientServiceDescriptor.package(), + WebClientServiceDescriptor.proto_name(), + method.proto_name() + ); + match method { + WebClientServiceMethodDescriptor::ValidateConfig => { + let request = ValidateConfigRequest::decode(input)?; + let network_config = request.config.unwrap_or_default(); + let config = hosted_network_config(&network_config).gen_config()?; + Ok(ValidateConfigResponse { + toml_config: config.dump(), + } + .encode_to_vec() + .into()) + } + WebClientServiceMethodDescriptor::RunNetworkInstance => { + let request = RunNetworkInstanceRequest::decode(input.clone())?; + let network_config = request + .config + .as_ref() + .ok_or_else(|| anyhow::anyhow!("config is required"))?; + let config = hosted_network_config(&network_config).gen_config()?; + let instance_id = request + .inst_id + .map(Into::into) + .unwrap_or_else(|| config.get_id()); + config.set_id(instance_id); + config.set_network_config_source(config_source_from_rpc(request.source)); + self.forward( + full_method_name, + input, + Some(config.dump()), + Some(instance_id), + ) + .await + } + _ => self.forward(full_method_name, input, None, None).await, + } + } +} + +struct WasiWebClientBackend { + handler: HostManagementHandler, +} + +#[async_trait] +impl WebClientBackend for WasiWebClientBackend { + fn register(&self, registry: &ServiceRegistry) { + registry.register(self.handler.clone(), ""); + register_forwarded_instance_management_rpc(self.handler.clone(), registry); + } + + async fn instance_ids(&self) -> anyhow::Result> { + let response = WebClientServiceClient::new(self.handler.clone()) + .list_network_instance(BaseController::default(), ListNetworkInstanceRequest {}) + .await?; + Ok(response.inst_ids.into_iter().map(Into::into).collect()) + } +} + +pub(super) struct WasiWebClientRuntime { + socket_runtime: HostSocketRuntime, + client: WebClient<()>, + execution: Mutex, +} + +struct WasiWebClientExecution { + runtime: tokio::runtime::Runtime, + runtime_driver: RuntimeDriver, + drive_again: bool, +} + +impl WasiWebClientRuntime { + pub(super) fn new( + config: WasiWebClientCreateConfig, + process_runtime: Arc, + ) -> anyhow::Result { + let endpoint = ConfigServerEndpoint::parse(&config.endpoint, |url| { + matches!(url.scheme(), "tcp" | "udp") + })?; + let machine_id = uuid::Uuid::parse_str(&config.machine_id)?; + let runtime_driver = RuntimeDriver::default(); + let park_driver = runtime_driver.clone(); + let runtime = Builder::new_current_thread() + .enable_time() + .on_thread_park(move || park_driver.on_thread_park()) + .build()?; + let socket_runtime = HostSocketRuntime::new(); + let client = { + let _domain = enter_domain(WEB_CLIENT_DOMAIN); + let _runtime = runtime.enter(); + let host = Arc::new(new_connector_host( + socket_runtime.clone(), + Arc::new(WasiHostSocketBackend::default()), + config.environment, + Arc::new(WasiHostConnectorEnvironmentIo), + )); + let dns = Arc::new(HostDnsResolver::new( + socket_runtime.clone(), + Arc::new(WasiHostDnsIo), + )); + let connector = process_runtime.manual_connector( + host, + dns.clone(), + dns, + Arc::new(CoreClientProtocolUpgrader::new( + CoreClientProtocolConfig::default(), + )), + ManualEndpointDiscoveryConfig::default(), + ManualConnectorOptions::default(), + ); + let backend = Arc::new(WasiWebClientBackend { + handler: HostManagementHandler { + client: HostManagementClient::new( + socket_runtime.clone(), + Arc::new(WasiHostManagementIo), + ), + }, + }); + WebClient::with_backend( + WasiConfigServerConnector { + url: endpoint.connect_url().clone(), + connector, + }, + WebClientConfig { + token: endpoint.token().to_owned(), + machine_id, + hostname: config.hostname, + device_os: DeviceOsInfo { + os_type: config.os_type, + version: String::new(), + distribution: String::new(), + }, + easytier_version: env!("CARGO_PKG_VERSION").to_owned(), + secure_mode: config.secure_mode, + }, + backend, + ) + }; + + Ok(Self { + socket_runtime, + client, + execution: Mutex::new(WasiWebClientExecution { + runtime, + runtime_driver, + drive_again: false, + }), + }) + } + + pub(super) fn drive(&self) { + let _domain = enter_domain(WEB_CLIENT_DOMAIN); + let advance_timers = next_deadline_millis(WEB_CLIENT_DOMAIN) == Some(0); + let mut execution = self.execution.lock().unwrap(); + execution.drive_again = execution + .runtime_driver + .drive(&execution.runtime, advance_timers) + == RuntimeDriveOutcome::BudgetExhausted; + } + + pub(super) fn notify_host_completions(&self) { + self.socket_runtime.notify_completions(); + } + + pub(super) fn next_wait_millis(&self) -> Option { + let execution = self.execution.lock().unwrap(); + if execution.drive_again { + Some(0) + } else { + next_deadline_millis(WEB_CLIENT_DOMAIN) + } + } + + pub(super) fn is_connected(&self) -> bool { + self.client.is_connected() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::proto::{ + api::manage::NetworkPeerConfig, + common::{CompressionAlgoPb, SecureModeConfig}, + }; + + #[test] + fn hosted_config_keeps_supported_fields_and_filters_tunnel_urls() { + let original = NetworkConfig { + instance_id: Some(uuid::Uuid::new_v4().to_string()), + dhcp: Some(true), + network_name: Some("network".to_owned()), + network_secret: Some("secret".to_owned()), + networking_method: Some(NetworkingMethod::Manual as i32), + peer_urls: vec![ + "tcp://peer.example:11010".to_owned(), + "wg://peer.example:11011".to_owned(), + ], + listener_urls: vec![ + "udp://0.0.0.0:11010".to_owned(), + "wg://0.0.0.0:11011".to_owned(), + ], + peers: vec![ + NetworkPeerConfig { + uri: "tcp://peer.example:11010".to_owned(), + peer_public_key: Some("key".to_owned()), + }, + NetworkPeerConfig { + uri: "quic://peer.example:11010".to_owned(), + peer_public_key: None, + }, + ], + secure_mode: Some(SecureModeConfig { + enabled: true, + ..Default::default() + }), + enable_private_mode: Some(true), + disable_relay_data: Some(true), + proxy_cidrs: vec!["10.88.0.0/24".to_owned()], + port_forwards: vec![crate::proto::api::manage::PortForwardConfig { + proto: "tcp".to_owned(), + bind_ip: "127.0.0.1".to_owned(), + bind_port: 18080, + dst_ip: "10.88.0.2".to_owned(), + dst_port: 80, + }], + enable_vpn_portal: Some(true), + data_compress_algo: Some(CompressionAlgoPb::Zstd as i32), + credential_file: Some("/unsupported".to_owned()), + ..Default::default() + }; + + let hosted = hosted_network_config(&original); + + assert_eq!( + hosted.peer_urls, + vec!["tcp://peer.example:11010".to_owned()] + ); + assert_eq!(hosted.listener_urls, vec!["udp://0.0.0.0:11010".to_owned()]); + assert_eq!(hosted.peers, original.peers[..1]); + assert_eq!(hosted.secure_mode, original.secure_mode); + assert_eq!(hosted.enable_private_mode, Some(true)); + assert_eq!(hosted.disable_relay_data, Some(true)); + assert_eq!(hosted.proxy_cidrs, original.proxy_cidrs); + assert_eq!(hosted.port_forwards, original.port_forwards); + assert_eq!(hosted.enable_vpn_portal, None); + assert_eq!(hosted.data_compress_algo, None); + assert_eq!(hosted.credential_file, None); + assert_eq!(original.listener_urls.len(), 2); + } + + #[test] + fn hosted_config_falls_back_to_standalone_without_a_supported_public_peer() { + let hosted = hosted_network_config(&NetworkConfig { + networking_method: Some(NetworkingMethod::PublicServer as i32), + public_server_url: Some("wg://peer.example:11010".to_owned()), + ..Default::default() + }); + + assert_eq!( + hosted.networking_method, + Some(NetworkingMethod::Standalone as i32) + ); + assert_eq!(hosted.public_server_url, None); + } +} diff --git a/easytier-proto/proto/api_config.proto b/easytier-proto/proto/api_config.proto index ccfa30cc..dd200af0 100644 --- a/easytier-proto/proto/api_config.proto +++ b/easytier-proto/proto/api_config.proto @@ -80,6 +80,7 @@ message GetConfigRequest { message GetConfigResponse { api.manage.NetworkConfig config = 1; + string toml_config = 2; } service ConfigRpc { diff --git a/easytier-proto/proto/common.proto b/easytier-proto/proto/common.proto index 632cab79..53b114a3 100644 --- a/easytier-proto/proto/common.proto +++ b/easytier-proto/proto/common.proto @@ -102,6 +102,24 @@ message RpcRequest { int32 timeout_ms = 3; } +// One transport-neutral RPC invocation submitted through a direct ABI. +// +// full_method_name uses the protobuf reflection form: +// ".." or "." for an empty package. +message DirectRpcRequest { + string full_method_name = 1; + bytes request = 2; + optional uint64 timeout_ms = 3; +} + +// One process-level management call delegated by the WASI WebClient to its host. +message HostManagementRequest { + DirectRpcRequest rpc = 1; + optional string prepared_config = 2; + // Runtime identity paired with prepared_config without rewriting rpc.request. + UUID prepared_instance_id = 3; +} + message RpcResponse { bytes response = 1; error.Error error = 2; diff --git a/script/build-wasi-core.sh b/script/build-wasi-core.sh index cb6c37cc..c1b01a4e 100755 --- a/script/build-wasi-core.sh +++ b/script/build-wasi-core.sh @@ -10,7 +10,7 @@ cd "$repository_root" readonly target_dir="${CARGO_TARGET_DIR:-target}" readonly raw_artifact="${target_dir}/wasm32-wasip1/release/easytier_core.wasm" readonly artifact="${target_dir}/wasm32-wasip1/release/easytier_core_go_host.wasm" -readonly core_features="proxy-smoltcp-stack,ring-crypto,wasi-crypto-offload" +readonly core_features="management-rpc,proxy-smoltcp-stack,ring-crypto,wasi-crypto-offload" sha256_file() { if command -v sha256sum >/dev/null; then