From 1e40350c896e2635fbb7cc2804819cd25001b028 Mon Sep 17 00:00:00 2001 From: KKRainbow <443152178@qq.com> Date: Fri, 7 Aug 2026 18:34:53 +0800 Subject: [PATCH] feat(wasi): expose protobuf RPC request ABI (#2477) * feat(wasi): expose protobuf RPC request ABI Add an instance-scoped asynchronous RPC session backed by the shared operation broker. Reuse the existing dispatcher and management handlers. WASI hosts can call PeerManageRpc and ConnectorManageRpc with the same protobuf payloads as easytier-cli. Export ABI version, submit, take, and free functions. Bind selectors to the WASM instance handle and keep method errors in RpcResponse. Enable management RPC explicitly in the Go-host WASM build. * fix(gateway): serialize UDP client eviction Serialize UDP client admission across forwarding rules so only one eviction can claim and wait for a released semaphore permit. Retry when cleanup concurrently removes the selected client. Add a multithreaded regression test for the permit handoff while the evicted client is still referenced. * fix(gateway): publish UDP client admission atomically Hold the admission guard through client and response-task publication so a concurrent eviction cannot leave an orphan task holding the slot permit. Open the data-plane flow before entering the critical section and extend the multithreaded regression test across the publication window. --- easytier-core/Cargo.toml | 4 +- easytier-core/src/foundation/mod.rs | 6 +- easytier-core/src/gateway/dataplane/stack.rs | 3 +- easytier-core/src/gateway/dataplane/tests.rs | 24 + easytier-core/src/gateway/port_forward.rs | 280 ++++++++++- easytier-core/src/host/management.rs | 41 ++ easytier-core/src/host/mod.rs | 2 + easytier-core/src/management/forwarded_rpc.rs | 152 ++++++ easytier-core/src/management/full/mod.rs | 2 + .../src/management/full/web_client.rs | 267 +++++----- .../src/management/instance_rpc/full.rs | 10 +- .../src/management/instance_rpc/mod.rs | 205 +++++++- easytier-core/src/management/mod.rs | 24 + easytier-core/src/rpc/client.rs | 10 +- easytier-core/src/rpc/dispatch.rs | 50 ++ easytier-core/src/rpc/mod.rs | 3 + easytier-core/src/rpc/operation.rs | 430 ++++++++++++++++ easytier-core/src/rpc/packet.rs | 12 +- easytier-core/src/rpc/server.rs | 42 +- easytier-core/src/rpc/service_registry.rs | 49 ++ easytier-core/src/wasi/abi.rs | 44 ++ easytier-core/src/wasi/adapter/management.rs | 66 +++ easytier-core/src/wasi/adapter/mod.rs | 2 + easytier-core/src/wasi/imports.rs | 8 + easytier-core/src/wasi/mod.rs | 2 + easytier-core/src/wasi/runtime.rs | 66 ++- easytier-core/src/wasi/runtime/abi/rpc.rs | 214 ++++++++ .../src/wasi/runtime/abi/web_client.rs | 121 +++++ easytier-core/src/wasi/schema.rs | 28 ++ easytier-core/src/wasi/web_client.rs | 474 ++++++++++++++++++ easytier-proto/proto/api_config.proto | 1 + easytier-proto/proto/common.proto | 18 + script/build-wasi-core.sh | 2 +- 33 files changed, 2479 insertions(+), 183 deletions(-) create mode 100644 easytier-core/src/host/management.rs create mode 100644 easytier-core/src/management/forwarded_rpc.rs create mode 100644 easytier-core/src/rpc/dispatch.rs create mode 100644 easytier-core/src/rpc/operation.rs create mode 100644 easytier-core/src/wasi/adapter/management.rs create mode 100644 easytier-core/src/wasi/runtime/abi/rpc.rs create mode 100644 easytier-core/src/wasi/runtime/abi/web_client.rs create mode 100644 easytier-core/src/wasi/web_client.rs 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