diff --git a/easytier-core/src/gateway/vpn_portal/runtime.rs b/easytier-core/src/gateway/vpn_portal/runtime.rs index ca5349e8..083c9ace 100644 --- a/easytier-core/src/gateway/vpn_portal/runtime.rs +++ b/easytier-core/src/gateway/vpn_portal/runtime.rs @@ -7,7 +7,7 @@ use std::{ collections::{BTreeMap, BTreeSet}, net::{IpAddr, Ipv4Addr}, - sync::Arc, + sync::{Arc, RwLock as StdRwLock}, }; use async_trait::async_trait; @@ -119,6 +119,15 @@ pub trait PortalHost: Send + Sync + 'static { fn name(&self) -> String; fn render_client_config(&self, plan: &PortalClientConfigPlan) -> String; + + /// Replaces the configured client set at runtime without restarting the + /// listeners. Established sessions of untouched clients must stay intact; + /// sessions of removed or changed clients are torn down through the + /// regular channel-close cleanup path. + async fn update_clients(&self, clients: &[PortalClientConfig]) -> anyhow::Result<()> { + let _ = clients; + anyhow::bail!("portal host does not support runtime client updates") + } } #[derive(Debug, Clone)] @@ -154,11 +163,11 @@ pub struct PortalModule { operation: Mutex<()>, peer_manager: Arc, runtime_config: CoreRuntimeConfigStore, - config: Option, + config: Option>>, host: Option>, events: Arc, statuses: Arc>>, - session_locks: Arc>>>, + session_locks: Arc>>>>, runtime: Mutex>, } @@ -171,37 +180,17 @@ impl PortalModule { events: Arc, ) -> anyhow::Result> { if let Some(config) = config.as_ref() { - validate_config(config, runtime_config.snapshot().as_ref())?; + validate_clients(config, runtime_config.snapshot().as_ref())?; } - let statuses = config - .as_ref() - .map(|config| { - config - .clients - .iter() - .map(|client| (client.name.clone(), ClientStatus::default())) - .collect() - }) - .unwrap_or_default(); - let session_locks = config - .as_ref() - .map(|config| { - config - .clients - .iter() - .map(|client| (client.name.clone(), Arc::new(Mutex::new(())))) - .collect() - }) - .unwrap_or_default(); Ok(Arc::new(Self { operation: Mutex::new(()), peer_manager, runtime_config, - config, + config: config.map(|config| Arc::new(StdRwLock::new(config))), host, events, - statuses: Arc::new(RwLock::new(statuses)), - session_locks: Arc::new(session_locks), + statuses: Arc::new(RwLock::new(BTreeMap::new())), + session_locks: Arc::new(RwLock::new(BTreeMap::new())), runtime: Mutex::new(None), })) } @@ -213,7 +202,77 @@ impl PortalModule { let Some(config) = self.config.as_ref() else { return Ok(()); }; - validate_runtime_compatibility(config, runtime_config) + let config = config.read().unwrap(); + validate_runtime_compatibility(&config, runtime_config) + } + + /// Replaces the configured client set at runtime. Untouched clients keep + /// their established sessions; removed and changed clients are torn down + /// through the regular cleanup path and may re-handshake afterwards. + /// + /// The caller supplies the runtime snapshot the new set must be validated + /// against, so a combined configuration patch is judged by its final + /// state rather than the currently running one. + pub async fn update_clients( + &self, + clients: Vec, + runtime: &CoreInstanceRuntimeConfig, + ) -> anyhow::Result> { + let _operation = self.operation.lock().await; + let Some(config) = self.config.as_ref() else { + anyhow::bail!("VPN portal is not configured"); + }; + if self.host.is_none() { + anyhow::bail!("VPN portal has no host adapter"); + } + let candidate = PortalRuntimeConfig { clients }; + validate_clients(&candidate, runtime)?; + + self.host + .as_ref() + .expect("checked above") + .update_clients(&candidate.clients) + .await?; + + let applied: BTreeSet = candidate + .clients + .iter() + .map(|client| client.name.clone()) + .collect(); + let removed: Vec = { + let mut current = config.write().unwrap(); + let removed: Vec = current + .clients + .iter() + .map(|client| client.name.clone()) + .filter(|name| !applied.contains(name)) + .collect(); + current.clients = candidate.clients.clone(); + removed + }; + let applied_clients = candidate.clients; + + { + let mut statuses = self.statuses.write().await; + statuses.retain(|name, _| applied.contains(name)); + for name in applied { + statuses.entry(name).or_default(); + } + } + { + let mut locks = self.session_locks.write().await; + for name in removed { + // Entries still held by a live session are left alone; the + // session drops its reference during its regular cleanup. + if locks + .get(&name) + .is_none_or(|lock| Arc::strong_count(lock) == 1) + { + locks.remove(&name); + } + } + } + Ok(applied_clients) } pub async fn start(&self) -> anyhow::Result<()> { @@ -288,9 +347,9 @@ impl PortalModule { listener_url: url::Url, peer_manager: Arc, runtime_config: CoreRuntimeConfigStore, - config: PortalRuntimeConfig, + config: Arc>, statuses: Arc>>, - session_locks: Arc>>>, + session_locks: Arc>>>>, events: Arc, cancel: CancellationToken, start_signal: CancellationToken, @@ -338,13 +397,15 @@ impl PortalModule { listener_url: url::Url, peer_manager: Arc, runtime_config: CoreRuntimeConfigStore, - config: PortalRuntimeConfig, + config: Arc>, statuses: Arc>>, - session_locks: Arc>>>, + session_locks: Arc>>>>, events: Arc, cancel: CancellationToken, ) { let Some(client) = config + .read() + .unwrap() .clients .iter() .find(|client| client.name == session.client_name) @@ -353,18 +414,17 @@ impl PortalModule { tracing::warn!(client = %session.client_name, "unknown VPN portal client session"); return; }; - let session_lock = session_locks - .get(&client.name) - .expect("validated client session lock exists"); + let session_lock = { + let mut locks = session_locks.write().await; + locks.entry(client.name.clone()).or_default().clone() + }; let _session_guard = tokio::select! { _ = cancel.cancelled() => return, guard = session_lock.lock() => guard, }; let generation = { let mut statuses = statuses.write().await; - let status = statuses - .get_mut(&client.name) - .expect("validated client status exists"); + let status = statuses.entry(client.name.clone()).or_default(); status.generation = status.generation.wrapping_add(1); status.state = PortalClientState::Connecting; status.endpoint = Some(session.endpoint.borrow_and_update().clone()); @@ -400,9 +460,11 @@ impl PortalModule { { let mut statuses = statuses.write().await; - let status = statuses - .get_mut(&client.name) - .expect("validated client status exists"); + let Some(status) = statuses.get_mut(&client.name) else { + drop(statuses); + attached.close().await; + return; + }; if status.generation != generation { drop(statuses); attached.close().await; @@ -617,7 +679,11 @@ impl PortalModule { } pub async fn info_snapshot(&self) -> PortalInfoSnapshot { - let Some(config) = self.config.as_ref() else { + let Some(config) = self + .config + .as_ref() + .map(|config| config.read().unwrap().clone()) + else { return PortalInfoSnapshot { vpn_type: "null".to_owned(), clients: Vec::new(), @@ -692,13 +758,14 @@ impl PortalModule { } } -fn validate_config( +/// Validates a client set. The empty set is legal in every lifecycle stage: +/// a portal with zero clients keeps listening and accepts nothing, so +/// clearing all clients never produces a configuration that fails a later +/// instance recreation. +fn validate_clients( config: &PortalRuntimeConfig, runtime_config: &CoreInstanceRuntimeConfig, ) -> anyhow::Result<()> { - if config.clients.is_empty() { - anyhow::bail!("VPN portal requires at least one configured client"); - } if config.clients.len() > MAX_VPN_PORTAL_CLIENTS { anyhow::bail!("VPN portal supports at most {MAX_VPN_PORTAL_CLIENTS} clients"); } @@ -1102,7 +1169,7 @@ mod tests { client("alice", Ipv4Addr::new(10, 82, 0, 3), &["ops"]), ], }; - let error = validate_config(&duplicate_name, snapshot.as_ref()) + let error = validate_clients(&duplicate_name, snapshot.as_ref()) .unwrap_err() .to_string(); assert!(error.contains("duplicate VPN portal client name")); @@ -1110,7 +1177,7 @@ mod tests { let duplicate_ip = PortalRuntimeConfig { clients: vec![alice, client("bob", Ipv4Addr::new(10, 82, 0, 2), &["ops"])], }; - let error = validate_config(&duplicate_ip, snapshot.as_ref()) + let error = validate_clients(&duplicate_ip, snapshot.as_ref()) .unwrap_err() .to_string(); assert!(error.contains("duplicate VPN portal virtual IP")); @@ -1122,7 +1189,7 @@ mod tests { let config = PortalRuntimeConfig { clients: vec![client("alice", Ipv4Addr::new(10, 82, 0, 2), &["unknown"])], }; - let error = validate_config(&config, runtime_config.snapshot().as_ref()) + let error = validate_clients(&config, runtime_config.snapshot().as_ref()) .unwrap_err() .to_string(); assert!(error.contains("unknown ACL group")); @@ -1138,7 +1205,7 @@ mod tests { clients: vec![client("alice", Ipv4Addr::new(10, 82, 0, 2), &["ops"])], }; - let error = validate_config(&config, runtime_config.snapshot().as_ref()) + let error = validate_clients(&config, runtime_config.snapshot().as_ref()) .unwrap_err() .to_string(); @@ -1175,10 +1242,10 @@ mod tests { "alice".to_owned(), ClientStatus::default(), )]))); - let session_locks = Arc::new(BTreeMap::from([( + let session_locks = Arc::new(RwLock::new(BTreeMap::from([( "alice".to_owned(), Arc::new(Mutex::new(())), - )])); + )]))); let (to_runtime, from_client) = mpsc::channel(1); let (to_client, _from_runtime) = mpsc::channel(1); let (endpoint_sender, endpoint) = tokio::sync::watch::channel("portal://alice".to_owned()); @@ -1195,7 +1262,7 @@ mod tests { "portal://listener".parse().unwrap(), peer_manager.clone(), runtime_config, - config, + Arc::new(StdRwLock::new(config)), statuses.clone(), session_locks, Arc::new(()), @@ -1263,10 +1330,10 @@ mod tests { "alice".to_owned(), ClientStatus::default(), )]))); - let session_locks = Arc::new(BTreeMap::from([( + let session_locks = Arc::new(RwLock::new(BTreeMap::from([( "alice".to_owned(), Arc::new(Mutex::new(())), - )])); + )]))); let (_to_runtime, from_client) = mpsc::channel(1); let (to_client, _from_runtime) = mpsc::channel(1); let (_endpoint_sender, endpoint) = tokio::sync::watch::channel("portal://alice".to_owned()); @@ -1287,7 +1354,7 @@ mod tests { "portal://listener".parse().unwrap(), peer_manager.clone(), runtime_config, - config, + Arc::new(StdRwLock::new(config)), statuses.clone(), session_locks, events.clone(), @@ -1373,10 +1440,10 @@ mod tests { "alice".to_owned(), ClientStatus::default(), )]))); - let session_locks = Arc::new(BTreeMap::from([( + let session_locks = Arc::new(RwLock::new(BTreeMap::from([( "alice".to_owned(), Arc::new(Mutex::new(())), - )])); + )]))); let (to_runtime, from_client) = mpsc::channel(1); let (to_client, from_runtime) = mpsc::channel(1); let (_endpoint_sender, endpoint) = tokio::sync::watch::channel("portal://alice".to_owned()); @@ -1393,7 +1460,7 @@ mod tests { "portal://listener".parse().unwrap(), peer_manager.clone(), runtime_config, - config, + Arc::new(StdRwLock::new(config)), statuses.clone(), session_locks, events.clone(), @@ -1652,4 +1719,144 @@ mod tests { assert!(validate_client_name("laptop_1").is_err()); assert!(validate_client_name("").is_err()); } + + #[derive(Default)] + struct RecordingPortalHost { + updates: StdMutex>>, + } + + impl RecordingPortalHost { + fn recorded(&self) -> Vec> { + self.updates.lock().unwrap().clone() + } + } + + #[async_trait] + impl PortalHost for RecordingPortalHost { + async fn start_listeners(&self) -> anyhow::Result> { + anyhow::bail!("recording host never starts listeners") + } + + fn name(&self) -> String { + "recording".to_owned() + } + + fn render_client_config(&self, plan: &PortalClientConfigPlan) -> String { + format!("config:{}", plan.name) + } + + async fn update_clients(&self, clients: &[PortalClientConfig]) -> anyhow::Result<()> { + self.updates.lock().unwrap().push(clients.to_vec()); + Ok(()) + } + } + + fn portal_module_with_recording_host( + initial: PortalRuntimeConfig, + ) -> ( + Arc, + Arc, + CoreRuntimeConfigStore, + ) { + let (peer_manager, runtime_config) = network_runtime(); + let host = Arc::new(RecordingPortalHost::default()); + let module = PortalModule::new( + peer_manager, + runtime_config.clone(), + Some(initial), + Some(host.clone()), + Arc::new(()), + ) + .unwrap(); + (module, host, runtime_config) + } + + #[tokio::test] + async fn portal_module_update_clients_rejects_invalid_sets() { + let (module, _host, runtime_config) = + portal_module_with_recording_host(PortalRuntimeConfig { + clients: vec![client("alice", Ipv4Addr::new(10, 82, 0, 2), &["ops"])], + }); + let runtime = runtime_config.snapshot(); + + let duplicate = module + .update_clients( + vec![ + client("bob", Ipv4Addr::new(10, 82, 0, 3), &["ops"]), + client("bob", Ipv4Addr::new(10, 82, 0, 4), &["ops"]), + ], + runtime.as_ref(), + ) + .await + .unwrap_err(); + assert!( + duplicate + .to_string() + .contains("duplicate VPN portal client name") + ); + + let unknown_group = module + .update_clients( + vec![client("bob", Ipv4Addr::new(10, 82, 0, 3), &["missing"])], + runtime.as_ref(), + ) + .await + .unwrap_err(); + assert!(unknown_group.to_string().contains("unknown ACL group")); + + // Runtime updates may drain the portal to zero clients. + let applied = module + .update_clients(Vec::new(), runtime.as_ref()) + .await + .unwrap(); + assert!(applied.is_empty()); + } + + #[tokio::test] + async fn portal_module_update_clients_replaces_shared_state_and_notifies_host() { + let (module, host, runtime_config) = + portal_module_with_recording_host(PortalRuntimeConfig { + clients: vec![client("alice", Ipv4Addr::new(10, 82, 0, 2), &["ops"])], + }); + + let applied = module + .update_clients( + vec![client("bob", Ipv4Addr::new(10, 82, 0, 3), &["ops"])], + runtime_config.snapshot().as_ref(), + ) + .await + .unwrap(); + assert_eq!(applied.len(), 1); + assert_eq!(applied[0].name, "bob"); + + let recorded = host.recorded(); + assert_eq!(recorded.len(), 1); + assert_eq!(recorded[0].len(), 1); + assert_eq!(recorded[0][0].name, "bob"); + + let snapshot = module.info_snapshot().await; + assert_eq!(snapshot.clients.len(), 1); + assert_eq!(snapshot.clients[0].name, "bob"); + } + + #[tokio::test] + async fn portal_module_update_clients_requires_configured_portal() { + let (peer_manager, runtime_config) = network_runtime(); + let module = PortalModule::new( + peer_manager, + runtime_config.clone(), + None, + None, + Arc::new(()), + ) + .unwrap(); + let error = module + .update_clients( + vec![client("alice", Ipv4Addr::new(10, 82, 0, 2), &["ops"])], + runtime_config.snapshot().as_ref(), + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("not configured")); + } } diff --git a/easytier-core/src/instance/vpn_portal_extension.rs b/easytier-core/src/instance/vpn_portal_extension.rs index a46d5630..77db302a 100644 --- a/easytier-core/src/instance/vpn_portal_extension.rs +++ b/easytier-core/src/instance/vpn_portal_extension.rs @@ -1,5 +1,6 @@ use crate::{ - gateway::vpn_portal::PortalInfoSnapshot, + config::runtime::CoreInstanceRuntimeConfig, + gateway::vpn_portal::{PortalClientConfig, PortalInfoSnapshot}, instance::{CoreInstance, CoreInstanceHost}, }; @@ -10,4 +11,19 @@ where pub async fn vpn_portal_info(&self) -> PortalInfoSnapshot { self.vpn_portal.info_snapshot().await } + + /// Replaces the VPN portal client set at runtime without restarting the + /// instance. Untouched clients keep their established sessions. + /// + /// The caller supplies the runtime snapshot the new set is validated + /// against, so a combined configuration patch is judged by its final + /// state rather than the currently running one. + #[cfg(feature = "vpn-portal")] + pub(crate) async fn update_vpn_portal_clients( + &self, + clients: Vec, + runtime: &CoreInstanceRuntimeConfig, + ) -> anyhow::Result> { + self.vpn_portal.update_clients(clients, runtime).await + } } diff --git a/easytier-core/src/management/full/config_patch.rs b/easytier-core/src/management/full/config_patch.rs index d32545d8..37b15075 100644 --- a/easytier-core/src/management/full/config_patch.rs +++ b/easytier-core/src/management/full/config_patch.rs @@ -3,7 +3,7 @@ use std::{fmt::Debug, sync::Arc}; use anyhow::Context as _; use easytier_proto::api::config::{ self, AclPatch, ConfigPatchAction, ExitNodePatch, InstanceConfigPatch, Patchable, - PortForwardPatch, ProxyNetworkPatch, RoutePatch, UrlPatch, + PortForwardPatch, ProxyNetworkPatch, RoutePatch, UrlPatch, VpnPortalClientPatch, }; use crate::{ @@ -95,6 +95,35 @@ where candidate.set_ipv6_public_addr_prefix(prefix); provider_config_changed = true; } + + // Runs last so client validation sees the fully patched candidate, + // including routes and the node IPv4 set earlier in this request. + if !patch.vpn_portal_clients.is_empty() { + apply_vpn_portal_client_patches(&candidate, patch.vpn_portal_clients)?; + // Deep-validate and hot-apply before committing, so a rejected + // client set leaves neither the shared TOML model nor the live + // portal changed. + let normalized = validate_candidate(instance, &candidate)?; + #[cfg(feature = "vpn-portal")] + { + let portal = normalized + .vpn_portal + .clone() + .ok_or_else(|| anyhow::anyhow!("VPN portal is not configured"))?; + instance + .update_vpn_portal_clients( + portal.clients, + &runtime_config_from_normalized(&normalized), + ) + .await?; + } + #[cfg(not(feature = "vpn-portal"))] + { + let _ = normalized; + } + validate_and_commit_candidate(instance, &config, &candidate)?; + } + let normalized = validate_and_commit_candidate(instance, &config, &candidate)?; let runtime = runtime_config_from_normalized(&normalized); instance @@ -120,9 +149,8 @@ where Ok(()) } -fn validate_and_commit_candidate( +fn validate_candidate( instance: &CoreInstance, - shared: &TomlConfig, candidate: &TomlConfig, ) -> anyhow::Result where @@ -132,6 +160,18 @@ where let runtime = runtime_config_from_normalized(&normalized); runtime.services.public_ipv6_provider.validate()?; instance.validate_runtime_config_capabilities(&runtime)?; + Ok(normalized) +} + +fn validate_and_commit_candidate( + instance: &CoreInstance, + shared: &TomlConfig, + candidate: &TomlConfig, +) -> anyhow::Result +where + H: CoreInstanceHost, +{ + let normalized = validate_candidate(instance, candidate)?; shared.replace_from_snapshot(candidate); Ok(normalized) } @@ -312,6 +352,67 @@ fn patch_mapped_listeners(config: &TomlConfig, patches: Vec) -> anyhow Ok(()) } +/// Applies VPN portal client patches to the candidate TOML model. The live +/// portal is updated by the caller after the candidate commits, so deep +/// validation runs against the final configuration state. +fn apply_vpn_portal_client_patches( + config: &TomlConfig, + patches: Vec, +) -> anyhow::Result<()> { + if patches.is_empty() { + return Ok(()); + } + let mut portal = config + .get_vpn_portal_config() + .ok_or_else(|| anyhow::anyhow!("VPN portal is not configured; cannot patch its clients"))?; + for patch in patches { + match ConfigPatchAction::try_from(patch.action) { + Ok(ConfigPatchAction::Add) => { + let Some(client) = patch.client else { + tracing::warn!("ignored VPN portal client add without client"); + continue; + }; + let virtual_ip = client + .virtual_ip + .parse::() + .with_context(|| { + format!( + "invalid VPN portal client virtual IP: {}", + client.virtual_ip + ) + })?; + portal + .clients + .push(crate::config::toml::VpnPortalClientConfig { + name: client.name, + virtual_ip, + groups: client.groups, + }); + } + Ok(ConfigPatchAction::Remove) => { + let Some(client) = patch.client else { + tracing::warn!("ignored VPN portal client remove without client"); + continue; + }; + let before = portal.clients.len(); + portal + .clients + .retain(|existing| existing.name != client.name); + if portal.clients.len() == before { + anyhow::bail!("VPN portal client not found: {}", client.name); + } + } + Ok(ConfigPatchAction::Clear) => portal.clients.clear(), + Err(_) => tracing::warn!( + action = patch.action, + "ignored invalid VPN portal client action" + ), + } + } + config.set_vpn_portal_config(portal); + Ok(()) +} + fn patch_connectors(instance: &CoreInstance, patches: Vec) -> anyhow::Result<()> where H: CoreInstanceHost, @@ -346,3 +447,97 @@ where } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::toml::{VpnPortalClientConfig, VpnPortalConfig}; + use easytier_proto::api::manage::VpnPortalClientConfig as ClientPb; + + fn portal_config() -> TomlConfig { + let config = TomlConfig::default(); + config.set_vpn_portal_config(VpnPortalConfig { + wireguard_listen: "0.0.0.0:51820".parse().unwrap(), + wireguard_private_key: None, + clients: vec![VpnPortalClientConfig { + name: "alice".to_owned(), + virtual_ip: "10.0.0.2".parse().unwrap(), + groups: Vec::new(), + }], + }); + config + } + + fn configured_names(config: &TomlConfig) -> Vec { + config + .get_vpn_portal_config() + .unwrap() + .clients + .into_iter() + .map(|client| client.name) + .collect() + } + + fn add(name: &str, ip: &str) -> VpnPortalClientPatch { + VpnPortalClientPatch { + action: ConfigPatchAction::Add as i32, + client: Some(ClientPb { + name: name.to_owned(), + virtual_ip: ip.to_owned(), + groups: Vec::new(), + }), + } + } + + fn remove(name: &str) -> VpnPortalClientPatch { + VpnPortalClientPatch { + action: ConfigPatchAction::Remove as i32, + client: Some(ClientPb { + name: name.to_owned(), + virtual_ip: String::new(), + groups: Vec::new(), + }), + } + } + + #[test] + fn vpn_portal_client_patches_add_remove_and_clear() { + let config = portal_config(); + + apply_vpn_portal_client_patches(&config, vec![add("bob", "10.0.0.3")]).unwrap(); + assert_eq!(configured_names(&config), ["alice", "bob"]); + + apply_vpn_portal_client_patches(&config, vec![remove("alice")]).unwrap(); + assert_eq!(configured_names(&config), ["bob"]); + + apply_vpn_portal_client_patches( + &config, + vec![VpnPortalClientPatch { + action: ConfigPatchAction::Clear as i32, + client: None, + }], + ) + .unwrap(); + assert!(configured_names(&config).is_empty()); + } + + #[test] + fn vpn_portal_client_patches_reject_missing_prerequisites() { + let bare = TomlConfig::default(); + let error = + apply_vpn_portal_client_patches(&bare, vec![add("alice", "10.0.0.2")]).unwrap_err(); + assert!(error.to_string().contains("not configured")); + + let config = portal_config(); + let error = apply_vpn_portal_client_patches(&config, vec![remove("ghost")]).unwrap_err(); + assert!(error.to_string().contains("not found")); + + let error = + apply_vpn_portal_client_patches(&config, vec![add("bob", "not-an-ip")]).unwrap_err(); + assert!( + error + .to_string() + .contains("invalid VPN portal client virtual IP") + ); + } +} diff --git a/easytier-gui/src-tauri/src/lib.rs b/easytier-gui/src-tauri/src/lib.rs index c6c169e0..b0053ebc 100644 --- a/easytier-gui/src-tauri/src/lib.rs +++ b/easytier-gui/src-tauri/src/lib.rs @@ -6,12 +6,16 @@ mod elevate; use anyhow::Context; #[cfg(target_os = "android")] use easytier::instance::factory::subscribe_native_instance_event; +use easytier::proto::api::config::{ + ConfigPatchAction, ConfigRpc, ConfigRpcClientFactory, InstanceConfigPatch, PatchConfigRequest, + VpnPortalClientPatch, +}; use easytier::proto::api::instance::{ GetVpnPortalInfoRequest, InstanceIdentifier, VpnPortalInfo, VpnPortalRpc, VpnPortalRpcClientFactory, instance_identifier, }; use easytier::proto::api::manage::{ - CollectNetworkInfoResponse, ValidateConfigResponse, WebClientService, + CollectNetworkInfoResponse, ValidateConfigResponse, VpnPortalClientConfig, WebClientService, WebClientServiceClientFactory, }; use easytier::proto::rpc_types::controller::BaseController; @@ -180,6 +184,58 @@ async fn get_vpn_portal_info(instance_id: String) -> Result, + virtual_ip: Option, + groups: Option>, +) -> Result<(), String> { + let instance_id = instance_id + .parse::() + .map_err(|e| e.to_string())?; + let action = match action.as_str() { + "add" => ConfigPatchAction::Add, + "remove" => ConfigPatchAction::Remove, + "clear" => ConfigPatchAction::Clear, + other => return Err(format!("invalid vpn portal client patch action: {other}")), + }; + let client = if action == ConfigPatchAction::Clear { + None + } else { + Some(VpnPortalClientConfig { + name: name.unwrap_or_default(), + virtual_ip: virtual_ip.unwrap_or_default(), + groups: groups.unwrap_or_default(), + }) + }; + + let client_manager = get_client_manager!()?; + let rpc = client_manager + .rpc_manager + .rpc_client() + .scoped_client::>(1, 1, "".to_string()); + rpc.patch_config( + BaseController::default(), + PatchConfigRequest { + instance: Some(InstanceIdentifier { + selector: Some(instance_identifier::Selector::Id(instance_id.into())), + }), + patch: Some(InstanceConfigPatch { + vpn_portal_clients: vec![VpnPortalClientPatch { + action: action as i32, + client, + }], + ..Default::default() + }), + }, + ) + .await + .map_err(|e| e.to_string())?; + Ok(()) +} + #[tauri::command] async fn set_logging_level(level: String) -> Result<(), String> { get_client_manager!()? @@ -1423,6 +1479,7 @@ pub fn run_gui() -> std::process::ExitCode { run_network_instance, collect_network_info, get_vpn_portal_info, + patch_vpn_portal_clients, set_logging_level, set_tun_fd, easytier_version, diff --git a/easytier-gui/src/composables/backend.ts b/easytier-gui/src/composables/backend.ts index 1a779aa7..a66ca615 100644 --- a/easytier-gui/src/composables/backend.ts +++ b/easytier-gui/src/composables/backend.ts @@ -71,6 +71,18 @@ export async function getVpnPortalInfo(instanceId: string) { return info ? NetworkTypes.normalizeVpnPortalInfo(info) : undefined } +export async function addVpnPortalClient(instanceId: string, client: { name: string, virtual_ip: string, groups: string[] }) { + return invoke('patch_vpn_portal_clients', { instanceId, action: 'add', name: client.name, virtualIp: client.virtual_ip, groups: client.groups }) +} + +export async function removeVpnPortalClient(instanceId: string, name: string) { + return invoke('patch_vpn_portal_clients', { instanceId, action: 'remove', name }) +} + +export async function clearVpnPortalClients(instanceId: string) { + return invoke('patch_vpn_portal_clients', { instanceId, action: 'clear' }) +} + export async function setLoggingLevel(level: string) { return await invoke('set_logging_level', { level }) } diff --git a/easytier-gui/src/modules/api.ts b/easytier-gui/src/modules/api.ts index ca4a97ae..b0076939 100644 --- a/easytier-gui/src/modules/api.ts +++ b/easytier-gui/src/modules/api.ts @@ -14,6 +14,15 @@ export class GUIRemoteClient implements Api.RemoteClient { async get_vpn_portal_info(inst_id: string): Promise { return backend.getVpnPortalInfo(inst_id); } + async add_vpn_portal_client(inst_id: string, client: { name: string, virtual_ip: string, groups: string[] }): Promise { + await backend.addVpnPortalClient(inst_id, client); + } + async remove_vpn_portal_client(inst_id: string, name: string): Promise { + await backend.removeVpnPortalClient(inst_id, name); + } + async clear_vpn_portal_clients(inst_id: string): Promise { + await backend.clearVpnPortalClients(inst_id); + } async list_network_instance_ids(): Promise { return backend.listNetworkInstanceIds(); } diff --git a/easytier-proto/proto/api_config.proto b/easytier-proto/proto/api_config.proto index dd200af0..3ea8183f 100644 --- a/easytier-proto/proto/api_config.proto +++ b/easytier-proto/proto/api_config.proto @@ -28,6 +28,12 @@ message InstanceConfigPatch { optional bool ipv6_public_addr_auto = 12; optional string ipv6_public_addr_prefix = 13; optional bool disable_relay_data = 14; + repeated VpnPortalClientPatch vpn_portal_clients = 15; +} + +message VpnPortalClientPatch { + ConfigPatchAction action = 1; + api.manage.VpnPortalClientConfig client = 2; } message PortForwardPatch { diff --git a/easytier-web/frontend-lib/src/modules/api.ts b/easytier-web/frontend-lib/src/modules/api.ts index 46b18158..d333f14f 100644 --- a/easytier-web/frontend-lib/src/modules/api.ts +++ b/easytier-web/frontend-lib/src/modules/api.ts @@ -58,6 +58,9 @@ export interface RemoteClient { run_network(config: NetworkConfig, save: boolean): Promise; get_network_info(inst_id: string): Promise; get_vpn_portal_info(inst_id: string): Promise; + add_vpn_portal_client(inst_id: string, client: { name: string, virtual_ip: string, groups: string[] }): Promise; + remove_vpn_portal_client(inst_id: string, name: string): Promise; + clear_vpn_portal_clients(inst_id: string): Promise; list_network_instance_ids(): Promise; delete_network(inst_id: string): Promise; update_network_instance_state(inst_id: string, disabled: boolean): Promise; diff --git a/easytier-web/frontend/src/modules/api.ts b/easytier-web/frontend/src/modules/api.ts index d584fdcd..bc0e2c2f 100644 --- a/easytier-web/frontend/src/modules/api.ts +++ b/easytier-web/frontend/src/modules/api.ts @@ -237,6 +237,38 @@ class WebRemoteClient implements Api.RemoteClient { ? NetworkTypes.normalizeVpnPortalInfo(response.vpn_portal_info) : undefined; } + async patch_vpn_portal_clients(inst_id: string, patches: Array>): Promise { + await this.client.post( + `/machines/${this.machine_id}/proxy-rpc`, + { + service_name: 'api.config.ConfigRpcService', + method_name: 'patch_config', + payload: { + instance: { + id: Utils.StrToUuid(inst_id), + }, + patch: { + vpn_portal_clients: patches, + }, + }, + }, + ); + } + async add_vpn_portal_client(inst_id: string, client: { name: string, virtual_ip: string, groups: string[] }): Promise { + await this.patch_vpn_portal_clients(inst_id, [{ + action: 'ADD', + client, + }]); + } + async remove_vpn_portal_client(inst_id: string, name: string): Promise { + await this.patch_vpn_portal_clients(inst_id, [{ + action: 'REMOVE', + client: { name, virtual_ip: '', groups: [] }, + }]); + } + async clear_vpn_portal_clients(inst_id: string): Promise { + await this.patch_vpn_portal_clients(inst_id, [{ action: 'CLEAR' }]); + } async list_network_instance_ids(): Promise { const response = await this.client.get('/machines/' + this.machine_id + '/networks'); return response; diff --git a/easytier-web/src/client_manager/runtime_reconcile.rs b/easytier-web/src/client_manager/runtime_reconcile.rs index f955721a..8c10d136 100644 --- a/easytier-web/src/client_manager/runtime_reconcile.rs +++ b/easytier-web/src/client_manager/runtime_reconcile.rs @@ -3,13 +3,15 @@ use easytier::{ common::config::{ ConfigLoader, EncryptionAlgorithm, NetworkConfigExt, PortForwardConfig as RuntimePortForwardConfig, + VpnPortalClientConfig as RuntimeVpnPortalClientConfig, + VpnPortalConfig as RuntimeVpnPortalConfig, }, proto::{ acl::Acl, api::{ config::{ AclPatch, ConfigPatchAction, InstanceConfigPatch, PatchConfigRequest, - PortForwardPatch, ProxyNetworkPatch, + PortForwardPatch, ProxyNetworkPatch, VpnPortalClientPatch, }, instance::{InstanceIdentifier, instance_identifier}, manage::{ @@ -61,6 +63,9 @@ fn hot_patch_base(config: &NetworkConfig) -> anyhow::Result { config.port_forwards.clear(); config.proxy_cidrs.clear(); config.disable_relay_data = None; + // VPN portal clients are diffed separately; the listener identity + // (address and private key) decides between patch and recreate. + config.vpn_portal_config = None; if config.dhcp.unwrap_or_default() { config.virtual_ipv4 = None; config.network_length = None; @@ -204,6 +209,53 @@ fn normalized_disable_relay_data(config: &NetworkConfig) -> anyhow::Result Ok(config.gen_config()?.get_flags().disable_relay_data) } +fn normalized_vpn_portal(config: &NetworkConfig) -> anyhow::Result> { + Ok(config.gen_config()?.get_vpn_portal_config()) +} + +fn diff_vpn_portal_clients( + current: &[RuntimeVpnPortalClientConfig], + desired: &[RuntimeVpnPortalClientConfig], +) -> Vec { + let mut patches = Vec::new(); + // Removals first so a virtual IP moved between clients never exists + // twice inside one patch request. + for client in current { + match desired.iter().find(|desired| desired.name == client.name) { + Some(matching) if matching == client => {} + _ => patches.push(VpnPortalClientPatch { + action: ConfigPatchAction::Remove as i32, + client: Some(client_name_only(&client.name)), + }), + } + } + for client in desired { + if current + .iter() + .find(|existing| existing.name == client.name) + .is_none_or(|existing| existing != client) + { + patches.push(VpnPortalClientPatch { + action: ConfigPatchAction::Add as i32, + client: Some(easytier::proto::api::manage::VpnPortalClientConfig { + name: client.name.clone(), + virtual_ip: client.virtual_ip.to_string(), + groups: client.groups.clone(), + }), + }); + } + } + patches +} + +fn client_name_only(name: &str) -> easytier::proto::api::manage::VpnPortalClientConfig { + easytier::proto::api::manage::VpnPortalClientConfig { + name: name.to_owned(), + virtual_ip: String::new(), + groups: Vec::new(), + } +} + fn web_source_runtime_patch( current: &NetworkConfig, desired: &NetworkConfig, @@ -256,6 +308,27 @@ fn web_source_runtime_patch( patch.disable_relay_data = Some(desired_disable_relay_data); } + match ( + normalized_vpn_portal(current)?, + normalized_vpn_portal(desired)?, + ) { + (Some(current_portal), Some(desired_portal)) => { + if current_portal.wireguard_listen != desired_portal.wireguard_listen + || current_portal.wireguard_private_key != desired_portal.wireguard_private_key + { + // The listener identity changed; the portal must be rebuilt. + return Ok(None); + } + if current_portal.clients != desired_portal.clients { + patch.vpn_portal_clients = + diff_vpn_portal_clients(¤t_portal.clients, &desired_portal.clients); + } + } + // Enabling or disabling the portal changes the listener lifecycle. + (Some(_), None) | (None, Some(_)) => return Ok(None), + (None, None) => {} + } + Ok(Some(patch)) } @@ -424,6 +497,151 @@ mod tests { ) } + fn portal_client(name: &str, ip: &str) -> easytier::proto::api::manage::VpnPortalClientConfig { + easytier::proto::api::manage::VpnPortalClientConfig { + name: name.to_owned(), + virtual_ip: ip.to_owned(), + groups: Vec::new(), + } + } + + fn config_with_vpn_portal( + clients: Vec, + listen: &str, + ) -> NetworkConfig { + let mut config = config_with_port_forwards(Vec::new()); + config.dhcp = Some(false); + config.virtual_ipv4 = Some("10.144.0.1".to_string()); + config.network_length = Some(24); + config.vpn_portal_config = Some(easytier::proto::api::manage::VpnPortalConfig { + wireguard_listen: listen.to_owned(), + wireguard_private_key: Some("dGVzdC1rZXk=".to_owned()), + clients, + }); + config + } + + fn patch_vpn_portal_actions(patch: &InstanceConfigPatch) -> Vec<(i32, String)> { + patch + .vpn_portal_clients + .iter() + .map(|client_patch| { + ( + client_patch.action, + client_patch + .client + .as_ref() + .map(|client| client.name.clone()) + .unwrap_or_default(), + ) + }) + .collect() + } + + #[test] + fn vpn_portal_client_changes_produce_hot_patches() { + let current = config_with_vpn_portal( + vec![ + portal_client("alice", "10.144.144.4"), + portal_client("carol", "10.144.144.6"), + ], + "0.0.0.0:22121", + ); + let desired = config_with_vpn_portal( + vec![ + portal_client("bob", "10.144.144.5"), + portal_client("carol", "10.144.144.7"), + ], + "0.0.0.0:22121", + ); + + let patch = web_source_runtime_patch(¤t, &desired) + .unwrap() + .expect("client-only changes must be hot-patchable"); + + assert_eq!( + patch_vpn_portal_actions(&patch), + vec![ + (ConfigPatchAction::Remove as i32, "alice".to_owned()), + (ConfigPatchAction::Remove as i32, "carol".to_owned()), + (ConfigPatchAction::Add as i32, "bob".to_owned()), + (ConfigPatchAction::Add as i32, "carol".to_owned()), + ], + "removals must precede additions; changed clients are remove+add" + ); + } + + #[test] + fn vpn_portal_client_no_op_produces_empty_patch_section() { + let current = config_with_vpn_portal( + vec![portal_client("alice", "10.144.144.4")], + "0.0.0.0:22121", + ); + let desired = config_with_vpn_portal( + vec![portal_client("alice", "10.144.144.4")], + "0.0.0.0:22121", + ); + + let patch = web_source_runtime_patch(¤t, &desired) + .unwrap() + .unwrap(); + assert!(patch.vpn_portal_clients.is_empty()); + } + + #[test] + fn vpn_portal_listener_identity_change_requires_recreate() { + let current = config_with_vpn_portal( + vec![portal_client("alice", "10.144.144.4")], + "0.0.0.0:22121", + ); + let desired = config_with_vpn_portal( + vec![portal_client("alice", "10.144.144.4")], + "0.0.0.0:22122", + ); + assert!( + web_source_runtime_patch(¤t, &desired) + .unwrap() + .is_none() + ); + + let mut different_key = desired.clone(); + different_key + .vpn_portal_config + .as_mut() + .unwrap() + .wireguard_listen = "0.0.0.0:22121".to_owned(); + different_key + .vpn_portal_config + .as_mut() + .unwrap() + .wireguard_private_key = Some("bm90LXRoZS1zYW1lLWtleQ==".to_owned()); + assert!( + web_source_runtime_patch(¤t, &different_key) + .unwrap() + .is_none() + ); + } + + #[test] + fn vpn_portal_enable_or_disable_requires_recreate() { + let without_portal = config_with_port_forwards(Vec::new()); + let with_portal = config_with_vpn_portal( + vec![portal_client("alice", "10.144.144.4")], + "0.0.0.0:22121", + ); + + assert!( + web_source_runtime_patch(&without_portal, &with_portal) + .unwrap() + .is_none() + ); + assert!( + web_source_runtime_patch(&with_portal, &without_portal) + .unwrap() + .is_none() + ); + } + #[test] fn runtime_patch_ignores_runtime_defaults_and_adds_port_forward() { let mut current = config_with_port_forwards(vec![port_forward(23000, 5174)]); diff --git a/easytier/src/easytier-cli.rs b/easytier/src/easytier-cli.rs index d7fd1cc5..569006d3 100644 --- a/easytier/src/easytier-cli.rs +++ b/easytier/src/easytier-cli.rs @@ -37,6 +37,7 @@ use easytier::{ config::{ AclPatch, ConfigPatchAction, ConfigRpc, ConfigRpcClientFactory, InstanceConfigPatch, PatchConfigRequest, PortForwardPatch, StringPatch, UrlPatch, + VpnPortalClientPatch, }, instance::{ AclManageRpc, AclManageRpcClientFactory, Connector, ConnectorManageRpc, @@ -64,7 +65,8 @@ use easytier::{ SetLoggerConfigRequest, }, manage::{ - ListNetworkInstanceMetaRequest, ListNetworkInstanceRequest, WebClientService, + ListNetworkInstanceMetaRequest, ListNetworkInstanceRequest, + VpnPortalClientConfig as ManageVpnPortalClientConfig, WebClientService, WebClientServiceClientFactory, }, }, @@ -130,8 +132,8 @@ enum SubCommand { Route(RouteArgs), #[command(about = "show global peers info")] PeerCenter, - #[command(about = "show vpn portal (wireguard) info")] - VpnPortal, + #[command(about = "manage vpn portal (wireguard) clients")] + VpnPortal(VpnPortalArgs), #[command(about = "inspect self easytier-core status")] Node(NodeArgs), #[command(about = "manage easytier-core as a system service")] @@ -265,6 +267,32 @@ enum MappedListenerSubCommand { List, } +#[derive(Args, Debug)] +struct VpnPortalArgs { + #[command(subcommand)] + sub_command: Option, +} + +#[derive(Subcommand, Debug)] +enum VpnPortalSubCommand { + /// Add a WireGuard portal client + AddClient { + #[arg(help = "client name")] + name: String, + #[arg(long, help = "client virtual IPv4 address inside the mesh network")] + virtual_ip: String, + #[arg(long, help = "ACL groups assigned to the client")] + groups: Vec, + }, + /// Remove a WireGuard portal client + RemoveClient { + #[arg(help = "client name")] + name: String, + }, + /// Remove all WireGuard portal clients + ClearClients, +} + #[derive(Subcommand, Debug)] enum NodeSubCommand { #[command(about = "show node info")] @@ -2502,6 +2530,86 @@ impl<'a> CommandHandler<'a> { }) } + async fn apply_vpn_portal_client_patch( + &self, + patch: VpnPortalClientPatch, + ) -> Result<(), Error> { + let client = self.get_config_client().await?; + let request = PatchConfigRequest { + instance: Some(self.instance_selector.clone()), + patch: Some(InstanceConfigPatch { + vpn_portal_clients: vec![patch], + ..Default::default() + }), + }; + let _response = client + .patch_config(BaseController::default(), request) + .await?; + Ok(()) + } + + async fn handle_vpn_portal_add_client( + &self, + name: String, + virtual_ip: String, + groups: Vec, + ) -> Result<(), Error> { + virtual_ip + .parse::() + .map_err(|e| anyhow::anyhow!("invalid virtual ip ({virtual_ip}): {e}"))?; + self.apply_to_instances(|handler| { + let name = name.clone(); + let virtual_ip = virtual_ip.clone(); + let groups = groups.clone(); + Box::pin(async move { + handler + .apply_vpn_portal_client_patch(VpnPortalClientPatch { + action: ConfigPatchAction::Add as i32, + client: Some(ManageVpnPortalClientConfig { + name, + virtual_ip, + groups, + }), + }) + .await + }) + }) + .await + } + + async fn handle_vpn_portal_remove_client(&self, name: String) -> Result<(), Error> { + self.apply_to_instances(|handler| { + let name = name.clone(); + Box::pin(async move { + handler + .apply_vpn_portal_client_patch(VpnPortalClientPatch { + action: ConfigPatchAction::Remove as i32, + client: Some(ManageVpnPortalClientConfig { + name, + virtual_ip: String::new(), + groups: Vec::new(), + }), + }) + .await + }) + }) + .await + } + + async fn handle_vpn_portal_clear_clients(&self) -> Result<(), Error> { + self.apply_to_instances(|handler| { + Box::pin(async move { + handler + .apply_vpn_portal_client_patch(VpnPortalClientPatch { + action: ConfigPatchAction::Clear as i32, + client: None, + }) + .await + }) + }) + .await + } + async fn handle_vpn_portal(&self) -> Result<(), Error> { let results = self .collect_instance_results(|handler| Box::pin(handler.fetch_vpn_portal_info())) @@ -3053,9 +3161,24 @@ async fn main() -> Result<(), Error> { SubCommand::PeerCenter => { handler.handle_peer_center().await?; } - SubCommand::VpnPortal => { - handler.handle_vpn_portal().await?; - } + SubCommand::VpnPortal(args) => match args.sub_command { + None => handler.handle_vpn_portal().await?, + Some(VpnPortalSubCommand::AddClient { + name, + virtual_ip, + groups, + }) => { + handler + .handle_vpn_portal_add_client(name, virtual_ip, groups) + .await?; + } + Some(VpnPortalSubCommand::RemoveClient { name }) => { + handler.handle_vpn_portal_remove_client(name).await?; + } + Some(VpnPortalSubCommand::ClearClients) => { + handler.handle_vpn_portal_clear_clients().await?; + } + }, SubCommand::Node(sub_cmd) => { handler.handle_node(sub_cmd.sub_command.as_ref()).await?; } diff --git a/easytier/src/tests/three_node.rs b/easytier/src/tests/three_node.rs index d7b0b7db..8ab5da90 100644 --- a/easytier/src/tests/three_node.rs +++ b/easytier/src/tests/three_node.rs @@ -108,6 +108,11 @@ use crate::{ #[cfg(feature = "wireguard")] use easytier_core::gateway::vpn_portal::PortalClientState; +#[cfg(feature = "wireguard")] +use easytier_proto::api::{ + config::{ConfigPatchAction, InstanceConfigPatch, VpnPortalClientPatch}, + manage::VpnPortalClientConfig as VpnPortalClientConfigPb, +}; pub fn prepare_linux_namespaces() { del_netns("net_a"); @@ -2133,6 +2138,199 @@ pub async fn wireguard_vpn_portal_client_roaming() { drop_insts(insts).await; } +#[cfg(feature = "wireguard")] +#[tokio::test] +#[serial_test::serial] +pub async fn wireguard_vpn_portal_dynamic_clients() { + let insts = init_three_node_ex( + "tcp", + |config| { + let identity = config.get_network_identity(); + config.set_network_identity(NetworkIdentity::new( + identity.network_name, + "wireguard-portal-dynamic-clients-test".to_owned(), + )); + if config.get_inst_name() == "inst3" { + config.set_vpn_portal_config(VpnPortalConfig { + wireguard_listen: "0.0.0.0:22121".parse().unwrap(), + wireguard_private_key: Some(BASE64_STANDARD.encode([42u8; 32])), + clients: vec![VpnPortalClientConfig { + name: "client-a".to_owned(), + virtual_ip: "10.144.144.4".parse().unwrap(), + groups: Vec::new(), + }], + }); + } + config + }, + false, + ) + .await; + + let core = insts[2].get_core_instance(); + let portal_config = insts[2] + .get_global_ctx() + .config + .get_vpn_portal_config() + .unwrap(); + + // 初始客户端上线 + { + let net_ns = NetNS::new(Some("net_d".into())); + let _g = net_ns.guard(); + let (server_public, client_private) = + test_wireguard_keys(&portal_config, "client-a").unwrap(); + run_wireguard_client( + &wireguard_ifname("wg0"), + "10.1.2.3:22121".parse().unwrap(), + Key::try_from(server_public.as_slice()).unwrap(), + Key::try_from(client_private.as_slice()).unwrap(), + vec!["10.144.144.0/24".to_string()], + "192.0.2.42".to_string(), + ) + .unwrap(); + } + wait_for_condition( + || async { ping_test("net_d", "10.144.144.1", None).await }, + Duration::from_secs(10), + ) + .await; + + // 不重启实例，通过配置补丁动态添加第二个客户端 + easytier_core::management::apply_config_patch( + &core, + InstanceConfigPatch { + vpn_portal_clients: vec![VpnPortalClientPatch { + action: ConfigPatchAction::Add as i32, + client: Some(VpnPortalClientConfigPb { + name: "client-b".to_owned(), + virtual_ip: "10.144.144.5".to_owned(), + groups: Vec::new(), + }), + }], + ..Default::default() + }, + ) + .await + .unwrap(); + assert_eq!( + insts[2] + .get_global_ctx() + .config + .get_vpn_portal_config() + .unwrap() + .clients + .len(), + 2, + "shared TOML model must reflect the runtime update" + ); + + // 拒绝的补丁不能污染共享 TOML 模型：重复添加 client-b 必须整体失败 + let error = easytier_core::management::apply_config_patch( + &core, + InstanceConfigPatch { + vpn_portal_clients: vec![VpnPortalClientPatch { + action: ConfigPatchAction::Add as i32, + client: Some(VpnPortalClientConfigPb { + name: "client-b".to_owned(), + virtual_ip: "10.144.144.9".to_owned(), + groups: Vec::new(), + }), + }], + ..Default::default() + }, + ) + .await + .unwrap_err(); + assert!( + error + .to_string() + .contains("duplicate VPN portal client name"), + "unexpected rejection reason: {error:#}" + ); + assert_eq!( + insts[2] + .get_global_ctx() + .config + .get_vpn_portal_config() + .unwrap() + .clients + .len(), + 2, + "rejected patch must leave the shared TOML model unchanged" + ); + + // 新客户端立即可以握手上线，原客户端不受影响 + { + let net_ns = NetNS::new(Some("net_f".into())); + let _g = net_ns.guard(); + let (server_public, client_private) = + test_wireguard_keys(&portal_config, "client-b").unwrap(); + run_wireguard_client( + &wireguard_ifname("wg0"), + "10.1.2.3:22121".parse().unwrap(), + Key::try_from(server_public.as_slice()).unwrap(), + Key::try_from(client_private.as_slice()).unwrap(), + vec!["10.144.144.0/24".to_string()], + "192.0.2.43".to_string(), + ) + .unwrap(); + } + wait_for_condition( + || async { ping_test("net_f", "10.144.144.1", None).await }, + Duration::from_secs(10), + ) + .await; + wait_for_condition( + || async { ping_test("net_d", "10.144.144.1", None).await }, + Duration::from_secs(10), + ) + .await; + + // 动态移除 client-a：其会话被拆除，client-b 保持在线 + easytier_core::management::apply_config_patch( + &core, + InstanceConfigPatch { + vpn_portal_clients: vec![VpnPortalClientPatch { + action: ConfigPatchAction::Remove as i32, + client: Some(VpnPortalClientConfigPb { + name: "client-a".to_owned(), + virtual_ip: String::new(), + groups: Vec::new(), + }), + }], + ..Default::default() + }, + ) + .await + .unwrap(); + + wait_for_condition( + || async { !ping_test("net_d", "10.144.144.1", None).await }, + Duration::from_secs(20), + ) + .await; + wait_for_condition( + || async { ping_test("net_f", "10.144.144.1", None).await }, + Duration::from_secs(10), + ) + .await; + + let info = core.vpn_portal_info().await; + assert_eq!(info.clients.len(), 1); + assert_eq!(info.clients[0].name, "client-b"); + assert_eq!(info.clients[0].state, PortalClientState::Online); + assert_eq!( + info.clients[0].tunnel_ip, + Some("192.0.2.43".parse().unwrap()) + ); + + // Release the held CoreInstance Arc so drop_insts can observe a clean + // drop instead of swallowing its debug assertion. + drop(core); + drop_insts(insts).await; +} + #[cfg(feature = "wireguard")] #[rstest::rstest] #[tokio::test] diff --git a/easytier/src/vpn_portal/wireguard.rs b/easytier/src/vpn_portal/wireguard.rs index a787270c..9a64fba1 100644 --- a/easytier/src/vpn_portal/wireguard.rs +++ b/easytier/src/vpn_portal/wireguard.rs @@ -8,9 +8,10 @@ mod engine; use std::{ + collections::{BTreeMap, BTreeSet}, fmt, net::{Ipv6Addr, SocketAddr, SocketAddrV6}, - sync::Arc, + sync::{Arc, Mutex as StdMutex, RwLock, Weak}, }; use anyhow::Context as _; @@ -18,7 +19,9 @@ use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STANDARD}; use boringtun::x25519::{PublicKey, StaticSecret}; use easytier_core::{ config::toml::VpnPortalConfig, - gateway::vpn_portal::{PortalClientConfigPlan, PortalHost, PortalListener, PortalSession}, + gateway::vpn_portal::{ + PortalClientConfig, PortalClientConfigPlan, PortalHost, PortalListener, PortalSession, + }, socket::{ ListenerConnectionCounter, NetNamespace, SocketContext, SocketListener, udp::{UdpBindOptions, VirtualUdpSocket, VirtualUdpSocketFactory}, @@ -147,36 +150,34 @@ pub struct WireGuardPortalHost { global_ctx: ArcGlobalCtx, config: VpnPortalConfig, setup: Result, + engine: StdMutex>>, } struct WireGuardPortalSetup { server_private: [u8; 32], server_public: PublicKey, - clients: Vec, + clients: RwLock>, } impl WireGuardPortalHost { pub fn new(global_ctx: ArcGlobalCtx, config: VpnPortalConfig) -> Arc { let setup = (|| -> anyhow::Result<_> { - let (master, server_private) = portal_master_and_server_key(&config)?; + // The derivation master equals the server private key. + let (_, server_private) = portal_master_and_server_key(&config)?; let server_public = PublicKey::from(&StaticSecret::from(server_private)); - let mut clients = Vec::with_capacity(config.clients.len()); + let mut clients = BTreeMap::new(); for client in &config.clients { - let wireguard_private = - derive_named_key(&master, b"wireguard-client", &client.name)?; - let identity_private_key = - derive_named_key(&master, b"attached-noise", &client.name)?; - clients.push(DerivedClient { - config: client.clone(), - wireguard_private, - wireguard_public: PublicKey::from(&StaticSecret::from(wireguard_private)), - identity_private_key, - }); + let client = PortalClientConfig { + name: client.name.clone(), + virtual_ip: client.virtual_ip, + groups: client.groups.clone(), + }; + clients.insert(client.name.clone(), derive_client(server_private, &client)?); } Ok(WireGuardPortalSetup { server_private, server_public, - clients, + clients: RwLock::new(clients), }) })() .map_err(|error| error.to_string()); @@ -184,6 +185,7 @@ impl WireGuardPortalHost { global_ctx, config, setup, + engine: StdMutex::new(None), }) } @@ -204,6 +206,69 @@ impl WireGuardPortalHost { ) .await } + + async fn apply_client_updates(&self, clients: &[PortalClientConfig]) -> anyhow::Result<()> { + let setup = self + .setup + .as_ref() + .map_err(|error| anyhow::anyhow!(error.clone()))?; + let engine = self + .engine + .lock() + .unwrap() + .clone() + .and_then(|engine| engine.upgrade()) + .ok_or_else(|| anyhow::anyhow!("WireGuard VPN portal is not running"))?; + + let desired: BTreeSet<&str> = clients.iter().map(|client| client.name.as_str()).collect(); + let stale: Vec = setup + .clients + .read() + .unwrap() + .keys() + .filter(|name| !desired.contains(name.as_str())) + .cloned() + .collect(); + for name in &stale { + engine.remove_client(name).await; + setup.clients.write().unwrap().remove(name); + } + for client in clients { + if setup + .clients + .read() + .unwrap() + .get(&client.name) + .is_some_and(|existing| existing.config == *client) + { + continue; + } + // A changed client keeps its WireGuard identity (keys derive from + // the name), but its attached peer must be rebuilt with the new + // virtual IP or groups. Expiring the session makes the client + // re-handshake into a fresh Core generation. + engine.remove_client(&client.name).await; + let derived = derive_client(setup.server_private, client)?; + engine.add_client(derived.clone())?; + setup + .clients + .write() + .unwrap() + .insert(client.name.clone(), derived); + } + Ok(()) + } +} + +fn derive_client(master: [u8; 32], client: &PortalClientConfig) -> anyhow::Result { + let wireguard_private = derive_named_key(&master, b"wireguard-client", &client.name)?; + let identity_private_key = derive_named_key(&master, b"attached-noise", &client.name)?; + Ok(DerivedClient { + config: client.clone(), + wireguard_private, + wireguard_public: PublicKey::from(&StaticSecret::from(wireguard_private)), + identity_private_key, + }) } fn secondary_ipv6_bind_address(address: SocketAddr, primary_port: u16) -> Option { let SocketAddr::V4(address) = address else { @@ -237,7 +302,12 @@ impl PortalHost for WireGuardPortalHost { } let url = url::Url::parse(&format!("wg://{local}"))?; let (accepted, receiver) = mpsc::unbounded_channel(); - let engine = PortalEngine::new(setup.server_private, setup.clients.clone(), accepted); + let engine = PortalEngine::new( + setup.server_private, + setup.clients.read().unwrap().values().cloned().collect(), + accepted, + ); + *self.engine.lock().unwrap() = Some(Arc::downgrade(&engine)); Ok(vec![Box::new(WireGuardPortalListener { url, sockets, @@ -252,27 +322,25 @@ impl PortalHost for WireGuardPortalHost { "wireguard".to_owned() } + async fn update_clients(&self, clients: &[PortalClientConfig]) -> anyhow::Result<()> { + self.apply_client_updates(clients).await + } + fn render_client_config(&self, plan: &PortalClientConfigPlan) -> String { - let client = self + let setup = self .setup .as_ref() - .expect("client config is rendered only after successful startup") - .clients - .iter() - .find(|client| client.config.name == plan.name) - .expect("Core only renders configured clients"); + .expect("client config is rendered only after successful startup"); + let clients = setup.clients.read().unwrap(); + let Some(client) = clients.get(&plan.name) else { + return String::new(); + }; let endpoint = &plan.listener_url[url::Position::BeforeHost..url::Position::AfterPort]; format!( "[Interface]\nPrivateKey = {}\nAddress = {}/32\n\n[Peer]\nPublicKey = {}\nAllowedIPs = {}\nEndpoint = {} # replace wildcard with the public address\nPersistentKeepalive = 25\n", BASE64_STANDARD.encode(client.wireguard_private), plan.address, - BASE64_STANDARD.encode( - self.setup - .as_ref() - .expect("client config is rendered only after successful startup") - .server_public - .as_bytes() - ), + BASE64_STANDARD.encode(setup.server_public.as_bytes()), plan.allowed_ips.join(", "), endpoint, ) diff --git a/easytier/src/vpn_portal/wireguard/engine.rs b/easytier/src/vpn_portal/wireguard/engine.rs index 8f83c46f..ad906945 100644 --- a/easytier/src/vpn_portal/wireguard/engine.rs +++ b/easytier/src/vpn_portal/wireguard/engine.rs @@ -2,9 +2,12 @@ use atomic_shim::AtomicU64; use std::{ - collections::HashMap, + collections::{BTreeSet, HashMap}, net::SocketAddr, - sync::{Arc, atomic::Ordering}, + sync::{ + Arc, RwLock, + atomic::{AtomicBool, Ordering}, + }, time::Duration, }; @@ -16,7 +19,7 @@ use boringtun::{ x25519::{PublicKey, StaticSecret}, }; use easytier_core::{ - config::toml::VpnPortalClientConfig, gateway::vpn_portal::PortalSession, + gateway::vpn_portal::{PortalClientConfig, PortalSession}, socket::udp::VirtualUdpSocket, }; use tokio::{ @@ -35,7 +38,7 @@ const TIMER_INTERVAL: Duration = Duration::from_millis(250); const PORTAL_PACKET_CAPACITY: usize = 128; #[derive(Clone)] pub(super) struct DerivedClient { - pub(super) config: VpnPortalClientConfig, + pub(super) config: PortalClientConfig, pub(super) wireguard_private: [u8; 32], pub(super) wireguard_public: PublicKey, pub(super) identity_private_key: [u8; 32], @@ -63,6 +66,30 @@ struct ClientSlot { index: u32, next_generation: AtomicU64, session: Mutex>, + retired: AtomicBool, +} + +#[derive(Default)] +struct EngineSlots { + by_name: HashMap>, + by_public_key: HashMap<[u8; 32], Arc>, + by_index: HashMap>, + free_indices: BTreeSet, + highest_index: u32, +} + +impl EngineSlots { + fn allocate_index(&mut self) -> anyhow::Result { + if let Some(index) = self.free_indices.pop_first() { + return Ok(index); + } + let next = self + .highest_index + .checked_add(1) + .ok_or_else(|| anyhow::anyhow!("WireGuard portal client index space is exhausted"))?; + self.highest_index = next; + Ok(next) + } } #[derive(Clone)] @@ -88,8 +115,7 @@ pub(super) struct PortalEngine { server_private: StaticSecret, server_public: PublicKey, rate_limiter: Arc, - by_public_key: HashMap<[u8; 32], Arc>, - by_index: HashMap>, + slots: RwLock, accepted: mpsc::UnboundedSender, cancel: CancellationToken, } @@ -102,19 +128,24 @@ impl PortalEngine { ) -> Arc { let server_private = StaticSecret::from(server_private); let server_public = PublicKey::from(&server_private); - let mut by_public_key = HashMap::with_capacity(clients.len()); - let mut by_index = HashMap::with_capacity(clients.len()); - for (offset, client) in clients.into_iter().enumerate() { - let index = u32::try_from(offset + 1).expect("client limit is below u32"); + let mut slots = EngineSlots::default(); + for client in clients { let public = *client.wireguard_public.as_bytes(); + let index = slots + .allocate_index() + .expect("initial portal clients fit the index space"); let slot = Arc::new(ClientSlot { client, index, next_generation: AtomicU64::new(1), session: Mutex::new(None), + retired: AtomicBool::new(false), }); - by_public_key.insert(public, slot.clone()); - by_index.insert(index, slot); + slots + .by_name + .insert(slot.client.config.name.clone(), slot.clone()); + slots.by_public_key.insert(public, slot.clone()); + slots.by_index.insert(index, slot); } Arc::new(Self { server_private, @@ -123,18 +154,61 @@ impl PortalEngine { &server_public, DOUBLE_VERIFY_HANDSHAKE_LIMIT, )), - by_public_key, - by_index, + slots: RwLock::new(slots), accepted, cancel: CancellationToken::new(), }) } + pub(super) fn add_client(&self, client: DerivedClient) -> anyhow::Result<()> { + let public = *client.wireguard_public.as_bytes(); + let name = client.config.name.clone(); + let mut slots = self.slots.write().unwrap(); + if slots.by_name.contains_key(&name) || slots.by_public_key.contains_key(&public) { + anyhow::bail!("WireGuard portal client {name} already exists"); + } + let index = slots.allocate_index()?; + let slot = Arc::new(ClientSlot { + client, + index, + next_generation: AtomicU64::new(1), + session: Mutex::new(None), + retired: AtomicBool::new(false), + }); + slots.by_name.insert(name, slot.clone()); + slots.by_public_key.insert(public, slot.clone()); + slots.by_index.insert(index, slot); + Ok(()) + } + + /// Removes a client by name. Any active session is expired so Core tears + /// down the attached peer through its regular channel-close path. + pub(super) async fn remove_client(&self, name: &str) -> bool { + let slot = { + let mut slots = self.slots.write().unwrap(); + slots.by_name.remove(name).inspect(|slot| { + slot.retired.store(true, Ordering::Relaxed); + let public = *slot.client.wireguard_public.as_bytes(); + slots.by_public_key.remove(&public); + slots.by_index.remove(&slot.index); + slots.free_indices.insert(slot.index); + }) + }; + let Some(slot) = slot else { + return false; + }; + let expired = slot.session.lock().await.take(); + Self::retire_session(expired); + true + } + pub(super) fn cancel(&self) { self.cancel.cancel(); } pub(super) fn connection_count(&self) -> u32 { - self.by_index + let slots = self.slots.read().unwrap(); + slots + .by_index .values() .filter(|slot| { slot.session.try_lock().is_ok_and(|guard| { @@ -169,7 +243,10 @@ impl PortalEngine { parse_handshake_anon(&self.server_private, &self.server_public, init) .ok() .and_then(|handshake| { - self.by_public_key + self.slots + .read() + .unwrap() + .by_public_key .get(&handshake.peer_static_public) .cloned() }) @@ -179,8 +256,17 @@ impl PortalEngine { Packet::PacketData(data) => self.slot_by_receiver(data.receiver_idx), }; let Some(slot) = slot else { return }; + if slot.retired.load(Ordering::Relaxed) { + return; + } let mut session = slot.session.lock().await; + // Re-check after acquiring the lock: remove_client retires the slot + // and drains the session under this same lock, so a datagram that + // raced with removal cannot resurrect a session here. + if slot.retired.load(Ordering::Relaxed) { + return; + } if session.is_none() { if !matches!(parsed, Packet::HandshakeInit(_)) { return; @@ -277,7 +363,12 @@ impl PortalEngine { } fn slot_by_receiver(&self, receiver: u32) -> Option> { - self.by_index.get(&(receiver >> 8)).cloned() + self.slots + .read() + .unwrap() + .by_index + .get(&(receiver >> 8)) + .cloned() } fn new_session( @@ -399,7 +490,15 @@ impl PortalEngine { _ = interval.tick() => {} } self.rate_limiter.reset_count(); - for slot in self.by_index.values() { + let slots = self + .slots + .read() + .unwrap() + .by_index + .values() + .cloned() + .collect::>(); + for slot in slots { let mut output = [0u8; 148]; let mut guard = slot.session.lock().await; let Some(session) = guard.as_mut() else { @@ -433,3 +532,57 @@ fn is_handshake_response_packet(packet: &[u8]) -> bool { fn is_transport_data_packet(packet: &[u8]) -> bool { packet.len() >= 32 && packet.get(..4) == Some(&4u32.to_le_bytes()) } + +#[cfg(test)] +mod tests { + use super::*; + + fn derived(name: &str, seed: u8) -> DerivedClient { + let secret = StaticSecret::from([seed; 32]); + DerivedClient { + config: PortalClientConfig { + name: name.to_owned(), + virtual_ip: "192.0.2.1".parse().unwrap(), + groups: Vec::new(), + }, + wireguard_private: secret.to_bytes(), + wireguard_public: PublicKey::from(&secret), + identity_private_key: [seed.wrapping_add(1); 32], + } + } + + fn slot_index(engine: &PortalEngine, name: &str) -> Option { + engine + .slots + .read() + .unwrap() + .by_name + .get(name) + .map(|slot| slot.index) + } + + #[tokio::test] + async fn remove_client_drops_slot_and_recycles_index() { + let (accepted, _receiver) = mpsc::unbounded_channel(); + let engine = PortalEngine::new([1; 32], vec![derived("a", 10), derived("b", 11)], accepted); + assert_eq!(slot_index(&engine, "a"), Some(1)); + assert_eq!(slot_index(&engine, "b"), Some(2)); + + assert!(engine.remove_client("a").await); + assert!(!engine.remove_client("a").await); + + engine.add_client(derived("c", 12)).unwrap(); + assert_eq!(slot_index(&engine, "c"), Some(1), "freed index is reused"); + assert!( + engine.add_client(derived("c", 13)).is_err(), + "duplicate client name is rejected" + ); + assert!( + engine.add_client(derived("d", 11)).is_err(), + "duplicate client public key is rejected" + ); + + engine.add_client(derived("d", 14)).unwrap(); + assert_eq!(slot_index(&engine, "d"), Some(3)); + } +}