From 5dbc71e58ecb66c2f9174519a6f54e23a8f22260 Mon Sep 17 00:00:00 2001 From: FrankHan Date: Sun, 2 Aug 2026 10:11:50 +0800 Subject: [PATCH] feat(ohos): expose nearby console through core rpc --- easytier-contrib/easytier-ohrs/Cargo.lock | 5 + easytier-contrib/easytier-ohrs/Cargo.toml | 9 + easytier-contrib/easytier-ohrs/src/lib.rs | 68 +- .../easytier-ohrs/src/nearby_management.rs | 962 ++++++++++++++++++ 4 files changed, 1040 insertions(+), 4 deletions(-) create mode 100644 easytier-contrib/easytier-ohrs/src/nearby_management.rs diff --git a/easytier-contrib/easytier-ohrs/Cargo.lock b/easytier-contrib/easytier-ohrs/Cargo.lock index 70ed9afe..d4713e9f 100644 --- a/easytier-contrib/easytier-ohrs/Cargo.lock +++ b/easytier-contrib/easytier-ohrs/Cargo.lock @@ -1288,10 +1288,15 @@ dependencies = [ name = "easytier-ohrs" version = "0.1.0" dependencies = [ + "anyhow", "async-trait", "base64 0.22.1", + "bytes", "easytier", + "easytier-core", + "easytier-proto", "flate2", + "futures", "gethostname 1.1.0", "ipnet", "napi-build-ohos", diff --git a/easytier-contrib/easytier-ohrs/Cargo.toml b/easytier-contrib/easytier-ohrs/Cargo.toml index 6400d2cc..9b6d2d5e 100644 --- a/easytier-contrib/easytier-ohrs/Cargo.toml +++ b/easytier-contrib/easytier-ohrs/Cargo.toml @@ -7,9 +7,18 @@ edition = "2024" crate-type=["cdylib"] [dependencies] +anyhow = "1.0" async-trait = "0.1" base64 = "0.22" +bytes = "1.5" +easytier-core = { path = "../../easytier-core", default-features = false } +easytier-proto = { path = "../../easytier-proto", default-features = false, features = [ + "api", + "core", + "json-rpc", +] } flate2 = "1.1" +futures = "0.3" gethostname = "1.1" easytier = { path = "../../easytier" } napi-derive-ohos = "1.1" diff --git a/easytier-contrib/easytier-ohrs/src/lib.rs b/easytier-contrib/easytier-ohrs/src/lib.rs index 0b224d05..00f0579c 100644 --- a/easytier-contrib/easytier-ohrs/src/lib.rs +++ b/easytier-contrib/easytier-ohrs/src/lib.rs @@ -37,6 +37,7 @@ macro_rules! ohrs_log_debug { mod config; mod exports; mod kernel_bridge; +mod nearby_management; mod platform; mod runtime; @@ -57,7 +58,7 @@ use easytier::common::config::NetworkConfigExt; use easytier::common::constants::EASYTIER_VERSION; use easytier::common::{ MachineIdOptions, - config::{ConfigFileControl, ConfigLoader, TomlConfigLoader}, + config::{ConfigLoader, TomlConfigLoader}, }; use easytier::instance::factory::{NativeInstanceManager, native_instance_manager_with_runtime}; use easytier::proto::api::manage::NetworkConfig; @@ -68,6 +69,7 @@ use kernel_bridge::{ stop_local_socket_server as stop_local_socket_server_inner, }; use napi_derive_ohos::napi; +use napi_ohos::bindgen_prelude::Uint8Array; use runtime::state::runtime_state::{RuntimeAggregateState, RuntimeInstanceState}; use std::collections::{HashMap, HashSet}; use std::format; @@ -380,6 +382,7 @@ fn stop_runtime_inner() -> bool { && ok; } maybe_stop_local_socket_server(); + let _ = nearby_management::stop_runtime_management_server(); ok } @@ -730,13 +733,24 @@ pub(crate) fn run_network_instance_from_json(cfg_json: &str) -> bool { return false; } - match INSTANCE_MANAGER.run_network_instance(cfg, ConfigFileControl::STATIC_CONFIG) { + let Some(config_control) = nearby_management::runtime_management_config_control(inst_id) else { + ohrs_log_error!("[Rust] nearby management config store is not initialized"); + return false; + }; + if !nearby_management::ensure_runtime_management_server_started() { + return false; + } + + match INSTANCE_MANAGER.run_network_instance(cfg, config_control) { Ok(_) => { cache_runtime_config_snapshot(inst_id.to_string(), inst_id.to_string(), config); true } Err(err) => { ohrs_log_error!("[Rust] start_kernel failed for {}: {}", inst_id, err); + if INSTANCE_MANAGER.instance_ids().is_empty() { + let _ = nearby_management::stop_runtime_management_server(); + } false } } @@ -849,12 +863,16 @@ pub fn start_kernel(config_id: String) -> bool { #[napi] pub fn stop_kernel(config_id: String) -> bool { - exports::runtime_api::stop_kernel( + let stopped = exports::runtime_api::stop_kernel( config_id, stop_web_client, parse_instance_uuid, maybe_stop_local_socket_server, - ) + ); + if stopped && INSTANCE_MANAGER.instance_ids().is_empty() { + let _ = nearby_management::stop_runtime_management_server(); + } + stopped } #[napi] @@ -959,6 +977,48 @@ pub fn run_network_instance(cfg_json: String) -> bool { run_network_instance_from_json(&cfg_json) } +/// Opens one Core RPC endpoint for a HarmonyOS collaboration session. +/// +/// The Harmony layer transports the returned native packets verbatim with +/// `abilityConnectionManager.sendData`; all RPC framing stays inside Core. +#[napi] +pub fn open_nearby_management_session(session_key: String, host: bool) -> bool { + nearby_management::open_nearby_management_session(session_key, host) +} + +#[napi] +pub fn close_nearby_management_session(session_key: String) -> bool { + nearby_management::close_nearby_management_session(session_key) +} + +#[napi] +pub fn push_nearby_management_packet(session_key: String, packet: Uint8Array) -> bool { + nearby_management::push_nearby_management_packet(session_key, packet) +} + +#[napi] +pub fn drain_nearby_management_packets(session_key: String) -> Vec { + nearby_management::drain_nearby_management_packets(session_key) +} + +#[napi] +pub async fn call_nearby_management_json_rpc( + session_key: String, + service_name: String, + method_name: String, + domain_name: Option, + payload_json: String, +) -> String { + nearby_management::call_nearby_management_json_rpc( + session_key, + service_name, + method_name, + domain_name, + payload_json, + ) + .await +} + #[napi] pub fn collect_network_infos() -> Vec { exports::runtime_api::collect_network_infos() diff --git a/easytier-contrib/easytier-ohrs/src/nearby_management.rs b/easytier-contrib/easytier-ohrs/src/nearby_management.rs new file mode 100644 index 00000000..2f8e7e79 --- /dev/null +++ b/easytier-contrib/easytier-ohrs/src/nearby_management.rs @@ -0,0 +1,962 @@ +//! HarmonyOS nearby transport for the canonical EasyTier management RPC surface. +//! +//! Cross-device payloads are native EasyTier `ZCPacket`s. The host-facing side +//! connects to a Unix-domain standalone RPC server owned by the process that +//! actually runs the EasyTier instance, so a VPN Extension never accidentally +//! exposes the empty `InstanceManager` loaded by `EntryAbility`. + +use std::{ + collections::HashMap, + fmt, io, + net::{Ipv4Addr, SocketAddr, SocketAddrV4}, + path::{Path, PathBuf}, + pin::Pin, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, + task::{Context, Poll}, +}; + +use async_trait::async_trait; +use bytes::BytesMut; +use easytier::{ + common::config::{ConfigLoader as _, NetworkConfigExt as _, TomlConfigLoader}, + instance::factory::NativeInstanceFactory, + rpc_service::logger::NativeLoggerControl, +}; +use easytier_core::{ + config::{api::network_config_from_toml, toml::TomlConfig}, + management::{ + ConfigFileControl, ConfigFilePermission, ConfigFileStorage, InstanceMutationHooks, + LoggerManagementRpc, ProcessManagementRpc, register_instance_management_rpc, + }, + packet::{PacketType, ZCPacket, ZCPacketType}, + rpc::{bidirect::BidirectRpcManager, client::Client, standalone::StandAloneServer}, + socket::{SocketListener, tcp::VirtualTcpSocket}, + tunnel::{SplitTunnel, Tunnel, tcp::TcpTunnelUpgrader}, +}; +use easytier_proto::{ + api::{ + config::{ConfigRpc, ConfigRpcClientFactory}, + instance::{ + AclManageRpc, AclManageRpcClientFactory, ConnectorManageRpc, + ConnectorManageRpcClientFactory, CredentialManageRpc, CredentialManageRpcClientFactory, + MappedListenerManageRpc, MappedListenerManageRpcClientFactory, PeerManageRpc, + PeerManageRpcClientFactory, PortForwardManageRpc, PortForwardManageRpcClientFactory, + StatsRpc, StatsRpcClientFactory, TcpProxyRpc, TcpProxyRpcClientFactory, VpnPortalRpc, + VpnPortalRpcClientFactory, + }, + logger::{LoggerRpc, LoggerRpcClientFactory, LoggerRpcServer}, + manage::{ + CollectNetworkInfoRequest, CollectNetworkInfoResponse, DeleteNetworkInstanceRequest, + DeleteNetworkInstanceResponse, GetNetworkInstanceConfigRequest, + GetNetworkInstanceConfigResponse, ListNetworkInstanceMetaRequest, + ListNetworkInstanceMetaResponse, ListNetworkInstanceRequest, + ListNetworkInstanceResponse, RetainNetworkInstanceRequest, + RetainNetworkInstanceResponse, RunNetworkInstanceRequest, RunNetworkInstanceResponse, + ValidateConfigRequest, ValidateConfigResponse, WebClientService, + WebClientServiceClientFactory, WebClientServiceServer, + }, + }, + common::TunnelInfo, + peer_rpc::{PeerCenterRpc, PeerCenterRpcClientFactory}, + rpc_types::{controller::BaseController, error::Error as RpcError}, +}; +use futures::{SinkExt, StreamExt}; +use napi_ohos::bindgen_prelude::Uint8Array; +use serde_json::{Value, json}; +use tokio::{ + io::{AsyncRead, AsyncWrite, ReadBuf}, + net::{UnixListener, UnixStream}, + sync::mpsc::{Receiver, Sender, channel}, + task::JoinHandle, +}; +use url::Url; +use uuid::Uuid; + +use crate::{ + ASYNC_RUNTIME, INSTANCE_MANAGER, + config::{ + repository::{ + cache_runtime_config_snapshot, config_root_dir, get_runtime_config_snapshot, + load_config_json, save_config_record, + }, + storage::config_meta::get_config_display_name, + }, +}; + +const MAX_NEARBY_SESSIONS: usize = 8; +const MAX_SESSION_KEY_LENGTH: usize = 128; +const MAX_PACKET_BYTES: usize = 64 * 1024; +const MAX_JSON_PAYLOAD_BYTES: usize = 4 * 1024 * 1024; +const MAX_DRAIN_PACKETS: usize = 256; +const PACKET_QUEUE_CAPACITY: usize = 256; +const RPC_PEER_ID: u32 = 1; +const MANAGEMENT_SOCKET_FILE_NAME: &str = "easytier-nearby-management.sock"; +const MANAGEMENT_CONFIG_DIR_NAME: &str = ".easytier-nearby-management"; + +#[derive(Clone, Copy, PartialEq, Eq)] +enum NearbyRole { + Client, + HostProxy, +} + +struct NearbyUnixSocket { + stream: UnixStream, +} + +impl NearbyUnixSocket { + fn new(stream: UnixStream) -> Self { + Self { stream } + } +} + +impl AsyncRead for NearbyUnixSocket { + fn poll_read( + mut self: Pin<&mut Self>, + context: &mut Context<'_>, + buffer: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.stream).poll_read(context, buffer) + } +} + +impl AsyncWrite for NearbyUnixSocket { + fn poll_write( + mut self: Pin<&mut Self>, + context: &mut Context<'_>, + buffer: &[u8], + ) -> Poll> { + Pin::new(&mut self.stream).poll_write(context, buffer) + } + + fn poll_flush(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.stream).poll_flush(context) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.stream).poll_shutdown(context) + } +} + +impl VirtualTcpSocket for NearbyUnixSocket { + fn local_addr(&self) -> io::Result { + Ok(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 1))) + } + + fn peer_addr(&self) -> io::Result { + Ok(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 2))) + } + + fn transport_label(&self) -> Option<&str> { + Some("unix") + } +} + +struct NearbyUnixListener { + socket_path: PathBuf, + listener: Option, +} + +impl NearbyUnixListener { + fn new(socket_path: PathBuf) -> Self { + Self { + socket_path, + listener: None, + } + } +} + +impl fmt::Debug for NearbyUnixListener { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("NearbyUnixListener") + .field("socket_path", &self.socket_path) + .field("listening", &self.listener.is_some()) + .finish() + } +} + +#[async_trait] +impl SocketListener for NearbyUnixListener { + type Accepted = Box; + + async fn listen(&mut self) -> anyhow::Result<()> { + if self.listener.is_some() { + return Ok(()); + } + if let Some(parent) = self.socket_path.parent() { + std::fs::create_dir_all(parent)?; + } + if self.socket_path.exists() { + std::fs::remove_file(&self.socket_path)?; + } + let listener = UnixListener::bind(&self.socket_path)?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt as _; + std::fs::set_permissions(&self.socket_path, std::fs::Permissions::from_mode(0o600))?; + } + self.listener = Some(listener); + Ok(()) + } + + async fn accept(&mut self) -> anyhow::Result { + let listener = self + .listener + .as_ref() + .ok_or_else(|| anyhow::anyhow!("nearby management listener is not started"))?; + let (stream, _) = listener.accept().await?; + let tunnel_info = TunnelInfo { + tunnel_type: "unix".to_owned(), + local_addr: Some(self.local_url().into()), + remote_addr: None, + resolved_remote_addr: None, + }; + Ok(TcpTunnelUpgrader::new(tunnel_info).upgrade(NearbyUnixSocket::new(stream))?) + } + + fn local_url(&self) -> Url { + management_socket_url(&self.socket_path) + } +} + +impl Drop for NearbyUnixListener { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.socket_path); + } +} + +struct NearbyConfigStorage; + +#[async_trait] +impl ConfigFileStorage for NearbyConfigStorage { + async fn inspect(&self, path: &Path) -> ConfigFileControl { + match config_id_from_management_path(path) { + Ok(_) => ConfigFileControl::new( + Some(path.to_owned()), + ConfigFilePermission::from(ConfigFilePermission::NO_DELETE), + ), + Err(_) => ConfigFileControl::new( + Some(path.to_owned()), + ConfigFilePermission::from( + ConfigFilePermission::READ_ONLY | ConfigFilePermission::NO_DELETE, + ), + ), + } + } + + async fn read(&self, path: &Path) -> anyhow::Result>> { + let config_id = config_id_from_management_path(path)?; + let Some(raw) = load_config_json(&config_id.to_string()) else { + return Ok(None); + }; + let config = serde_json::from_str::(&raw)?; + Ok(Some(config.gen_config()?.dump().into_bytes())) + } + + async fn write(&self, path: &Path, contents: &[u8]) -> anyhow::Result<()> { + let config_id = config_id_from_management_path(path)?; + let toml = std::str::from_utf8(contents)?; + let config = TomlConfigLoader::new_from_str(toml)?; + if config.get_id() != config_id { + anyhow::bail!("nearby management cannot change the running instance ID"); + } + let network_config = network_config_from_toml(&config); + let display_name = get_config_display_name(&config_id.to_string()) + .or_else(|| { + get_runtime_config_snapshot(&config_id.to_string()) + .map(|snapshot| snapshot.display_name) + }) + .unwrap_or_else(|| config.get_inst_name()); + let raw = serde_json::to_string(&network_config)?; + save_config_record(config_id.to_string(), display_name, raw) + .ok_or_else(|| anyhow::anyhow!("failed to persist nearby management config"))?; + Ok(()) + } + + async fn remove(&self, _path: &Path) -> anyhow::Result<()> { + anyhow::bail!("nearby management never deletes a saved HarmonyOS config") + } +} + +struct NearbyMutationHooks; + +#[async_trait] +impl InstanceMutationHooks for NearbyMutationHooks { + async fn pre_run_network_instance(&self, config: &TomlConfig) -> Result<(), String> { + let instance_ids = INSTANCE_MANAGER.instance_ids(); + if instance_ids.len() != 1 || instance_ids[0] != config.get_id() { + return Err( + "nearby management may only overwrite the single running HarmonyOS instance" + .to_owned(), + ); + } + Ok(()) + } + + async fn post_run_network_instance(&self, instance_id: &Uuid) -> Result<(), String> { + let config = INSTANCE_MANAGER + .config(*instance_id) + .ok_or_else(|| format!("instance {instance_id} is missing after restart"))?; + let display_name = get_config_display_name(&instance_id.to_string()) + .or_else(|| { + get_runtime_config_snapshot(&instance_id.to_string()) + .map(|snapshot| snapshot.display_name) + }) + .unwrap_or_else(|| config.get_inst_name()); + cache_runtime_config_snapshot( + instance_id.to_string(), + display_name, + network_config_from_toml(&config), + ); + Ok(()) + } +} + +#[derive(Clone)] +struct NearbyWebClientService { + inner: ProcessManagementRpc, +} + +impl NearbyWebClientService { + fn new() -> Self { + Self { + inner: ProcessManagementRpc::new( + INSTANCE_MANAGER.clone(), + Arc::new(NearbyMutationHooks), + Arc::new(NearbyConfigStorage), + ), + } + } + + fn ensure_safe_overwrite(&self, request: &RunNetworkInstanceRequest) -> Result<(), RpcError> { + if !request.overwrite { + return Err(anyhow::anyhow!("nearby management requires an explicit overwrite").into()); + } + let requested_id = request + .inst_id + .clone() + .map(Uuid::from) + .ok_or_else(|| anyhow::anyhow!("nearby management requires an instance ID"))?; + let running = INSTANCE_MANAGER.instance_ids(); + if running.len() != 1 || running[0] != requested_id { + return Err(anyhow::anyhow!( + "nearby management may only overwrite the single running HarmonyOS instance" + ) + .into()); + } + let expected_path = management_config_path(requested_id) + .ok_or_else(|| anyhow::anyhow!("HarmonyOS config store is not initialized"))?; + let control = INSTANCE_MANAGER + .config_control(requested_id) + .ok_or_else(|| anyhow::anyhow!("running instance config control is missing"))?; + if control.path.as_deref() != Some(expected_path.as_path()) + || control.is_read_only() + || !control.is_no_delete() + { + return Err(anyhow::anyhow!( + "running instance is not owned by the HarmonyOS nearby console" + ) + .into()); + } + Ok(()) + } +} + +#[async_trait] +impl WebClientService for NearbyWebClientService { + type Controller = BaseController; + + async fn validate_config( + &self, + controller: BaseController, + request: ValidateConfigRequest, + ) -> Result { + self.inner.validate_config(controller, request).await + } + + async fn run_network_instance( + &self, + controller: BaseController, + request: RunNetworkInstanceRequest, + ) -> Result { + self.ensure_safe_overwrite(&request)?; + self.inner.run_network_instance(controller, request).await + } + + async fn retain_network_instance( + &self, + _controller: BaseController, + _request: RetainNetworkInstanceRequest, + ) -> Result { + Err(anyhow::anyhow!("nearby management cannot retain or remove HarmonyOS instances").into()) + } + + async fn collect_network_info( + &self, + controller: BaseController, + request: CollectNetworkInfoRequest, + ) -> Result { + self.inner.collect_network_info(controller, request).await + } + + async fn list_network_instance( + &self, + controller: BaseController, + request: ListNetworkInstanceRequest, + ) -> Result { + self.inner.list_network_instance(controller, request).await + } + + async fn delete_network_instance( + &self, + _controller: BaseController, + _request: DeleteNetworkInstanceRequest, + ) -> Result { + Err(anyhow::anyhow!("nearby management cannot stop or delete a HarmonyOS instance").into()) + } + + async fn get_network_instance_config( + &self, + controller: BaseController, + request: GetNetworkInstanceConfigRequest, + ) -> Result { + self.inner + .get_network_instance_config(controller, request) + .await + } + + async fn list_network_instance_meta( + &self, + controller: BaseController, + request: ListNetworkInstanceMetaRequest, + ) -> Result { + self.inner + .list_network_instance_meta(controller, request) + .await + } +} + +struct NearbyManagementHostServer { + socket_path: PathBuf, + _server: StandAloneServer, +} + +static NEARBY_HOST_SERVER: once_cell::sync::Lazy>> = + once_cell::sync::Lazy::new(|| Mutex::new(None)); + +pub(crate) fn runtime_management_config_control(instance_id: Uuid) -> Option { + management_config_path(instance_id).map(|path| { + ConfigFileControl::new( + Some(path), + ConfigFilePermission::from(ConfigFilePermission::NO_DELETE), + ) + }) +} + +pub(crate) fn ensure_runtime_management_server_started() -> bool { + let Some(socket_path) = management_socket_path() else { + ohrs_log_error!("[Rust] nearby management config store is not initialized"); + return false; + }; + let Ok(mut state) = NEARBY_HOST_SERVER.lock() else { + return false; + }; + if state + .as_ref() + .is_some_and(|server| server.socket_path == socket_path && socket_path.exists()) + { + return true; + } + state.take(); + + let mut server = StandAloneServer::new(NearbyUnixListener::new(socket_path.clone())); + register_instance_management_rpc(INSTANCE_MANAGER.clone(), server.registry()); + server.registry().register( + LoggerRpcServer::new(LoggerManagementRpc::new(Arc::new(NativeLoggerControl))), + "", + ); + server.registry().register( + WebClientServiceServer::new(NearbyWebClientService::new()), + "", + ); + if let Err(error) = ASYNC_RUNTIME.block_on(server.serve()) { + ohrs_log_error!("[Rust] nearby management server failed to start: {}", error); + return false; + } + *state = Some(NearbyManagementHostServer { + socket_path, + _server: server, + }); + true +} + +pub(crate) fn stop_runtime_management_server() -> bool { + NEARBY_HOST_SERVER + .lock() + .map(|mut state| state.take().is_some()) + .unwrap_or(false) +} + +struct NearbyManagementSession { + role: NearbyRole, + rpc: Option>, + inbound: Mutex>>>, + outbound: Mutex>>, + tasks: Mutex>>, + closed: AtomicBool, +} + +impl NearbyManagementSession { + fn client() -> Self { + let rpc = Arc::new(BidirectRpcManager::new()); + let tunnel = rpc.run_and_create_tunnel(); + Self::from_split(NearbyRole::Client, Some(rpc), tunnel.split()) + } + + fn host_proxy() -> anyhow::Result { + let socket_path = management_socket_path() + .ok_or_else(|| anyhow::anyhow!("HarmonyOS config store is not initialized"))?; + let stream = ASYNC_RUNTIME.block_on(UnixStream::connect(&socket_path))?; + let tunnel_info = TunnelInfo { + tunnel_type: "unix".to_owned(), + local_addr: None, + remote_addr: Some(management_socket_url(&socket_path).into()), + resolved_remote_addr: None, + }; + let tunnel = TcpTunnelUpgrader::new(tunnel_info).upgrade(NearbyUnixSocket::new(stream))?; + Ok(Self::from_split( + NearbyRole::HostProxy, + None, + tunnel.split(), + )) + } + + fn from_split( + role: NearbyRole, + rpc: Option>, + split: SplitTunnel, + ) -> Self { + let (mut bridge_stream, mut bridge_sink) = split; + let (inbound_tx, mut inbound_rx) = channel::>(PACKET_QUEUE_CAPACITY); + let (outbound_tx, outbound_rx) = channel::>(PACKET_QUEUE_CAPACITY); + + let inbound_task = ASYNC_RUNTIME.spawn(async move { + while let Some(bytes) = inbound_rx.recv().await { + let packet = + ZCPacket::new_from_buf(BytesMut::from(bytes.as_slice()), ZCPacketType::NIC); + if bridge_sink.send(packet).await.is_err() { + break; + } + } + let _ = bridge_sink.close().await; + }); + let outbound_task = ASYNC_RUNTIME.spawn(async move { + while let Some(result) = bridge_stream.next().await { + let Ok(packet) = result else { + break; + }; + let bytes = packet.convert_type(ZCPacketType::NIC).into_bytes().to_vec(); + if outbound_tx.send(bytes).await.is_err() { + break; + } + } + }); + + Self { + role, + rpc, + inbound: Mutex::new(Some(inbound_tx)), + outbound: Mutex::new(outbound_rx), + tasks: Mutex::new(vec![inbound_task, outbound_task]), + closed: AtomicBool::new(false), + } + } + + fn push_packet(&self, bytes: Vec) -> bool { + if self.closed.load(Ordering::Acquire) || !valid_rpc_packet_bytes(&bytes) { + return false; + } + self.inbound + .lock() + .ok() + .and_then(|guard| guard.as_ref().cloned()) + .is_some_and(|sender| sender.try_send(bytes).is_ok()) + } + + fn drain_packets(&self) -> Vec { + let Ok(mut receiver) = self.outbound.lock() else { + return Vec::new(); + }; + let mut packets = Vec::new(); + while packets.len() < MAX_DRAIN_PACKETS { + match receiver.try_recv() { + Ok(packet) => packets.push(Uint8Array::from(packet)), + Err(_) => break, + } + } + packets + } + + async fn call_json( + &self, + service_name: &str, + method_name: &str, + domain_name: &str, + payload: Value, + ) -> Result { + if self.role != NearbyRole::Client { + return Err(RpcError::ExecutionError(anyhow::anyhow!( + "nearby host proxy sessions cannot originate management calls" + ))); + } + let rpc = self.rpc.as_ref().ok_or_else(|| { + RpcError::ExecutionError(anyhow::anyhow!("nearby RPC client is unavailable")) + })?; + call_official_management_json_rpc( + rpc.rpc_client(), + service_name, + method_name, + domain_name, + payload, + ) + .await + } + + fn shutdown(self: Arc) { + if self.closed.swap(true, Ordering::AcqRel) { + return; + } + if let Ok(mut inbound) = self.inbound.lock() { + inbound.take(); + } + if let Ok(mut tasks) = self.tasks.lock() { + for task in tasks.drain(..) { + task.abort(); + } + } + if let Some(rpc) = self.rpc.clone() { + ASYNC_RUNTIME.spawn(async move { + rpc.stop().await; + }); + } + } +} + +static NEARBY_SESSIONS: once_cell::sync::Lazy< + Mutex>>, +> = once_cell::sync::Lazy::new(|| Mutex::new(HashMap::new())); + +fn management_socket_path() -> Option { + config_root_dir().map(|root| root.join(MANAGEMENT_SOCKET_FILE_NAME)) +} + +fn management_socket_url(path: &Path) -> Url { + format!("unix://{}", path.display()) + .parse() + .expect("HarmonyOS sandbox path must form a Unix URL") +} + +fn management_config_path(instance_id: Uuid) -> Option { + config_root_dir().map(|root| { + root.join(MANAGEMENT_CONFIG_DIR_NAME) + .join(format!("{instance_id}.toml")) + }) +} + +fn config_id_from_management_path(path: &Path) -> anyhow::Result { + let file_stem = path + .file_stem() + .and_then(|value| value.to_str()) + .ok_or_else(|| anyhow::anyhow!("nearby management config path is invalid"))?; + let instance_id = Uuid::parse_str(file_stem)?; + let expected = management_config_path(instance_id) + .ok_or_else(|| anyhow::anyhow!("HarmonyOS config store is not initialized"))?; + if expected != path { + anyhow::bail!("nearby management config path is outside the managed store"); + } + Ok(instance_id) +} + +fn valid_session_key(session_key: &str) -> bool { + !session_key.is_empty() + && session_key.len() <= MAX_SESSION_KEY_LENGTH + && session_key + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b':' | b'.')) +} + +fn valid_rpc_packet_bytes(bytes: &[u8]) -> bool { + if bytes.is_empty() || bytes.len() > MAX_PACKET_BYTES { + return false; + } + let packet = ZCPacket::new_from_buf(BytesMut::from(bytes), ZCPacketType::NIC); + let Some(header) = packet.peer_manager_header() else { + return false; + }; + matches!( + header.packet_type, + value if value == PacketType::RpcReq as u8 || value == PacketType::RpcResp as u8 + ) +} + +fn session(session_key: &str) -> Option> { + NEARBY_SESSIONS + .lock() + .ok() + .and_then(|sessions| sessions.get(session_key).cloned()) +} + +fn open_session(session_key: String, role: NearbyRole) -> bool { + if !valid_session_key(&session_key) { + return false; + } + let session = match role { + NearbyRole::Client => NearbyManagementSession::client(), + NearbyRole::HostProxy => match NearbyManagementSession::host_proxy() { + Ok(session) => session, + Err(error) => { + ohrs_log_error!("[Rust] nearby management host proxy failed: {}", error); + return false; + } + }, + }; + let Ok(mut sessions) = NEARBY_SESSIONS.lock() else { + return false; + }; + if sessions.contains_key(&session_key) || sessions.len() >= MAX_NEARBY_SESSIONS { + return false; + } + sessions.insert(session_key, Arc::new(session)); + true +} + +pub(crate) fn open_nearby_management_session(session_key: String, host: bool) -> bool { + open_session( + session_key, + if host { + NearbyRole::HostProxy + } else { + NearbyRole::Client + }, + ) +} + +pub(crate) fn close_nearby_management_session(session_key: String) -> bool { + let removed = NEARBY_SESSIONS + .lock() + .ok() + .and_then(|mut sessions| sessions.remove(&session_key)); + if let Some(session) = removed { + session.shutdown(); + true + } else { + false + } +} + +pub(crate) fn push_nearby_management_packet(session_key: String, packet: Uint8Array) -> bool { + session(&session_key).is_some_and(|session| session.push_packet(packet.to_vec())) +} + +pub(crate) fn drain_nearby_management_packets(session_key: String) -> Vec { + session(&session_key) + .map(|session| session.drain_packets()) + .unwrap_or_default() +} + +pub(crate) async fn call_nearby_management_json_rpc( + session_key: String, + service_name: String, + method_name: String, + domain_name: Option, + payload_json: String, +) -> String { + if payload_json.len() > MAX_JSON_PAYLOAD_BYTES { + return json!({ "ok": false, "error": "management payload is too large" }).to_string(); + } + let Some(session) = session(&session_key) else { + return json!({ "ok": false, "error": "nearby management session not found" }).to_string(); + }; + let payload = match serde_json::from_str::(&payload_json) { + Ok(payload) => payload, + Err(error) => { + return json!({ "ok": false, "error": format!("invalid management JSON: {error}") }) + .to_string(); + } + }; + match session + .call_json( + service_name.trim(), + method_name.trim(), + domain_name.as_deref().unwrap_or_default().trim(), + payload, + ) + .await + { + Ok(result) => json!({ "ok": true, "result": result }).to_string(), + Err(error) => json!({ "ok": false, "error": error.to_string() }).to_string(), + } +} + +async fn call_official_management_json_rpc( + client: &Client, + service_name: &str, + method_name: &str, + domain_name: &str, + payload: Value, +) -> Result { + let domain = domain_name.to_owned(); + let controller = BaseController::default(); + match service_name { + "api.manage.WebClientService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + "api.logger.LoggerRpcService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + "api.instance.PeerManageRpcService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + "api.instance.PeerCenterManageRpcService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + "api.instance.ConnectorManageRpcService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + "api.instance.MappedListenerManageRpcService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + "api.instance.VpnPortalRpcService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + "api.instance.TcpProxyRpcService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + "api.instance.AclManageRpcService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + "api.instance.PortForwardManageRpcService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + "api.instance.StatsRpcService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + "api.instance.CredentialManageRpcService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + "api.config.ConfigRpcService" => { + client + .scoped_client::>( + RPC_PEER_ID, + RPC_PEER_ID, + domain, + ) + .json_call_method(controller, method_name, payload) + .await + } + _ => Err(RpcError::InvalidServiceKey( + service_name.to_owned(), + service_name.to_owned(), + )), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rejects_non_rpc_packets_before_the_harmony_transport() { + assert!(!valid_rpc_packet_bytes(&[])); + assert!(!valid_rpc_packet_bytes(&[1, 2, 3])); + assert!(!valid_session_key("contains space")); + assert!(valid_session_key("42:client")); + } +}