From e38b1354b346ce59b6665ddf95b803e78873ab83 Mon Sep 17 00:00:00 2001 From: KKRainbow <443152178@qq.com> Date: Sun, 7 Jun 2026 12:40:09 +0800 Subject: [PATCH] Fix credential ospf logic, fix udp subnet proxy loop protection (#2315) --- .../easytier-ffi/src/data_plane_async.rs | 6 +- easytier-contrib/easytier-ffi/src/lib.rs | 145 +++ easytier-gui/src-tauri/src/lib.rs | 66 +- easytier-gui/src/composables/backend.ts | 7 +- easytier-gui/src/composables/config_source.ts | 13 + easytier-gui/src/composables/event.ts | 5 +- easytier-web/src/client_manager/mod.rs | 33 +- easytier-web/src/client_manager/session.rs | 908 ++++++++---------- easytier-web/src/client_manager/storage.rs | 35 + easytier-web/src/db/mod.rs | 12 +- ...260421_000003_add_network_config_source.rs | 2 +- ...0260514_000004_rename_web_config_source.rs | 42 + easytier-web/src/migrator/mod.rs | 2 + easytier-web/src/restful/network.rs | 56 +- easytier-web/src/restful/rpc.rs | 22 +- easytier-web/src/webhook.rs | 19 +- easytier/src/common/config.rs | 16 +- easytier/src/gateway/udp_proxy.rs | 341 ++++++- easytier/src/peers/credential_manager.rs | 81 ++ easytier/src/peers/peer_manager.rs | 137 ++- easytier/src/peers/peer_ospf_route.rs | 609 ++++++++++-- easytier/src/peers/tests.rs | 3 - easytier/src/proto/api_manage.proto | 2 +- easytier/src/rpc_service/remote_client.rs | 33 +- 24 files changed, 1892 insertions(+), 703 deletions(-) create mode 100644 easytier-gui/src/composables/config_source.ts create mode 100644 easytier-web/src/migrator/m20260514_000004_rename_web_config_source.rs diff --git a/easytier-contrib/easytier-ffi/src/data_plane_async.rs b/easytier-contrib/easytier-ffi/src/data_plane_async.rs index 8fb684a3..41241b7e 100644 --- a/easytier-contrib/easytier-ffi/src/data_plane_async.rs +++ b/easytier-contrib/easytier-ffi/src/data_plane_async.rs @@ -78,7 +78,7 @@ struct DataPlaneAsyncOp { #[cfg(feature = "ffi-dataplane")] enum DataPlaneAsyncOpState { Pending, - Ready(DataPlaneAsyncOpResult), + Ready(Box), Failed(String), Consumed, } @@ -158,7 +158,7 @@ fn complete_op(op: &DataPlaneAsyncOp, result: Result DataPlaneAsyncOpState::Ready(result), + Ok(result) => DataPlaneAsyncOpState::Ready(Box::new(result)), Err(err) => DataPlaneAsyncOpState::Failed(err), }; op.ready.notify_all(); @@ -325,7 +325,7 @@ fn take_completed_op( }; match completed { - DataPlaneAsyncOpState::Ready(result) => Some(result), + DataPlaneAsyncOpState::Ready(result) => Some(*result), DataPlaneAsyncOpState::Failed(err) => { set_error_msg(&err); None diff --git a/easytier-contrib/easytier-ffi/src/lib.rs b/easytier-contrib/easytier-ffi/src/lib.rs index a56012af..58fb49d6 100644 --- a/easytier-contrib/easytier-ffi/src/lib.rs +++ b/easytier-contrib/easytier-ffi/src/lib.rs @@ -551,6 +551,15 @@ pub extern "C" fn data_plane_free_bytes(ptr: *const c_uchar, len: u32) { data_plane_async::data_plane_free_bytes(ptr, len) } +/// Start an asynchronous TCP data-plane connection. +/// +/// # Safety +/// `inst_name` and `dst_ip` must be non-null pointers to null-terminated UTF-8 +/// strings. The strings only need to remain valid for the duration of this +/// call. +/// +/// # Return +/// Returns a non-zero async operation handle on success, or `0` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_tcp_connect_start( @@ -564,6 +573,18 @@ pub unsafe extern "C" fn data_plane_tcp_connect_start( } } +/// Finish an asynchronous TCP data-plane connection. +/// +/// On success, writes the stream local address into `out_local_ip` and +/// `out_local_port`. The returned IP string is allocated by this library and +/// must be released with `free_string`. +/// +/// # Safety +/// `out_local_ip` and `out_local_port` must be non-null pointers to writable +/// storage. +/// +/// # Return +/// Returns a non-zero TCP stream handle on success, or `0` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_tcp_connect_finish( @@ -576,6 +597,14 @@ pub unsafe extern "C" fn data_plane_tcp_connect_finish( } } +/// Start an asynchronous TCP data-plane bind. +/// +/// # Safety +/// `inst_name` must be a non-null pointer to a null-terminated UTF-8 string. +/// The string only needs to remain valid for the duration of this call. +/// +/// # Return +/// Returns a non-zero async operation handle on success, or `0` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_tcp_bind_start( @@ -586,6 +615,18 @@ pub unsafe extern "C" fn data_plane_tcp_bind_start( unsafe { data_plane_async::data_plane_tcp_bind_start(inst_name, local_port, timeout_ms) } } +/// Finish an asynchronous TCP data-plane bind. +/// +/// On success, writes the listener local address into `out_local_ip` and +/// `out_local_port`. The returned IP string is allocated by this library and +/// must be released with `free_string`. +/// +/// # Safety +/// `out_local_ip` and `out_local_port` must be non-null pointers to writable +/// storage. +/// +/// # Return +/// Returns a non-zero TCP listener handle on success, or `0` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_tcp_bind_finish( @@ -596,12 +637,33 @@ pub unsafe extern "C" fn data_plane_tcp_bind_finish( unsafe { data_plane_async::data_plane_tcp_bind_finish(op_handle, out_local_ip, out_local_port) } } +/// Start an asynchronous TCP data-plane accept on a listener handle. +/// +/// # Safety +/// `handle` must be a valid TCP listener handle returned by +/// `data_plane_tcp_bind` or `data_plane_tcp_bind_finish`. +/// +/// # Return +/// Returns a non-zero async operation handle on success, or `0` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_tcp_accept_start(handle: u64, timeout_ms: u64) -> u64 { unsafe { data_plane_async::data_plane_tcp_accept_start(handle, timeout_ms) } } +/// Finish an asynchronous TCP data-plane accept. +/// +/// On success, writes the accepted stream local address into `out_local_ip` and +/// `out_local_port`, and the peer address into `out_peer_ip` and +/// `out_peer_port`. Returned IP strings are allocated by this library and must +/// be released with `free_string`. +/// +/// # Safety +/// `out_local_ip`, `out_local_port`, `out_peer_ip`, and `out_peer_port` must be +/// non-null pointers to writable storage. +/// +/// # Return +/// Returns a non-zero TCP stream handle on success, or `0` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_tcp_accept_finish( @@ -622,6 +684,14 @@ pub unsafe extern "C" fn data_plane_tcp_accept_finish( } } +/// Start an asynchronous TCP data-plane read. +/// +/// # Safety +/// `handle` must be a valid TCP stream handle returned by +/// `data_plane_tcp_connect_finish` or `data_plane_tcp_accept_finish`. +/// +/// # Return +/// Returns a non-zero async operation handle on success, or `0` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_tcp_read_start( @@ -632,6 +702,17 @@ pub unsafe extern "C" fn data_plane_tcp_read_start( unsafe { data_plane_async::data_plane_tcp_read_start(handle, max_len, timeout_ms) } } +/// Finish an asynchronous TCP data-plane read. +/// +/// On success, writes the received buffer pointer and length into `out_buf` and +/// `out_len`. The returned buffer is allocated by this library and must be +/// released with `data_plane_free_bytes`. +/// +/// # Safety +/// `out_buf` and `out_len` must be non-null pointers to writable storage. +/// +/// # Return +/// Returns the number of bytes read, or `-1` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_tcp_read_finish( @@ -642,6 +723,17 @@ pub unsafe extern "C" fn data_plane_tcp_read_finish( unsafe { data_plane_async::data_plane_tcp_read_finish(op_handle, out_buf, out_len) } } +/// Start an asynchronous TCP data-plane write. +/// +/// The input bytes are copied before this function returns. +/// +/// # Safety +/// `handle` must be a valid TCP stream handle returned by +/// `data_plane_tcp_connect_finish` or `data_plane_tcp_accept_finish`. If `len` +/// is non-zero, `buf` must be non-null and readable for `len` bytes. +/// +/// # Return +/// Returns a non-zero async operation handle on success, or `0` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_tcp_write_start( @@ -659,6 +751,14 @@ pub extern "C" fn data_plane_tcp_write_finish(op_handle: u64) -> c_int { data_plane_async::data_plane_tcp_write_finish(op_handle) } +/// Start an asynchronous UDP data-plane bind. +/// +/// # Safety +/// `inst_name` must be a non-null pointer to a null-terminated UTF-8 string. +/// The string only needs to remain valid for the duration of this call. +/// +/// # Return +/// Returns a non-zero async operation handle on success, or `0` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_udp_bind_start( @@ -669,6 +769,18 @@ pub unsafe extern "C" fn data_plane_udp_bind_start( unsafe { data_plane_async::data_plane_udp_bind_start(inst_name, local_port, timeout_ms) } } +/// Finish an asynchronous UDP data-plane bind. +/// +/// On success, writes the socket local address into `out_local_ip` and +/// `out_local_port`. The returned IP string is allocated by this library and +/// must be released with `free_string`. +/// +/// # Safety +/// `out_local_ip` and `out_local_port` must be non-null pointers to writable +/// storage. +/// +/// # Return +/// Returns a non-zero UDP socket handle on success, or `0` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_udp_bind_finish( @@ -679,6 +791,18 @@ pub unsafe extern "C" fn data_plane_udp_bind_finish( unsafe { data_plane_async::data_plane_udp_bind_finish(op_handle, out_local_ip, out_local_port) } } +/// Start an asynchronous UDP data-plane send. +/// +/// The input bytes are copied before this function returns. +/// +/// # Safety +/// `handle` must be a valid UDP socket handle returned by +/// `data_plane_udp_bind_finish`. `dst_ip` must be a non-null pointer to a +/// null-terminated UTF-8 string. If `len` is non-zero, `buf` must be non-null +/// and readable for `len` bytes. +/// +/// # Return +/// Returns a non-zero async operation handle on success, or `0` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_udp_send_to_start( @@ -702,6 +826,14 @@ pub extern "C" fn data_plane_udp_send_to_finish(op_handle: u64) -> c_int { data_plane_async::data_plane_udp_send_to_finish(op_handle) } +/// Start an asynchronous UDP data-plane receive. +/// +/// # Safety +/// `handle` must be a valid UDP socket handle returned by +/// `data_plane_udp_bind_finish`. +/// +/// # Return +/// Returns a non-zero async operation handle on success, or `0` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_udp_recv_from_start( @@ -712,6 +844,19 @@ pub unsafe extern "C" fn data_plane_udp_recv_from_start( unsafe { data_plane_async::data_plane_udp_recv_from_start(handle, max_len, timeout_ms) } } +/// Finish an asynchronous UDP data-plane receive. +/// +/// On success, writes the received buffer into `out_buf` and `out_len`, and +/// the peer address into `out_ip` and `out_port`. The returned buffer is +/// allocated by this library and must be released with `data_plane_free_bytes`; +/// the returned IP string must be released with `free_string`. +/// +/// # Safety +/// `out_buf`, `out_len`, `out_ip`, and `out_port` must be non-null pointers to +/// writable storage. +/// +/// # Return +/// Returns the number of bytes received, or `-1` on failure. #[cfg(feature = "ffi-dataplane")] #[cfg_attr(feature = "c-abi", unsafe(no_mangle))] pub unsafe extern "C" fn data_plane_udp_recv_from_finish( diff --git a/easytier-gui/src-tauri/src/lib.rs b/easytier-gui/src-tauri/src/lib.rs index 066b0c1e..876f201f 100644 --- a/easytier-gui/src-tauri/src/lib.rs +++ b/easytier-gui/src-tauri/src/lib.rs @@ -654,7 +654,8 @@ mod manager { #[derive(Default)] pub(super) enum PersistedConfigSource { User, - Webhook, + #[serde(alias = "webhook")] + Web, #[serde(other)] #[default] Legacy, @@ -664,15 +665,15 @@ mod manager { pub(super) fn from_runtime_source(source: ConfigSource) -> Self { match source { ConfigSource::User => Self::User, - ConfigSource::Webhook => Self::Webhook, + ConfigSource::Web => Self::Web, } } fn merge_persisted(self, incoming: Self) -> Self { match (self, incoming) { // Older runtimes report missing source as `user`. Keep the stronger persisted - // ownership until webhook sync or an explicit user save repairs it. - (Self::Webhook, Self::User) | (Self::Legacy, Self::User) => self, + // ownership until web sync or an explicit user save repairs it. + (Self::Web, Self::User) | (Self::Legacy, Self::User) => self, (_, next) => next, } } @@ -680,13 +681,13 @@ mod manager { fn to_runtime_source(self) -> ConfigSource { match self { Self::User | Self::Legacy => ConfigSource::User, - Self::Webhook => ConfigSource::Webhook, + Self::Web => ConfigSource::Web, } } #[cfg(any(test, target_os = "android"))] - fn is_webhook_like(self) -> bool { - matches!(self, Self::Webhook) + fn is_web_like(self) -> bool { + matches!(self, Self::Web) } } @@ -918,7 +919,7 @@ mod manager { } #[cfg(target_os = "android")] - pub fn get_enabled_instances_with_webhook_like_tun_ids( + pub fn get_enabled_instances_with_web_like_tun_ids( &self, ) -> impl Iterator + '_ { self.storage @@ -926,7 +927,7 @@ mod manager { .iter() .filter(|v| self.storage.enabled_networks.contains(v.key())) .filter(|v| !v.config.no_tun()) - .filter(|v| v.source.is_webhook_like()) + .filter(|v| v.source.is_web_like()) .filter_map(|c| c.config.instance_id().parse::().ok()) } @@ -934,12 +935,11 @@ mod manager { pub(super) async fn disable_instances_with_tun( &self, app: &AppHandle, - webhook_only: bool, + web_only: bool, ) -> Result<(), easytier::rpc_service::remote_client::RemoteClientError> { - let inst_ids: Vec = if webhook_only { - self.get_enabled_instances_with_webhook_like_tun_ids() - .collect() + let inst_ids: Vec = if web_only { + self.get_enabled_instances_with_web_like_tun_ids().collect() } else { self.get_enabled_instances_with_tun_ids().collect() }; @@ -977,7 +977,7 @@ mod manager { .await .map_err(|e| e.to_string())?; } - PersistedConfigSource::Webhook => { + PersistedConfigSource::Web => { self.disable_instances_with_tun(app, true) .await .map_err(|e| e.to_string())?; @@ -1187,26 +1187,46 @@ mod manager { } #[test] - fn persisted_source_merge_keeps_legacy_and_webhook_over_ambiguous_user() { + fn stored_gui_config_deserializes_webhook_source_as_web() { + let stored: StoredGuiConfig = serde_json::from_value(serde_json::json!({ + "config": NetworkConfig::default(), + "source": "webhook", + })) + .unwrap(); + assert_eq!(stored.source, PersistedConfigSource::Web); + } + + #[test] + fn stored_gui_config_defaults_unknown_source_to_legacy() { + let stored: StoredGuiConfig = serde_json::from_value(serde_json::json!({ + "config": NetworkConfig::default(), + "source": "unknown", + })) + .unwrap(); + assert_eq!(stored.source, PersistedConfigSource::Legacy); + } + + #[test] + fn persisted_source_merge_keeps_legacy_and_web_over_ambiguous_user() { assert_eq!( PersistedConfigSource::Legacy.merge_persisted(PersistedConfigSource::User), PersistedConfigSource::Legacy ); assert_eq!( - PersistedConfigSource::Webhook.merge_persisted(PersistedConfigSource::User), - PersistedConfigSource::Webhook + PersistedConfigSource::Web.merge_persisted(PersistedConfigSource::User), + PersistedConfigSource::Web ); assert_eq!( - PersistedConfigSource::Legacy.merge_persisted(PersistedConfigSource::Webhook), - PersistedConfigSource::Webhook + PersistedConfigSource::Legacy.merge_persisted(PersistedConfigSource::Web), + PersistedConfigSource::Web ); } #[test] - fn only_webhook_configs_are_webhook_like() { - assert!(!PersistedConfigSource::Legacy.is_webhook_like()); - assert!(!PersistedConfigSource::User.is_webhook_like()); - assert!(PersistedConfigSource::Webhook.is_webhook_like()); + fn only_web_configs_are_web_like() { + assert!(!PersistedConfigSource::Legacy.is_web_like()); + assert!(!PersistedConfigSource::User.is_web_like()); + assert!(PersistedConfigSource::Web.is_web_like()); } } } diff --git a/easytier-gui/src/composables/backend.ts b/easytier-gui/src/composables/backend.ts index 9b9a2f62..a16835f5 100644 --- a/easytier-gui/src/composables/backend.ts +++ b/easytier-gui/src/composables/backend.ts @@ -1,12 +1,11 @@ import { invoke } from '@tauri-apps/api/core' import { Api, NetworkTypes } from 'easytier-frontend-lib' import { GetNetworkMetasResponse } from 'node_modules/easytier-frontend-lib/dist/modules/api' - +import { type ConfigSource, normalizeConfigSource } from './config_source' type NetworkConfig = NetworkTypes.NetworkConfig type ValidateConfigResponse = Api.ValidateConfigResponse type ListNetworkInstanceIdResponse = Api.ListNetworkInstanceIdResponse -type ConfigSource = 'user' | 'webhook' | 'legacy' interface ServiceOptions { config_dir: string rpc_portal: string @@ -32,14 +31,14 @@ function parseStoredConfigs(raw: string | null): StoredGuiConfig[] { if (entry && typeof entry === 'object' && 'config' in entry) { const { config, source } = entry as { config?: NetworkConfig - source?: ConfigSource + source?: unknown } if (!config) { return [] } return [{ config: NetworkTypes.normalizeNetworkConfig(config), - source: source === 'user' || source === 'webhook' ? source : 'legacy', + source: normalizeConfigSource(source), }] } diff --git a/easytier-gui/src/composables/config_source.ts b/easytier-gui/src/composables/config_source.ts new file mode 100644 index 00000000..476b9455 --- /dev/null +++ b/easytier-gui/src/composables/config_source.ts @@ -0,0 +1,13 @@ +export type ConfigSource = 'user' | 'web' | 'legacy' + +export function normalizeConfigSource(source: unknown): ConfigSource { + if (source === 'user' || source === 'web' || source === 'legacy') { + return source + } + + if (source === 'webhook') { + return 'web' + } + + return 'legacy' +} diff --git a/easytier-gui/src/composables/event.ts b/easytier-gui/src/composables/event.ts index 588be86e..6fb37023 100644 --- a/easytier-gui/src/composables/event.ts +++ b/easytier-gui/src/composables/event.ts @@ -2,10 +2,11 @@ import { Event, listen } from "@tauri-apps/api/event"; import { type } from "@tauri-apps/plugin-os"; import { NetworkTypes } from "easytier-frontend-lib" import { Utils } from "easytier-frontend-lib"; +import { normalizeConfigSource } from './config_source' interface StoredGuiConfig { config: NetworkTypes.NetworkConfig - source?: 'user' | 'webhook' | 'legacy' + source?: unknown } const EVENTS = Object.freeze({ @@ -24,7 +25,7 @@ function onSaveConfigs(event: Event) { 'networkList', JSON.stringify(event.payload.map(({ config, source }) => ({ config: NetworkTypes.normalizeNetworkConfig(config), - source: source ?? 'legacy', + source: normalizeConfigSource(source), }))), ); } diff --git a/easytier-web/src/client_manager/mod.rs b/easytier-web/src/client_manager/mod.rs index 7b0dc4dc..1befe3e5 100644 --- a/easytier-web/src/client_manager/mod.rs +++ b/easytier-web/src/client_manager/mod.rs @@ -20,7 +20,7 @@ use session::{Location, Session}; use storage::{Storage, StorageToken}; use crate::FeatureFlags; -use crate::webhook::SharedWebhookConfig; +use crate::webhook::{ManagedNetworkConfig, SharedWebhookConfig}; use tokio::task::JoinSet; use crate::db::{Db, UserIdInDb, entity::user_running_network_configs}; @@ -146,20 +146,7 @@ impl ClientManager { } pub async fn list_sessions(&self) -> Vec { - let sessions = self - .client_sessions - .iter() - .map(|item| item.value().clone()) - .collect::>(); - - let mut ret: Vec = vec![]; - for s in sessions { - if let Some(t) = s.get_token().await { - ret.push(t); - } - } - - ret + self.storage.list_clients() } pub fn get_session_by_machine_id( @@ -197,6 +184,22 @@ impl ClientManager { self.storage.list_user_clients(user_id) } + pub async fn reconcile_managed_network_configs( + &self, + user_id: UserIdInDb, + machine_id: uuid::Uuid, + desired_configs: Vec, + ) -> anyhow::Result<()> { + session::SessionRpcService::reconcile_web_source_configs( + &self.storage, + user_id, + machine_id, + desired_configs, + ) + .await?; + Ok(()) + } + pub async fn get_heartbeat_requests(&self, client_url: &url::Url) -> Option { let s = self.client_sessions.get(client_url)?.clone(); s.data().read().await.req() diff --git a/easytier-web/src/client_manager/session.rs b/easytier-web/src/client_manager/session.rs index 5fe39451..9636f16e 100644 --- a/easytier-web/src/client_manager/session.rs +++ b/easytier-web/src/client_manager/session.rs @@ -10,9 +10,11 @@ use easytier::{ common::config::ConfigSource, proto::{ api::manage::{ - ConfigSource as RpcConfigSource, NetworkConfig, NetworkMeta, RunNetworkInstanceRequest, + ConfigSource as RpcConfigSource, DeleteNetworkInstanceRequest, + ListNetworkInstanceMetaRequest, NetworkConfig, NetworkMeta, RunNetworkInstanceRequest, WebClientService, WebClientServiceClientFactory, }, + common::Uuid as RpcUuid, rpc_impl::bidirect::BidirectRpcManager, rpc_types::{self, controller::BaseController}, web::{HeartbeatRequest, HeartbeatResponse, WebServerService, WebServerServiceServer}, @@ -27,21 +29,17 @@ use super::storage::{Storage, StorageToken, WeakRefStorage}; use crate::FeatureFlags; use crate::webhook::SharedWebhookConfig; -const LEGACY_NETWORK_CONFIG_SOURCE: &str = "legacy"; - #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum PersistedConfigSource { User, - Webhook, - Legacy, + Web, } impl PersistedConfigSource { fn from_db(source: &str) -> Self { match source { - "webhook" => Self::Webhook, + "web" => Self::Web, "user" => Self::User, - LEGACY_NETWORK_CONFIG_SOURCE => Self::Legacy, _ => Self::User, } } @@ -49,24 +47,23 @@ impl PersistedConfigSource { fn should_update_from_runtime(self, runtime_source: ConfigSource) -> bool { match (self, runtime_source) { // Older clients report missing source as `user`, which is not authoritative enough - // to downgrade an existing webhook-owned or legacy row. - (Self::Webhook | Self::Legacy, ConfigSource::User) => false, + // to downgrade an existing web-owned row. + (Self::Web, ConfigSource::User) => false, _ => self.as_runtime_source() != runtime_source, } } fn as_runtime_source(self) -> ConfigSource { match self { - Self::User | Self::Legacy => ConfigSource::User, - Self::Webhook => ConfigSource::Webhook, + Self::User => ConfigSource::User, + Self::Web => ConfigSource::Web, } } - fn auto_run_rpc_source(self) -> Option { + fn auto_run_rpc_source(self) -> RpcConfigSource { match self { - Self::User => Some(RpcConfigSource::User), - Self::Webhook => Some(RpcConfigSource::Webhook), - Self::Legacy => None, + Self::User => RpcConfigSource::User, + Self::Web => RpcConfigSource::Web, } } } @@ -91,6 +88,7 @@ pub struct SessionData { notifier: broadcast::Sender, req: Option, location: Option, + heartbeat_count: std::sync::atomic::AtomicU32, } impl SessionData { @@ -114,6 +112,7 @@ impl SessionData { notifier: tx, req: None, location, + heartbeat_count: std::sync::atomic::AtomicU32::new(0), } } @@ -164,7 +163,7 @@ impl Drop for SessionData { pub type SharedSessionData = Arc>; #[derive(Clone)] -struct SessionRpcService { +pub(super) struct SessionRpcService { data: SharedSessionData, } @@ -193,7 +192,7 @@ impl SessionRpcService { Ok(serde_json::from_value::(network_config)?) } - async fn reconcile_webhook_source_configs( + pub(super) async fn reconcile_web_source_configs( storage: &Storage, user_id: i32, machine_id: uuid::Uuid, @@ -212,10 +211,10 @@ impl SessionRpcService { .map(|inst_id| (inst_id, PersistedConfigSource::from_db(&cfg.source))) }) .collect::>(); - let existing_webhook_ids = existing_sources + let existing_web_ids = existing_sources .iter() .filter_map(|(inst_id, source)| { - (*source == PersistedConfigSource::Webhook).then_some(*inst_id) + (*source == PersistedConfigSource::Web).then_some(*inst_id) }) .collect::>(); @@ -224,29 +223,18 @@ impl SessionRpcService { for desired in desired_configs { let inst_id = uuid::Uuid::parse_str(&desired.instance_id).with_context(|| { format!( - "invalid desired webhook config instance id: {}", + "invalid desired web config instance id: {}", desired.instance_id ) })?; - match existing_sources.get(&inst_id) { - Some(PersistedConfigSource::User) => { - tracing::warn!( - ?user_id, - ?machine_id, - instance_id = %inst_id, - "skip webhook config because a user-owned config already exists" - ); - continue; - } - Some(PersistedConfigSource::Legacy) => { - tracing::info!( - ?user_id, - ?machine_id, - instance_id = %inst_id, - "adopt legacy config as webhook-owned during reconciliation" - ); - } - _ => {} + if let Some(PersistedConfigSource::User) = existing_sources.get(&inst_id) { + tracing::warn!( + ?user_id, + ?machine_id, + instance_id = %inst_id, + "skip web config because a user-owned config already exists" + ); + continue; } let config = Self::normalize_network_config(desired.network_config, inst_id)?; desired_ids.insert(inst_id); @@ -260,19 +248,15 @@ impl SessionRpcService { (user_id, machine_id), inst_id, config, - ConfigSource::Webhook, + ConfigSource::Web, ) .await .map_err(|e| { - anyhow::anyhow!( - "failed to persist webhook network config {}: {:?}", - inst_id, - e - ) + anyhow::anyhow!("failed to persist web network config {}: {:?}", inst_id, e) })?; } - let stale_ids = existing_webhook_ids + let stale_ids = existing_web_ids .difference(&desired_ids) .copied() .collect::>(); @@ -287,15 +271,42 @@ impl SessionRpcService { Ok(()) } + fn managed_configs_for_revision( + applied_config_revision: Option<&str>, + resp: crate::webhook::ValidateTokenResponse, + ) -> anyhow::Result<(Vec, String)> { + let config_revision = resp.config_revision; + let managed_configs = match resp.managed_network_configs { + Some(configs) => configs, + None if applied_config_revision == Some(config_revision.as_str()) => Vec::new(), + None => { + anyhow::bail!( + "Webhook token validation response omitted managed configs for changed revision {:?}", + config_revision + ); + } + }; + + Ok((managed_configs, config_revision)) + } + async fn handle_heartbeat( &self, req: HeartbeatRequest, ) -> rpc_types::error::Result { - let mut data = self.data.write().await; - - let Ok(storage) = Storage::try_from(data.storage.clone()) else { - tracing::error!("Failed to get storage"); - return Ok(HeartbeatResponse {}); + let (storage, feature_flags, webhook_config, client_url, applied_config_revision) = { + let data = self.data.read().await; + let Ok(storage) = Storage::try_from(data.storage.clone()) else { + tracing::error!("Failed to get storage"); + return Ok(HeartbeatResponse {}); + }; + ( + storage, + data.feature_flags.clone(), + data.webhook_config.clone(), + data.client_url.clone(), + data.applied_config_revision.clone(), + ) }; let machine_id: uuid::Uuid = req.machine_id.map(Into::into).ok_or(anyhow::anyhow!( @@ -303,60 +314,94 @@ impl SessionRpcService { req.machine_id ))?; + // First heartbeat must validate token through webhook; + // afterwards only every 10th heartbeat calls the webhook. + let (should_call_webhook, cached_storage_token) = { + let data = self.data.read().await; + let count = data + .heartbeat_count + .fetch_add(1, std::sync::atomic::Ordering::Relaxed) + + 1; + let is_first = data.req.is_none(); + let should_call = webhook_config.is_enabled() && (is_first || count % 10 == 1); + (should_call, data.storage_token.clone()) + }; + let ( user_id, webhook_source_configs, webhook_config_revision, webhook_validated, binding_version, - ) = if data.webhook_config.is_enabled() { - let webhook_req = crate::webhook::ValidateTokenRequest { - token: req.user_token.clone(), - machine_id: machine_id.to_string(), - public_ip: data.client_url.host_str().map(str::to_string), - hostname: req.hostname.clone(), - version: req.easytier_version.clone(), - os_type: req.device_os.as_ref().map(|info| info.os_type.clone()), - os_version: req.device_os.as_ref().map(|info| info.version.clone()), - os_distribution: req.device_os.as_ref().map(|info| info.distribution.clone()), - web_instance_id: data.webhook_config.web_instance_id.clone(), - web_instance_api_base_url: data.webhook_config.web_instance_api_base_url.clone(), - }; - let resp = data - .webhook_config - .validate_token(&webhook_req) - .await - .map_err(|e| anyhow::anyhow!("Webhook token validation failed: {:?}", e))?; - - if resp.valid { - let user_id = match storage - .db() - .get_user_id_by_token(req.user_token.clone()) - .await - .map_err(|e| anyhow::anyhow!("DB error: {:?}", e))? - { - Some(id) => id, - None => storage - .auto_create_user(&req.user_token) - .await - .with_context(|| { - format!("Failed to auto-create webhook user: {:?}", req.user_token) - })?, + ) = if webhook_config.is_enabled() { + if should_call_webhook { + let webhook_req = crate::webhook::ValidateTokenRequest { + token: req.user_token.clone(), + machine_id: machine_id.to_string(), + public_ip: client_url.host_str().map(str::to_string), + hostname: req.hostname.clone(), + version: req.easytier_version.clone(), + os_type: req.device_os.as_ref().map(|info| info.os_type.clone()), + os_version: req.device_os.as_ref().map(|info| info.version.clone()), + os_distribution: req.device_os.as_ref().map(|info| info.distribution.clone()), + web_instance_id: webhook_config.web_instance_id.clone(), + web_instance_api_base_url: webhook_config.web_instance_api_base_url.clone(), + applied_config_revision: applied_config_revision.clone(), }; - ( - user_id, - resp.managed_network_configs, - resp.config_revision, - true, - Some(resp.binding_version), - ) + let resp = webhook_config + .validate_token(&webhook_req) + .await + .map_err(|e| anyhow::anyhow!("Webhook token validation failed: {:?}", e))?; + + if resp.valid { + let user_id = match storage + .db() + .get_user_id_by_token(req.user_token.clone()) + .await + .map_err(|e| anyhow::anyhow!("DB error: {:?}", e))? + { + Some(id) => id, + None => storage + .auto_create_user(&req.user_token) + .await + .with_context(|| { + format!("Failed to auto-create webhook user: {:?}", req.user_token) + })?, + }; + let binding_version = resp.binding_version; + let (webhook_source_configs, webhook_config_revision) = + Self::managed_configs_for_revision( + applied_config_revision.as_deref(), + resp, + ) + .map_err(rpc_types::error::Error::from)?; + ( + user_id, + webhook_source_configs, + webhook_config_revision, + true, + Some(binding_version), + ) + } else { + return Err(anyhow::anyhow!( + "Webhook rejected token for machine {:?}: {:?}", + machine_id, + req.user_token + ) + .into()); + } } else { - return Err(anyhow::anyhow!( - "Webhook rejected token for machine {:?}: {:?}", - machine_id, - req.user_token - ) - .into()); + let user_id = cached_storage_token + .as_ref() + .ok_or_else(|| { + anyhow::anyhow!("Storage token not found for non-first heartbeat") + })? + .user_id; + let binding_version = { + let data = self.data.read().await; + data.binding_version + }; + (user_id, Vec::new(), String::new(), false, binding_version) } } else { let user_id = match storage @@ -370,7 +415,7 @@ impl SessionRpcService { ) })? { Some(id) => id, - None if data.feature_flags.allow_auto_create_user => storage + None if feature_flags.allow_auto_create_user => storage .auto_create_user(&req.user_token) .await .with_context(|| format!("Failed to auto-create user: {:?}", req.user_token))?, @@ -383,10 +428,10 @@ impl SessionRpcService { (user_id, Vec::new(), String::new(), false, None) }; - if webhook_validated - && data.applied_config_revision.as_deref() != Some(webhook_config_revision.as_str()) - { - Self::reconcile_webhook_source_configs( + let should_reconcile = webhook_validated + && applied_config_revision.as_deref() != Some(webhook_config_revision.as_str()); + if should_reconcile { + Self::reconcile_web_source_configs( &storage, user_id, machine_id, @@ -394,50 +439,68 @@ impl SessionRpcService { ) .await .map_err(rpc_types::error::Error::from)?; - data.applied_config_revision = Some(webhook_config_revision); } - if data.req.replace(req.clone()).is_none() { - assert!(data.storage_token.is_none()); - data.storage_token = Some(StorageToken { - token: req.user_token.clone(), - client_url: data.client_url.clone(), - machine_id, - user_id, - }); - data.binding_version = binding_version; + let mut connect_notification = None; + let (storage_token, notifier) = { + let mut data = self.data.write().await; - // Notify the webhook receiver on the first successful heartbeat. - if data.webhook_config.is_enabled() { - let webhook = data.webhook_config.clone(); - let connect_req = crate::webhook::NodeConnectedRequest { - machine_id: machine_id.to_string(), - token: req.user_token.clone(), - user_id: Some(user_id), - hostname: req.hostname.clone(), - version: req.easytier_version.clone(), - os_type: req.device_os.as_ref().map(|info| info.os_type.clone()), - os_version: req.device_os.as_ref().map(|info| info.version.clone()), - os_distribution: req.device_os.as_ref().map(|info| info.distribution.clone()), - web_instance_id: webhook.web_instance_id.clone(), - binding_version, - }; - tokio::spawn(async move { - webhook.notify_node_connected(&connect_req).await; - }); + if should_reconcile { + data.applied_config_revision = Some(webhook_config_revision); } + + if data.req.replace(req.clone()).is_none() { + assert!(data.storage_token.is_none()); + data.storage_token = Some(StorageToken { + token: req.user_token.clone(), + client_url: data.client_url.clone(), + machine_id, + user_id, + }); + data.binding_version = binding_version; + + if data.webhook_config.is_enabled() { + connect_notification = Some(( + data.webhook_config.clone(), + crate::webhook::NodeConnectedRequest { + machine_id: machine_id.to_string(), + token: req.user_token.clone(), + user_id: Some(user_id), + hostname: req.hostname.clone(), + version: req.easytier_version.clone(), + os_type: req.device_os.as_ref().map(|info| info.os_type.clone()), + os_version: req.device_os.as_ref().map(|info| info.version.clone()), + os_distribution: req + .device_os + .as_ref() + .map(|info| info.distribution.clone()), + web_instance_id: data.webhook_config.web_instance_id.clone(), + binding_version, + }, + )); + } + } + + let Some(storage_token) = data.storage_token.as_ref().cloned() else { + tracing::error!("Heartbeat succeeded before session token was initialized"); + return Ok(HeartbeatResponse {}); + }; + (storage_token, data.notifier.clone()) + }; + + if let Some((webhook, connect_req)) = connect_notification { + tokio::spawn(async move { + webhook.notify_node_connected(&connect_req).await; + }); } let Ok(report_time) = chrono::DateTime::::from_str(&req.report_time) else { tracing::error!("Failed to parse report time: {:?}", req.report_time); return Ok(HeartbeatResponse {}); }; - storage.update_client( - data.storage_token.as_ref().unwrap().clone(), - report_time.timestamp(), - ); + storage.update_client(storage_token, report_time.timestamp()); - let _ = data.notifier.send(req); + let _ = notifier.send(req); Ok(HeartbeatResponse {}) } } @@ -476,7 +539,7 @@ pub struct Session { data: SharedSessionData, - run_network_on_start_task: Option>, + config_reconcile_task: Option>, } impl Debug for Session { @@ -510,7 +573,7 @@ impl Session { Session { rpc_mgr, data, - run_network_on_start_task: None, + config_reconcile_task: None, } } @@ -518,9 +581,9 @@ impl Session { self.rpc_mgr.run_with_tunnel(tunnel); let data = self.data.read().await; - self.run_network_on_start_task + self.config_reconcile_task .replace(AbortOnDropHandle::new(tokio::spawn( - Self::run_network_on_start( + Self::reconcile_network_configs_on_heartbeat( data.heartbeat_waiter(), data.storage.clone(), self.scoped_rpc_client(), @@ -528,22 +591,52 @@ impl Session { ))); } - fn collect_webhook_source_instance_ids( - metas: Vec, - ) -> HashSet { + fn collect_web_source_instance_ids(metas: &[NetworkMeta]) -> HashSet { metas - .into_iter() + .iter() .filter_map(|meta| { - (RpcConfigSource::try_from(meta.source).ok() == Some(RpcConfigSource::Webhook)) + (RpcConfigSource::try_from(meta.source).ok() == Some(RpcConfigSource::Web)) .then(|| { meta.inst_id - .map(|inst_id| Into::::into(inst_id).to_string()) + .as_ref() + .map(|inst_id| Into::::into(*inst_id).to_string()) }) .flatten() }) .collect() } + fn desired_web_source_instance_ids( + local_configs: &[crate::db::entity::user_running_network_configs::Model], + ) -> HashSet { + local_configs + .iter() + .filter(|cfg| cfg.get_runtime_network_config_source() == ConfigSource::Web) + .map(|cfg| cfg.network_instance_id.clone()) + .collect() + } + + fn running_web_source_instance_ids( + running_inst_ids: &HashSet, + db_web_inst_ids: &HashSet, + running_metas: Option<&[NetworkMeta]>, + ) -> HashSet { + match running_metas { + Some(metas) => Self::collect_web_source_instance_ids(metas), + None => running_inst_ids + .intersection(db_web_inst_ids) + .cloned() + .collect(), + } + } + + fn parse_instance_ids(instance_ids: impl Iterator) -> Vec { + instance_ids + .filter_map(|inst_id| uuid::Uuid::parse_str(&inst_id).ok()) + .map(Into::into) + .collect() + } + async fn sync_running_config_sources( db: &crate::db::Db, user_id: i32, @@ -597,65 +690,25 @@ impl Session { Ok(()) } - async fn repair_legacy_running_config_sources( - db: &crate::db::Db, - user_id: i32, - machine_id: uuid::Uuid, - local_configs: &[crate::db::entity::user_running_network_configs::Model], - ) -> anyhow::Result { - let legacy_configs = local_configs - .iter() - .filter(|cfg| { - PersistedConfigSource::from_db(&cfg.source) == PersistedConfigSource::Legacy - }) - .collect::>(); - - if legacy_configs.is_empty() { - return Ok(false); - } - - for local_cfg in legacy_configs { - let inst_id = - uuid::Uuid::parse_str(&local_cfg.network_instance_id).with_context(|| { - format!( - "failed to parse legacy network config instance id {}", - local_cfg.network_instance_id - ) - })?; - - db.insert_or_update_user_network_config( - (user_id, machine_id), - inst_id, - local_cfg.get_network_config().map_err(|e| { - anyhow::anyhow!( - "failed to decode legacy network config {}: {:?}", - inst_id, - e - ) - })?, - ConfigSource::User, - ) - .await - .map_err(|e| { - anyhow::anyhow!( - "failed to repair legacy network config source {}: {:?}", - inst_id, - e - ) - })?; - } - - Ok(true) - } - - async fn run_network_on_start( + async fn reconcile_network_configs_on_heartbeat( mut heartbeat_waiter: broadcast::Receiver, storage: WeakRefStorage, rpc_client: SessionRpcClient, ) { - let mut cleaned_webhook_source_instances = false; - let mut last_desired_webhook_inst_ids: Option> = None; + // This is a per-session background task. It starts when the RPC session is + // created, then reconciles after each heartbeat reports the client's runtime + // instances. It is deliberately best-effort: a failed round is retried by a + // later heartbeat instead of blocking heartbeat handling itself. + let mut cleaned_web_source_instances = false; + // This is only an in-memory guard for RPC cleanup, not a second source of + // truth. The DB still owns desired state; the cache lets us avoid listing + // and deleting runtime instances on every heartbeat when desired web-owned + // configs have not changed. + let mut last_desired_web_inst_ids: Option> = None; loop { + // Drop any heartbeat backlog accumulated while the previous reconcile + // round was doing DB/RPC IO. The newest heartbeat has the freshest + // runtime instance list, which is all this task needs. heartbeat_waiter = heartbeat_waiter.resubscribe(); let req = heartbeat_waiter.recv().await; if req.is_err() { @@ -718,12 +771,10 @@ impl Session { rpc_client .list_network_instance_meta( BaseController::default(), - easytier::proto::api::manage::ListNetworkInstanceMetaRequest { - inst_ids: running_inst_ids - .iter() - .filter_map(|inst_id| uuid::Uuid::parse_str(inst_id).ok()) - .map(Into::into) - .collect(), + ListNetworkInstanceMetaRequest { + inst_ids: Self::parse_instance_ids( + running_inst_ids.iter().cloned(), + ), }, ) .await @@ -781,105 +832,50 @@ impl Session { None }; - match Self::repair_legacy_running_config_sources( - &storage.db, - user_id, - machine_id.into(), - &local_configs, - ) - .await - { - Ok(true) => { - local_configs = match storage - .db - .list_network_configs( - (user_id, machine_id.into()), - ListNetworkProps::EnabledOnly, - ) - .await - { - Ok(configs) => configs, - Err(e) => { - tracing::error!( - "Failed to reload network configs after legacy source repair, error: {:?}", - e - ); - return; - } - }; - } - Ok(false) => {} - Err(e) => { - tracing::warn!( - ?user_id, - ?machine_id, - %e, - "Failed to repair legacy running network config sources" - ); - } - } + let should_be_alive_web_inst_ids = + Self::desired_web_source_instance_ids(&local_configs); + let desired_changed = last_desired_web_inst_ids + .as_ref() + .is_none_or(|last| last != &should_be_alive_web_inst_ids); let mut has_failed = false; - let should_be_alive_webhook_inst_ids = local_configs - .iter() - .filter(|cfg| cfg.get_runtime_network_config_source() == ConfigSource::Webhook) - .map(|cfg| cfg.network_instance_id.clone()) - .collect::>(); - let desired_changed = last_desired_webhook_inst_ids - .as_ref() - .is_none_or(|last| last != &should_be_alive_webhook_inst_ids); - - if !cleaned_webhook_source_instances || desired_changed { - let db_webhook_inst_ids = match storage + if !cleaned_web_source_instances || desired_changed { + let db_web_inst_ids = match storage .db .list_network_configs((user_id, machine_id.into()), ListNetworkProps::All) .await { - Ok(configs) => configs - .iter() - .filter(|cfg| { - cfg.get_runtime_network_config_source() == ConfigSource::Webhook - }) - .map(|cfg| cfg.network_instance_id.clone()) - .collect::>(), + Ok(configs) => Self::desired_web_source_instance_ids(&configs), Err(e) => { tracing::error!("Failed to list all network configs, error: {:?}", e); return; } }; - let running_webhook_inst_ids = if let Some(metas) = running_metas.as_ref() { - Self::collect_webhook_source_instance_ids(metas.clone()) - } else { - running_inst_ids - .intersection(&db_webhook_inst_ids) - .cloned() - .collect() - }; + let running_web_inst_ids = Self::running_web_source_instance_ids( + &running_inst_ids, + &db_web_inst_ids, + running_metas.as_deref(), + ); - let should_delete_inst_ids = running_webhook_inst_ids - .difference(&should_be_alive_webhook_inst_ids) - .cloned() - .collect::>(); - - let should_delete_ids = should_delete_inst_ids - .iter() - .filter_map(|inst_id| uuid::Uuid::parse_str(inst_id).ok()) - .map(Into::into) - .collect::>(); + let should_delete_ids = Self::parse_instance_ids( + running_web_inst_ids + .difference(&should_be_alive_web_inst_ids) + .cloned(), + ); if !should_delete_ids.is_empty() { let ret = rpc_client .delete_network_instance( BaseController::default(), - easytier::proto::api::manage::DeleteNetworkInstanceRequest { + DeleteNetworkInstanceRequest { inst_ids: should_delete_ids, }, ) .await; tracing::info!( ?user_id, - "Clean stale webhook-source network instances on start: {:?}, user_token: {:?}", + "Clean stale web-source network instances on heartbeat: {:?}, user_token: {:?}", ret, req.user_token ); @@ -887,33 +883,39 @@ impl Session { } if !has_failed { - cleaned_webhook_source_instances = true; - last_desired_webhook_inst_ids = Some(should_be_alive_webhook_inst_ids.clone()); + cleaned_web_source_instances = true; + last_desired_web_inst_ids = Some(should_be_alive_web_inst_ids.clone()); } } + // After stale web-owned instances are removed, start every enabled + // config that the latest heartbeat did not report as running. for c in local_configs { if running_inst_ids.contains(&c.network_instance_id) { continue; } - let Some(source) = PersistedConfigSource::from_db(&c.source).auto_run_rpc_source() - else { - tracing::warn!( - ?user_id, - ?machine_id, - instance_id = %c.network_instance_id, - "skip auto-run for legacy config until source ownership is repaired" - ); - continue; + let source = PersistedConfigSource::from_db(&c.source).auto_run_rpc_source(); + let network_config = match serde_json::from_str::(&c.network_config) + { + Ok(cfg) => cfg, + Err(e) => { + tracing::error!( + ?user_id, + ?machine_id, + instance_id = %c.network_instance_id, + "Failed to deserialize network config, skipping: {:?}", + e + ); + has_failed = true; + continue; + } }; let ret = rpc_client .run_network_instance( BaseController::default(), RunNetworkInstanceRequest { inst_id: Some(c.network_instance_id.clone().into()), - config: Some( - serde_json::from_str::(&c.network_config).unwrap(), - ), + config: Some(network_config), overwrite: false, source: source as i32, }, @@ -930,7 +932,7 @@ impl Session { } if !has_failed { - last_desired_webhook_inst_ids = Some(should_be_alive_webhook_inst_ids); + last_desired_web_inst_ids = Some(should_be_alive_web_inst_ids); } } } @@ -953,6 +955,15 @@ impl Session { .scoped_client::(1, 1, "".to_string()) } + pub fn scoped_client_with_domain( + &self, + domain_name: String, + ) -> F::ClientImpl { + self.rpc_mgr + .rpc_client() + .scoped_client::(1, 1, domain_name) + } + pub fn scoped_rpc_client(&self) -> SessionRpcClient { self.scoped_client::>() } @@ -972,20 +983,14 @@ mod tests { common::config::ConfigSource, rpc_service::remote_client::{ListNetworkProps, PersistentConfig as _, Storage as _}, }; - use sea_orm::{ActiveModelTrait, Set}; use serde_json::json; use super::{super::storage::Storage, *}; #[tokio::test] - async fn reconcile_webhook_source_configs_upserts_and_deletes_exact_set() { + async fn reconcile_web_source_configs_upserts_and_deletes_exact_set() { let storage = Storage::new(crate::db::Db::memory_db().await); - let user_id = storage - .db() - .auto_create_user("webhook-user") - .await - .unwrap() - .id; + let user_id = storage.db().auto_create_user("web-user").await.unwrap().id; let machine_id = uuid::Uuid::new_v4(); let keep_id = uuid::Uuid::new_v4(); let stale_id = uuid::Uuid::new_v4(); @@ -1000,7 +1005,7 @@ mod tests { network_name: Some("old-name".to_string()), ..Default::default() }, - ConfigSource::Webhook, + ConfigSource::Web, ) .await .unwrap(); @@ -1013,12 +1018,12 @@ mod tests { network_name: Some("stale".to_string()), ..Default::default() }, - ConfigSource::Webhook, + ConfigSource::Web, ) .await .unwrap(); - SessionRpcService::reconcile_webhook_source_configs( + SessionRpcService::reconcile_web_source_configs( &storage, user_id, machine_id, @@ -1069,24 +1074,21 @@ mod tests { updated_keep_config.network_name.as_deref(), Some("updated-name") ); - assert_eq!( - updated_keep.get_network_config_source(), - ConfigSource::Webhook - ); + assert_eq!(updated_keep.get_network_config_source(), ConfigSource::Web); } #[tokio::test] - async fn reconcile_webhook_source_configs_keep_user_owned_configs() { + async fn reconcile_web_source_configs_keep_user_owned_configs() { let storage = Storage::new(crate::db::Db::memory_db().await); let user_id = storage .db() - .auto_create_user("webhook-user-keep-user") + .auto_create_user("web-user-keep-user") .await .unwrap() .id; let machine_id = uuid::Uuid::new_v4(); let user_owned_id = uuid::Uuid::new_v4(); - let webhook_owned_id = uuid::Uuid::new_v4(); + let web_owned_id = uuid::Uuid::new_v4(); storage .db() @@ -1105,17 +1107,17 @@ mod tests { .db() .insert_or_update_user_network_config( (user_id, machine_id), - webhook_owned_id, + web_owned_id, NetworkConfig { - network_name: Some("webhook-owned".to_string()), + network_name: Some("web-owned".to_string()), ..Default::default() }, - ConfigSource::Webhook, + ConfigSource::Web, ) .await .unwrap(); - SessionRpcService::reconcile_webhook_source_configs( + SessionRpcService::reconcile_web_source_configs( &storage, user_id, machine_id, @@ -1123,7 +1125,7 @@ mod tests { instance_id: user_owned_id.to_string(), network_config: json!({ "instance_id": user_owned_id.to_string(), - "network_name": "webhook-tries-to-take-over" + "network_name": "web-tries-to-take-over" }), }], ) @@ -1141,100 +1143,68 @@ mod tests { serde_json::from_str(&user_owned.network_config).unwrap(); assert_eq!(user_owned_cfg.network_name.as_deref(), Some("user-owned")); - let webhook_owned = storage + let web_owned = storage .db() - .get_network_config((user_id, machine_id), &webhook_owned_id.to_string()) + .get_network_config((user_id, machine_id), &web_owned_id.to_string()) .await .unwrap(); - assert!(webhook_owned.is_none()); + assert!(web_owned.is_none()); } - #[tokio::test] - async fn reconcile_webhook_source_configs_adopts_legacy_rows_for_webhook() { - let storage = Storage::new(crate::db::Db::memory_db().await); - let user_id = storage - .db() - .auto_create_user("webhook-user-legacy") - .await - .unwrap() - .id; - let machine_id = uuid::Uuid::new_v4(); - let legacy_match_id = uuid::Uuid::new_v4(); - let legacy_user_id = uuid::Uuid::new_v4(); + #[test] + fn validate_token_request_includes_applied_config_revision() { + let req = crate::webhook::ValidateTokenRequest { + token: "token".to_string(), + machine_id: "machine".to_string(), + public_ip: Some("127.0.0.1".to_string()), + hostname: "host".to_string(), + version: "1.0.0".to_string(), + os_type: None, + os_version: None, + os_distribution: None, + web_instance_id: Some("web-1".to_string()), + web_instance_api_base_url: Some("http://console".to_string()), + applied_config_revision: Some("rev-1".to_string()), + }; - crate::db::entity::user_running_network_configs::ActiveModel { - user_id: Set(user_id), - device_id: Set(machine_id.to_string()), - network_instance_id: Set(legacy_match_id.to_string()), - network_config: Set(serde_json::to_string(&NetworkConfig { - network_name: Some("legacy-webhook".to_string()), - ..Default::default() - }) - .unwrap()), - source: Set(LEGACY_NETWORK_CONFIG_SOURCE.to_string()), - disabled: Set(false), - create_time: Set(sqlx::types::chrono::Local::now().fixed_offset()), - update_time: Set(sqlx::types::chrono::Local::now().fixed_offset()), - ..Default::default() - } - .insert(storage.db().orm_db()) - .await - .unwrap(); - - crate::db::entity::user_running_network_configs::ActiveModel { - user_id: Set(user_id), - device_id: Set(machine_id.to_string()), - network_instance_id: Set(legacy_user_id.to_string()), - network_config: Set(serde_json::to_string(&NetworkConfig { - network_name: Some("legacy-user".to_string()), - ..Default::default() - }) - .unwrap()), - source: Set(LEGACY_NETWORK_CONFIG_SOURCE.to_string()), - disabled: Set(false), - create_time: Set(sqlx::types::chrono::Local::now().fixed_offset()), - update_time: Set(sqlx::types::chrono::Local::now().fixed_offset()), - ..Default::default() - } - .insert(storage.db().orm_db()) - .await - .unwrap(); - - SessionRpcService::reconcile_webhook_source_configs( - &storage, - user_id, - machine_id, - vec![crate::webhook::ManagedNetworkConfig { - instance_id: legacy_match_id.to_string(), - network_config: json!({ - "instance_id": legacy_match_id.to_string(), - "network_name": "managed-by-webhook" - }), - }], - ) - .await - .unwrap(); - - let adopted = storage - .db() - .get_network_config((user_id, machine_id), &legacy_match_id.to_string()) - .await - .unwrap() - .unwrap(); - assert_eq!(adopted.source, ConfigSource::Webhook.as_str()); - let adopted_cfg: NetworkConfig = serde_json::from_str(&adopted.network_config).unwrap(); + let value = serde_json::to_value(req).unwrap(); assert_eq!( - adopted_cfg.network_name.as_deref(), - Some("managed-by-webhook") + value + .get("applied_config_revision") + .and_then(|v| v.as_str()), + Some("rev-1") ); + } - let untouched_legacy = storage - .db() - .get_network_config((user_id, machine_id), &legacy_user_id.to_string()) - .await - .unwrap() - .unwrap(); - assert_eq!(untouched_legacy.source, LEGACY_NETWORK_CONFIG_SOURCE); + #[test] + fn validate_token_response_without_configs_reuses_same_revision() { + let resp = crate::webhook::ValidateTokenResponse { + valid: true, + pre_approved: true, + binding_version: 1, + managed_network_configs: None, + config_revision: "rev-1".to_string(), + }; + + let (configs, revision) = + SessionRpcService::managed_configs_for_revision(Some("rev-1"), resp).unwrap(); + assert!(configs.is_empty()); + assert_eq!(revision, "rev-1"); + } + + #[test] + fn validate_token_response_without_configs_rejects_changed_revision() { + let resp = crate::webhook::ValidateTokenResponse { + valid: true, + pre_approved: true, + binding_version: 1, + managed_network_configs: None, + config_revision: "rev-2".to_string(), + }; + + let err = SessionRpcService::managed_configs_for_revision(Some("rev-1"), resp) + .expect_err("omitted configs with a changed revision must fail"); + assert!(err.to_string().contains("omitted managed configs")); } #[tokio::test] @@ -1242,7 +1212,7 @@ mod tests { let storage = Storage::new(crate::db::Db::memory_db().await); let user_id = storage .db() - .auto_create_user("webhook-user-sync-source") + .auto_create_user("web-user-sync-source") .await .unwrap() .id; @@ -1255,10 +1225,10 @@ mod tests { (user_id, machine_id), inst_id, NetworkConfig { - network_name: Some("webhook-owned".to_string()), + network_name: Some("web-owned".to_string()), ..Default::default() }, - ConfigSource::Webhook, + ConfigSource::Web, ) .await .unwrap(); @@ -1288,134 +1258,18 @@ mod tests { .await .unwrap() .unwrap(); - assert_eq!(updated.get_network_config_source(), ConfigSource::Webhook); - } - - #[tokio::test] - async fn sync_running_config_sources_keeps_legacy_rows_when_runtime_source_is_user() { - let storage = Storage::new(crate::db::Db::memory_db().await); - let user_id = storage - .db() - .auto_create_user("webhook-user-sync-legacy") - .await - .unwrap() - .id; - let machine_id = uuid::Uuid::new_v4(); - let inst_id = uuid::Uuid::new_v4(); - - crate::db::entity::user_running_network_configs::ActiveModel { - user_id: Set(user_id), - device_id: Set(machine_id.to_string()), - network_instance_id: Set(inst_id.to_string()), - network_config: Set(serde_json::to_string(&NetworkConfig { - network_name: Some("legacy".to_string()), - ..Default::default() - }) - .unwrap()), - source: Set(LEGACY_NETWORK_CONFIG_SOURCE.to_string()), - disabled: Set(false), - create_time: Set(sqlx::types::chrono::Local::now().fixed_offset()), - update_time: Set(sqlx::types::chrono::Local::now().fixed_offset()), - ..Default::default() - } - .insert(storage.db().orm_db()) - .await - .unwrap(); - - let local_configs = storage - .db() - .list_network_configs((user_id, machine_id), ListNetworkProps::EnabledOnly) - .await - .unwrap(); - Session::sync_running_config_sources( - storage.db(), - user_id, - machine_id, - &local_configs, - &[easytier::proto::api::manage::NetworkMeta { - inst_id: Some(inst_id.into()), - source: RpcConfigSource::User as i32, - ..Default::default() - }], - ) - .await - .unwrap(); - - let updated = storage - .db() - .get_network_config((user_id, machine_id), &inst_id.to_string()) - .await - .unwrap() - .unwrap(); - assert_eq!(updated.source, LEGACY_NETWORK_CONFIG_SOURCE); - } - - #[tokio::test] - async fn repair_legacy_running_config_sources_promotes_remaining_legacy_rows_to_user() { - let storage = Storage::new(crate::db::Db::memory_db().await); - let user_id = storage - .db() - .auto_create_user("webhook-user-repair-legacy") - .await - .unwrap() - .id; - let machine_id = uuid::Uuid::new_v4(); - let inst_id = uuid::Uuid::new_v4(); - - crate::db::entity::user_running_network_configs::ActiveModel { - user_id: Set(user_id), - device_id: Set(machine_id.to_string()), - network_instance_id: Set(inst_id.to_string()), - network_config: Set(serde_json::to_string(&NetworkConfig { - network_name: Some("legacy".to_string()), - ..Default::default() - }) - .unwrap()), - source: Set(LEGACY_NETWORK_CONFIG_SOURCE.to_string()), - disabled: Set(false), - create_time: Set(sqlx::types::chrono::Local::now().fixed_offset()), - update_time: Set(sqlx::types::chrono::Local::now().fixed_offset()), - ..Default::default() - } - .insert(storage.db().orm_db()) - .await - .unwrap(); - - let local_configs = storage - .db() - .list_network_configs((user_id, machine_id), ListNetworkProps::EnabledOnly) - .await - .unwrap(); - assert!( - Session::repair_legacy_running_config_sources( - storage.db(), - user_id, - machine_id, - &local_configs, - ) - .await - .unwrap() - ); - - let updated = storage - .db() - .get_network_config((user_id, machine_id), &inst_id.to_string()) - .await - .unwrap() - .unwrap(); - assert_eq!(updated.source, ConfigSource::User.as_str()); + assert_eq!(updated.get_network_config_source(), ConfigSource::Web); } #[test] - fn legacy_configs_are_not_auto_run_until_repaired() { - assert_eq!(PersistedConfigSource::Legacy.auto_run_rpc_source(), None); + fn persisted_sources_map_to_rpc_sources() { assert_eq!( - PersistedConfigSource::Webhook.auto_run_rpc_source(), - Some(RpcConfigSource::Webhook) + PersistedConfigSource::Web.auto_run_rpc_source(), + RpcConfigSource::Web ); assert_eq!( PersistedConfigSource::User.auto_run_rpc_source(), - Some(RpcConfigSource::User) + RpcConfigSource::User ); } } diff --git a/easytier-web/src/client_manager/storage.rs b/easytier-web/src/client_manager/storage.rs index 73291986..9a8ba0d5 100644 --- a/easytier-web/src/client_manager/storage.rs +++ b/easytier-web/src/client_manager/storage.rs @@ -114,6 +114,20 @@ impl Storage { .unwrap_or_default() } + pub fn list_clients(&self) -> Vec { + self.0 + .user_clients_map + .iter() + .flat_map(|user_clients| { + user_clients + .value() + .iter() + .map(|info| info.value().storage_token.clone()) + .collect::>() + }) + .collect() + } + pub fn db(&self) -> &Db { &self.0.db } @@ -174,4 +188,25 @@ mod tests { assert_eq!(storage.get_client_url_by_machine_id(2, &machine_id), None); } + + #[tokio::test] + async fn list_clients_returns_current_storage_tokens() { + let storage = Storage::new(Db::memory_db().await); + let user1_token = make_storage_token(1, uuid::Uuid::new_v4(), "tcp://127.0.0.1:1001"); + let user2_token = make_storage_token(2, uuid::Uuid::new_v4(), "tcp://127.0.0.1:1002"); + + storage.update_client(user1_token.clone(), 10); + storage.update_client(user2_token.clone(), 20); + + let tokens = storage.list_clients(); + assert_eq!(tokens.len(), 2); + assert!(tokens.iter().any(|token| token.token == user1_token.token)); + assert!(tokens.iter().any(|token| token.token == user2_token.token)); + + storage.remove_client(&user1_token); + + let tokens = storage.list_clients(); + assert_eq!(tokens.len(), 1); + assert_eq!(tokens[0].token, user2_token.token); + } } diff --git a/easytier-web/src/db/mod.rs b/easytier-web/src/db/mod.rs index 66b0e567..33cb3c8e 100644 --- a/easytier-web/src/db/mod.rs +++ b/easytier-web/src/db/mod.rs @@ -331,7 +331,7 @@ mod tests { (user_id, device_id), inst_id, network_config, - ConfigSource::Webhook, + ConfigSource::Web, ) .await .unwrap(); @@ -344,10 +344,10 @@ mod tests { .unwrap(); println!("device: {}, {:?}", device_id, result2); assert_eq!(result2.network_config, network_config_json); - assert_eq!(result2.get_network_config_source(), ConfigSource::Webhook); + assert_eq!(result2.get_network_config_source(), ConfigSource::Web); assert_eq!( result2.get_runtime_network_config_source(), - ConfigSource::Webhook + ConfigSource::Web ); assert_eq!(result.create_time, result2.create_time); @@ -373,7 +373,7 @@ mod tests { } #[tokio::test] - async fn test_legacy_network_config_defaults_to_user_runtime_source() { + async fn test_unknown_network_config_source_defaults_to_user_runtime_source() { let db = Db::memory_db().await; let user_id = 1; let inst_id = uuid::Uuid::new_v4(); @@ -384,11 +384,11 @@ mod tests { device_id: Set(device_id.to_string()), network_instance_id: Set(inst_id.to_string()), network_config: Set(serde_json::to_string(&NetworkConfig { - network_name: Some("legacy".to_string()), + network_name: Some("unknown-source".to_string()), ..Default::default() }) .unwrap()), - source: Set("legacy".to_string()), + source: Set("unknown".to_string()), disabled: Set(false), create_time: Set(sqlx::types::chrono::Local::now().fixed_offset()), update_time: Set(sqlx::types::chrono::Local::now().fixed_offset()), diff --git a/easytier-web/src/migrator/m20260421_000003_add_network_config_source.rs b/easytier-web/src/migrator/m20260421_000003_add_network_config_source.rs index e2ee1719..cd11b441 100644 --- a/easytier-web/src/migrator/m20260421_000003_add_network_config_source.rs +++ b/easytier-web/src/migrator/m20260421_000003_add_network_config_source.rs @@ -48,7 +48,7 @@ impl MigrationTrait for Migration { device_id, network_instance_id, network_config, - 'legacy', + 'user', disabled, create_time, update_time diff --git a/easytier-web/src/migrator/m20260514_000004_rename_web_config_source.rs b/easytier-web/src/migrator/m20260514_000004_rename_web_config_source.rs new file mode 100644 index 00000000..7b3c7c21 --- /dev/null +++ b/easytier-web/src/migrator/m20260514_000004_rename_web_config_source.rs @@ -0,0 +1,42 @@ +use sea_orm_migration::prelude::*; + +pub struct Migration; + +impl MigrationName for Migration { + fn name(&self) -> &str { + "m20260514_000004_rename_web_config_source" + } +} + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let db = manager.get_connection(); + db.execute_unprepared( + r#" + UPDATE user_running_network_configs + SET source = 'web' + WHERE source = 'webhook'; + + UPDATE user_running_network_configs + SET source = 'user' + WHERE source = 'legacy'; + "#, + ) + .await?; + Ok(()) + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + let db = manager.get_connection(); + db.execute_unprepared( + r#" + UPDATE user_running_network_configs + SET source = 'webhook' + WHERE source = 'web'; + "#, + ) + .await?; + Ok(()) + } +} diff --git a/easytier-web/src/migrator/mod.rs b/easytier-web/src/migrator/mod.rs index 03be5eea..6eea9f11 100644 --- a/easytier-web/src/migrator/mod.rs +++ b/easytier-web/src/migrator/mod.rs @@ -3,6 +3,7 @@ use sea_orm_migration::prelude::*; mod m20241029_000001_init; mod m20260403_000002_scope_network_config_unique; mod m20260421_000003_add_network_config_source; +mod m20260514_000004_rename_web_config_source; pub struct Migrator; @@ -13,6 +14,7 @@ impl MigratorTrait for Migrator { Box::new(m20241029_000001_init::Migration), Box::new(m20260403_000002_scope_network_config_unique::Migration), Box::new(m20260421_000003_add_network_config_source::Migration), + Box::new(m20260514_000004_rename_web_config_source::Migration), ] } } diff --git a/easytier-web/src/restful/network.rs b/easytier-web/src/restful/network.rs index b47f5167..46c65aab 100644 --- a/easytier-web/src/restful/network.rs +++ b/easytier-web/src/restful/network.rs @@ -3,6 +3,7 @@ use axum::http::StatusCode; use axum::routing::{delete, post}; use axum::{Json, Router, extract::State, routing::get}; use axum_login::AuthUser; +use easytier::common::config::ConfigSource as RuntimeConfigSource; use easytier::launcher::NetworkConfig; use easytier::proto::common::Void; use easytier::proto::{api::manage::*, web::*}; @@ -60,6 +61,7 @@ struct SaveNetworkJsonReq { struct RunNetworkJsonReq { config: NetworkConfig, save: bool, + source: Option, } #[derive(Debug, serde::Deserialize, serde::Serialize)] @@ -82,6 +84,17 @@ struct RemoveNetworkJsonReq { inst_ids: Vec, } +#[derive(Debug, serde::Deserialize, serde::Serialize)] +struct ManagedNetworkConfigJson { + instance_id: uuid::Uuid, + network_config: serde_json::Value, +} + +#[derive(Debug, serde::Deserialize, serde::Serialize)] +struct ReconcileManagedNetworkConfigsJsonReq { + managed_network_configs: Vec, +} + #[derive(Debug, serde::Deserialize, serde::Serialize)] struct ListMachineItem { client_url: Option, @@ -130,10 +143,11 @@ impl NetworkApi { Json(payload): Json, ) -> Result, HttpHandleError> { client_mgr - .handle_run_network_instance( + .handle_run_network_instance_with_source( (Self::get_user_id(&auth_session)?, machine_id), payload.config, payload.save, + RuntimeConfigSource::Web, ) .await .map_err(convert_error)?; @@ -274,10 +288,11 @@ impl NetworkApi { )); } client_mgr - .handle_save_network_config( + .handle_save_network_config_with_source( (Self::get_user_id(&auth_session)?, machine_id), inst_id, payload.config, + RuntimeConfigSource::Web, ) .await .map_err(convert_error) @@ -302,8 +317,17 @@ impl NetworkApi { Path((user_id, machine_id)): Path<(UserIdInDb, uuid::Uuid)>, Json(payload): Json, ) -> Result, HttpHandleError> { + let source = payload + .source + .and_then(RuntimeConfigSource::from_rpc) + .unwrap_or(RuntimeConfigSource::Web); client_mgr - .handle_run_network_instance((user_id, machine_id), payload.config, payload.save) + .handle_run_network_instance_with_source( + (user_id, machine_id), + payload.config, + payload.save, + source, + ) .await .map_err(convert_error)?; Ok(Void::default().into()) @@ -319,6 +343,31 @@ impl NetworkApi { .map_err(convert_error) } + async fn handle_reconcile_managed_network_configs_internal( + State(client_mgr): AppState, + Path((user_id, machine_id)): Path<(UserIdInDb, uuid::Uuid)>, + Json(payload): Json, + ) -> Result, HttpHandleError> { + let desired = payload + .managed_network_configs + .into_iter() + .map(|item| crate::webhook::ManagedNetworkConfig { + instance_id: item.instance_id.to_string(), + network_config: item.network_config, + }) + .collect(); + client_mgr + .reconcile_managed_network_configs(user_id, machine_id, desired) + .await + .map_err(|err| { + ( + StatusCode::INTERNAL_SERVER_ERROR, + other_error(err.to_string()).into(), + ) + })?; + Ok(Void::default().into()) + } + async fn handle_list_network_instance_ids_internal( State(client_mgr): AppState, Path((user_id, machine_id)): Path<(UserIdInDb, uuid::Uuid)>, @@ -347,6 +396,7 @@ impl NetworkApi { .route( "/api/internal/users/:user-id/machines/:machine-id/networks", post(Self::handle_run_network_instance_internal) + .put(Self::handle_reconcile_managed_network_configs_internal) .get(Self::handle_list_network_instance_ids_internal), ) .route( diff --git a/easytier-web/src/restful/rpc.rs b/easytier-web/src/restful/rpc.rs index e938a87f..f8b13cbc 100644 --- a/easytier-web/src/restful/rpc.rs +++ b/easytier-web/src/restful/rpc.rs @@ -16,6 +16,7 @@ pub struct ProxyRpcRequest { pub service_name: String, pub method_name: String, pub payload: serde_json::Value, + pub scope: Option, } macro_rules! match_service { @@ -35,6 +36,7 @@ async fn handle_proxy_rpc_by_session( service_name, method_name, payload, + scope, } = req; let resp = match service_name.as_str() { @@ -74,12 +76,20 @@ async fn handle_proxy_rpc_by_session( payload, session ), - "api.instance.TcpProxyRpcService" => match_service!( - easytier::proto::api::instance::TcpProxyRpcClientFactory, - method_name, - payload, - session - ), + "api.instance.TcpProxyRpcService" => { + let client = if let Some(ref domain) = scope { + session.scoped_client_with_domain::< + easytier::proto::api::instance::TcpProxyRpcClientFactory, + >(domain.clone()) + } else { + session.scoped_client::< + easytier::proto::api::instance::TcpProxyRpcClientFactory, + >() + }; + client + .json_call_method(BaseController::default(), &method_name, payload) + .await + } "api.instance.AclManageRpcService" => match_service!( easytier::proto::api::instance::AclManageRpcClientFactory, method_name, diff --git a/easytier-web/src/webhook.rs b/easytier-web/src/webhook.rs index be252b11..ce65b10f 100644 --- a/easytier-web/src/webhook.rs +++ b/easytier-web/src/webhook.rs @@ -57,6 +57,8 @@ pub struct ValidateTokenRequest { pub os_distribution: Option, pub web_instance_id: Option, pub web_instance_api_base_url: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub applied_config_revision: Option, } #[derive(Debug, Deserialize)] @@ -66,7 +68,8 @@ pub struct ValidateTokenResponse { pub pre_approved: bool, #[serde(default)] pub binding_version: u64, - pub managed_network_configs: Vec, + #[serde(default)] + pub managed_network_configs: Option>, pub config_revision: String, } @@ -184,3 +187,17 @@ impl WebhookConfig { } pub type SharedWebhookConfig = Arc; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn validate_token_response_allows_missing_managed_configs() { + let resp: ValidateTokenResponse = + serde_json::from_str(r#"{"valid":true,"config_revision":"rev-1"}"#).unwrap(); + assert!(resp.valid); + assert_eq!(resp.config_revision, "rev-1"); + assert!(resp.managed_network_configs.is_none()); + } +} diff --git a/easytier/src/common/config.rs b/easytier/src/common/config.rs index e997a2be..57cb8f0c 100644 --- a/easytier/src/common/config.rs +++ b/easytier/src/common/config.rs @@ -278,20 +278,20 @@ pub struct NetworkIdentity { pub enum ConfigSource { #[default] User, - Webhook, + Web, } impl ConfigSource { pub fn as_str(self) -> &'static str { match self { Self::User => "user", - Self::Webhook => "webhook", + Self::Web => "web", } } pub fn from_rpc(source: i32) -> Option { match RpcConfigSource::try_from(source).ok() { - Some(RpcConfigSource::Webhook) => Some(Self::Webhook), + Some(RpcConfigSource::Web) => Some(Self::Web), Some(RpcConfigSource::User) => Some(Self::User), _ => None, } @@ -300,7 +300,7 @@ impl ConfigSource { pub fn to_rpc(self) -> i32 { match self { Self::User => RpcConfigSource::User as i32, - Self::Webhook => RpcConfigSource::Webhook as i32, + Self::Web => RpcConfigSource::Web as i32, } } } @@ -311,7 +311,7 @@ impl std::str::FromStr for ConfigSource { fn from_str(s: &str) -> Result { match s { "user" => Ok(Self::User), - "webhook" => Ok(Self::Webhook), + "web" => Ok(Self::Web), other => Err(format!("unknown network config source: {other}")), } } @@ -1357,14 +1357,14 @@ stun_servers = [ let config = TomlConfigLoader::default(); assert_eq!(config.get_network_config_source(), ConfigSource::User); - config.set_network_config_source(Some(ConfigSource::Webhook)); + config.set_network_config_source(Some(ConfigSource::Web)); let dumped = config.dump(); assert!(dumped.contains("[source]")); - assert!(dumped.contains("source = \"webhook\"")); + assert!(dumped.contains("source = \"web\"")); let loaded = TomlConfigLoader::new_from_str(&dumped).unwrap(); - assert_eq!(loaded.get_network_config_source(), ConfigSource::Webhook); + assert_eq!(loaded.get_network_config_source(), ConfigSource::Web); } #[test] diff --git a/easytier/src/gateway/udp_proxy.rs b/easytier/src/gateway/udp_proxy.rs index 45e71f30..c4cab103 100644 --- a/easytier/src/gateway/udp_proxy.rs +++ b/easytier/src/gateway/udp_proxy.rs @@ -40,6 +40,16 @@ use crate::{ #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] struct UdpNatKey { src_socket: SocketAddr, + dst_socket: SocketAddr, +} + +impl UdpNatKey { + fn new(src_socket: SocketAddr, dst_socket: SocketAddr) -> Self { + Self { + src_socket, + dst_socket, + } + } } #[derive(Debug)] @@ -204,13 +214,23 @@ impl UdpNatEntry { self_clone.mark_active(); - if src_v4.ip().is_loopback() { - src_v4.set_ip(virtual_ipv4); + let has_mapped_dst = real_ipv4 != mapped_ipv4; + let mut reply_src_ip = *src_v4.ip(); + + // Preserve the existing priority for proxy rules that expose a + // real loopback address as a mapped address. Other loopback + // replies come from local delivery to 127.0.0.1 for the local + // virtual IP and may need the mapped rewrite below. + if has_mapped_dst && reply_src_ip == real_ipv4 { + reply_src_ip = mapped_ipv4; + } else if reply_src_ip.is_loopback() { + reply_src_ip = virtual_ipv4; } - if *src_v4.ip() == real_ipv4 { - src_v4.set_ip(mapped_ipv4); + if has_mapped_dst && reply_src_ip == real_ipv4 { + reply_src_ip = mapped_ipv4; } + src_v4.set_ip(reply_src_ip); let Ok(_) = Self::compose_ipv4_packet( &self_clone, @@ -321,9 +341,10 @@ impl UdpProxy { "udp nat packet request received" ); - let nat_key = UdpNatKey { - src_socket: SocketAddr::new(ipv4.get_source().into(), udp_packet.get_source()), - }; + let nat_key = UdpNatKey::new( + SocketAddr::new(ipv4.get_source().into(), udp_packet.get_source()), + SocketAddr::new(ipv4.get_destination().into(), udp_packet.get_destination()), + ); let nat_entry = self .nat_table .entry(nat_key) @@ -487,3 +508,309 @@ impl Drop for UdpProxy { } } } + +#[cfg(test)] +mod tests { + use std::{ + net::{Ipv4Addr, SocketAddr}, + sync::Arc, + time::Duration, + }; + + use pnet::packet::{ + MutablePacket, Packet, + ip::IpNextHeaderProtocols, + ipv4::{self, Ipv4Packet, MutableIpv4Packet}, + udp::{self, MutableUdpPacket, UdpPacket}, + }; + use tokio::{net::UdpSocket, sync::mpsc::Receiver, time::timeout}; + + use crate::{ + common::{config::ConfigLoader, global_ctx::tests::get_mock_global_ctx}, + peers::{ + create_packet_recv_chan, + peer_manager::{PeerManager, RouteAlgoType}, + }, + tunnel::packet_def::{PacketType, ZCPacket}, + }; + + use super::UdpProxy; + + fn build_udp_proxy_packet( + src_ip: Ipv4Addr, + src_port: u16, + dst_socket: SocketAddr, + payload: &[u8], + ) -> ZCPacket { + let SocketAddr::V4(dst_socket) = dst_socket else { + panic!("test only builds IPv4 UDP packets"); + }; + let dst_ip = *dst_socket.ip(); + let mut packet = vec![0; 20 + 8 + payload.len()]; + let packet_len = packet.len() as u16; + + { + let mut ipv4_packet = MutableIpv4Packet::new(&mut packet).unwrap(); + ipv4_packet.set_version(4); + ipv4_packet.set_header_length(5); + ipv4_packet.set_total_length(packet_len); + ipv4_packet.set_ttl(64); + ipv4_packet.set_next_level_protocol(IpNextHeaderProtocols::Udp); + ipv4_packet.set_source(src_ip); + ipv4_packet.set_destination(dst_ip); + } + + { + let mut udp_packet = MutableUdpPacket::new(&mut packet[20..]).unwrap(); + udp_packet.set_source(src_port); + udp_packet.set_destination(dst_socket.port()); + udp_packet.set_length((8 + payload.len()) as u16); + udp_packet.payload_mut().copy_from_slice(payload); + udp_packet.set_checksum(udp::ipv4_checksum( + &udp_packet.to_immutable(), + &src_ip, + &dst_ip, + )); + } + + { + let mut ipv4_packet = MutableIpv4Packet::new(&mut packet).unwrap(); + ipv4_packet.set_checksum(ipv4::checksum(&ipv4_packet.to_immutable())); + } + + let mut packet = ZCPacket::new_with_payload(&packet); + packet.fill_peer_manager_hdr(1009867077, 3831440917, PacketType::Data as u8); + packet + } + + async fn wait_proxy_cidr_loaded(proxy: &UdpProxy) { + timeout(Duration::from_secs(1), async { + while proxy.cidr_set.is_empty() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + } + + async fn recv_payload(socket: &UdpSocket) -> (Vec, SocketAddr) { + let mut buf = [0; 64]; + let (len, addr) = timeout(Duration::from_secs(1), socket.recv_from(&mut buf)) + .await + .unwrap() + .unwrap(); + (buf[..len].to_vec(), addr) + } + + async fn recv_response_packet(receiver: &mut Receiver) -> ZCPacket { + timeout(Duration::from_secs(1), receiver.recv()) + .await + .unwrap() + .unwrap() + } + + fn assert_udp_response( + packet: ZCPacket, + src_socket: SocketAddr, + dst_ip: Ipv4Addr, + dst_port: u16, + payload: &[u8], + ) { + let SocketAddr::V4(src_socket) = src_socket else { + panic!("test only checks IPv4 UDP packets"); + }; + let ipv4_packet = Ipv4Packet::new(packet.payload()).unwrap(); + assert_eq!(ipv4_packet.get_source(), *src_socket.ip()); + assert_eq!(ipv4_packet.get_destination(), dst_ip); + + let udp_packet = UdpPacket::new(ipv4_packet.payload()).unwrap(); + assert_eq!(udp_packet.get_source(), src_socket.port()); + assert_eq!(udp_packet.get_destination(), dst_port); + assert_eq!(udp_packet.payload(), payload); + } + + async fn stop_nat_entries(proxy: &UdpProxy) { + let nat_socket_addrs = proxy + .nat_table + .iter() + .filter_map(|entry| { + entry + .socket + .as_ref() + .and_then(|socket| socket.local_addr().ok()) + .map(|addr| SocketAddr::from((Ipv4Addr::LOCALHOST, addr.port()))) + }) + .collect::>(); + + for entry in proxy.nat_table.iter() { + entry.stop(); + } + + let wake_socket = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap(); + for addr in nat_socket_addrs { + let _ = wake_socket.send_to(b"wake", addr).await; + } + } + + #[tokio::test] + async fn udp_proxy_rewrites_unmapped_loopback_reply_to_virtual_ip() { + let global_ctx = get_mock_global_ctx(); + global_ctx.set_ipv4(Some("10.144.144.204/24".parse().unwrap())); + global_ctx + .config + .add_proxy_cidr("127.0.0.1/32".parse().unwrap(), None) + .unwrap(); + + let (packet_sender, _packet_receiver) = create_packet_recv_chan(); + let peer_manager = Arc::new(PeerManager::new( + RouteAlgoType::Ospf, + global_ctx.clone(), + packet_sender, + )); + let proxy = UdpProxy::new(global_ctx, peer_manager).unwrap(); + wait_proxy_cidr_loaded(&proxy).await; + let mut response_receiver = proxy.receiver.lock().await.take().unwrap(); + + let real_dst = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap(); + let real_dst_port = real_dst.local_addr().unwrap().port(); + let dst_socket = SocketAddr::from((Ipv4Addr::LOCALHOST, real_dst_port)); + let src_ip = Ipv4Addr::new(10, 144, 144, 206); + let src_port = 53864; + + let packet = build_udp_proxy_packet(src_ip, src_port, dst_socket, b"request"); + assert!(proxy.try_handle_packet(&packet).await.is_some()); + let (payload, nat_socket) = recv_payload(&real_dst).await; + assert_eq!(payload, b"request"); + + real_dst.send_to(b"reply", nat_socket).await.unwrap(); + assert_udp_response( + recv_response_packet(&mut response_receiver).await, + SocketAddr::from((Ipv4Addr::new(10, 144, 144, 204), real_dst_port)), + src_ip, + src_port, + b"reply", + ); + + stop_nat_entries(&proxy).await; + } + + #[tokio::test] + async fn udp_proxy_maps_local_virtual_destination_reply_to_mapped_source() { + let global_ctx = get_mock_global_ctx(); + global_ctx.set_ipv4(Some("10.144.144.204/24".parse().unwrap())); + global_ctx + .config + .add_proxy_cidr( + "10.144.144.204/32".parse().unwrap(), + Some("10.10.10.3/32".parse().unwrap()), + ) + .unwrap(); + + let (packet_sender, _packet_receiver) = create_packet_recv_chan(); + let peer_manager = Arc::new(PeerManager::new( + RouteAlgoType::Ospf, + global_ctx.clone(), + packet_sender, + )); + let proxy = UdpProxy::new(global_ctx, peer_manager).unwrap(); + wait_proxy_cidr_loaded(&proxy).await; + let mut response_receiver = proxy.receiver.lock().await.take().unwrap(); + + let real_dst = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap(); + let real_dst_port = real_dst.local_addr().unwrap().port(); + let mapped_dst = SocketAddr::from((Ipv4Addr::new(10, 10, 10, 3), real_dst_port)); + let src_ip = Ipv4Addr::new(10, 144, 144, 206); + let src_port = 53864; + + let packet = build_udp_proxy_packet(src_ip, src_port, mapped_dst, b"request"); + assert!(proxy.try_handle_packet(&packet).await.is_some()); + let (payload, nat_socket) = recv_payload(&real_dst).await; + assert_eq!(payload, b"request"); + + real_dst.send_to(b"reply", nat_socket).await.unwrap(); + assert_udp_response( + recv_response_packet(&mut response_receiver).await, + mapped_dst, + src_ip, + src_port, + b"reply", + ); + + stop_nat_entries(&proxy).await; + } + + #[tokio::test] + async fn udp_proxy_separates_same_source_port_to_multiple_mapped_destinations() { + let global_ctx = get_mock_global_ctx(); + global_ctx.set_ipv4(Some("10.144.144.204/24".parse().unwrap())); + global_ctx + .config + .add_proxy_cidr( + "127.0.0.1/32".parse().unwrap(), + Some("10.10.10.1/32".parse().unwrap()), + ) + .unwrap(); + global_ctx + .config + .add_proxy_cidr( + "127.0.0.1/32".parse().unwrap(), + Some("10.10.10.2/32".parse().unwrap()), + ) + .unwrap(); + + let (packet_sender, _packet_receiver) = create_packet_recv_chan(); + let peer_manager = Arc::new(PeerManager::new( + RouteAlgoType::Ospf, + global_ctx.clone(), + packet_sender, + )); + let proxy = UdpProxy::new(global_ctx, peer_manager).unwrap(); + wait_proxy_cidr_loaded(&proxy).await; + let mut response_receiver = proxy.receiver.lock().await.take().unwrap(); + + let real_dst = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap(); + let real_dst_port = real_dst.local_addr().unwrap().port(); + let first_mapped_dst = SocketAddr::from((Ipv4Addr::new(10, 10, 10, 1), real_dst_port)); + let second_mapped_dst = SocketAddr::from((Ipv4Addr::new(10, 10, 10, 2), real_dst_port)); + let src_ip = Ipv4Addr::new(10, 144, 144, 206); + let src_port = 53864; + + let first_packet = build_udp_proxy_packet(src_ip, src_port, first_mapped_dst, b"first"); + assert!(proxy.try_handle_packet(&first_packet).await.is_some()); + let (payload, first_nat_socket) = recv_payload(&real_dst).await; + assert_eq!(payload, b"first"); + + let second_packet = build_udp_proxy_packet(src_ip, src_port, second_mapped_dst, b"second"); + assert!(proxy.try_handle_packet(&second_packet).await.is_some()); + let (payload, second_nat_socket) = recv_payload(&real_dst).await; + assert_eq!(payload, b"second"); + + assert_eq!(proxy.nat_table.len(), 2); + + real_dst + .send_to(b"first-reply", first_nat_socket) + .await + .unwrap(); + assert_udp_response( + recv_response_packet(&mut response_receiver).await, + first_mapped_dst, + src_ip, + src_port, + b"first-reply", + ); + + real_dst + .send_to(b"second-reply", second_nat_socket) + .await + .unwrap(); + assert_udp_response( + recv_response_packet(&mut response_receiver).await, + second_mapped_dst, + src_ip, + src_port, + b"second-reply", + ); + + stop_nat_entries(&proxy).await; + } +} diff --git a/easytier/src/peers/credential_manager.rs b/easytier/src/peers/credential_manager.rs index bc5988dd..c6e35645 100644 --- a/easytier/src/peers/credential_manager.rs +++ b/easytier/src/peers/credential_manager.rs @@ -127,6 +127,8 @@ impl CredentialManager { credential_id: Option, reusable: bool, ) -> (String, String) { + self.remove_expired_credentials(); + let mut credentials = self.credentials.lock().unwrap(); let id = if let Some(id) = credential_id .map(|x| x.trim().to_string()) @@ -194,6 +196,25 @@ impl CredentialManager { removed } + pub fn remove_expired_credentials(&self) -> bool { + self.remove_expired_credentials_at(current_unix_timestamp()) + } + + fn remove_expired_credentials_at(&self, now: i64) -> bool { + let removed = { + let mut credentials = self.credentials.lock().unwrap(); + let before = credentials.len(); + credentials.retain(|_, entry| entry.is_active_at(now)); + before != credentials.len() + }; + + if removed { + self.save_to_disk(); + } + + removed + } + pub fn get_trusted_pubkeys(&self, network_secret: &str) -> Vec { let now = current_unix_timestamp(); @@ -496,6 +517,35 @@ mod tests { assert_eq!(list.len(), 1); } + #[test] + fn test_remove_expired_credentials_removes_and_persists() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("creds.json"); + let mgr = CredentialManager::new(Some(path.clone())); + mgr.generate_credential_with_id( + vec!["active".to_string()], + false, + vec![], + Duration::from_secs(3600), + Some("active-id".to_string()), + ); + mgr.generate_credential_with_id( + vec!["expired".to_string()], + false, + vec![], + Duration::from_secs(0), + Some("expired-id".to_string()), + ); + + assert!(mgr.remove_expired_credentials()); + assert_eq!(mgr.list_credentials().len(), 1); + + let reloaded = CredentialManager::new(Some(path)); + let list = reloaded.list_credentials(); + assert_eq!(list.len(), 1); + assert_eq!(list[0].credential_id, "active-id"); + } + #[test] fn test_generate_with_specified_id_reuses_existing_result() { let mgr = CredentialManager::new(None); @@ -528,6 +578,37 @@ mod tests { assert_eq!(list[0].reusable, Some(true)); } + #[test] + fn test_generate_with_specified_id_replaces_expired_existing_result() { + let mgr = CredentialManager::new(None); + let fixed_id = "fixed-credential-id".to_string(); + let (id1, secret1) = mgr.generate_credential_with_id( + vec!["expired".to_string()], + false, + vec![], + Duration::from_secs(0), + Some(fixed_id.clone()), + ); + let (id2, secret2) = mgr.generate_credential_with_id( + vec!["fresh".to_string()], + true, + vec!["10.0.0.0/24".to_string()], + Duration::from_secs(3600), + Some(fixed_id.clone()), + ); + + assert_eq!(id1, fixed_id); + assert_eq!(id2, fixed_id); + assert_ne!(secret1, secret2); + + let list = mgr.list_credentials(); + assert_eq!(list.len(), 1); + assert_eq!(list[0].credential_id, fixed_id); + assert_eq!(list[0].groups, vec!["fresh".to_string()]); + assert!(list[0].allow_relay); + assert_eq!(list[0].allowed_proxy_cidrs, vec!["10.0.0.0/24".to_string()]); + } + #[test] fn test_generate_non_reusable_credential() { let mgr = CredentialManager::new(None); diff --git a/easytier/src/peers/peer_manager.rs b/easytier/src/peers/peer_manager.rs index 43a2a6c7..2a76e654 100644 --- a/easytier/src/peers/peer_manager.rs +++ b/easytier/src/peers/peer_manager.rs @@ -503,6 +503,33 @@ impl PeerManager { }); } + async fn close_untrusted_credential_peers(peer_map: &Arc, global_ctx: &ArcGlobalCtx) { + let network_name = global_ctx.get_network_name(); + for peer_id in peer_map.list_peers() { + if !matches!( + peer_map.get_peer_identity_type(peer_id), + Some(PeerIdentityType::Credential) + ) { + continue; + } + let Some(peer) = peer_map.get_peer_by_id(peer_id) else { + continue; + }; + let Some(pubkey) = peer.get_peer_public_key() else { + continue; + }; + + if global_ctx.is_pubkey_trusted(&pubkey, &network_name) { + continue; + } + + tracing::warn!(?peer_id, "closing untrusted credential peer"); + if let Err(e) = peer_map.close_peer(peer_id).await { + tracing::warn!(?e, ?peer_id, "failed to close untrusted credential peer"); + } + } + } + fn build_foreign_network_manager_accessor( peer_map: &Arc, ) -> Box { @@ -1849,6 +1876,26 @@ impl PeerManager { }); } + async fn run_credential_gc_routine(&self) { + let global_ctx = self.global_ctx.clone(); + let peer_map = self.peers.clone(); + self.tasks.lock().await.spawn(async move { + loop { + if global_ctx.get_network_identity().network_secret.is_some() { + if global_ctx + .get_credential_manager() + .remove_expired_credentials() + { + global_ctx.issue_event(GlobalCtxEvent::CredentialChanged); + } + + Self::close_untrusted_credential_peers(&peer_map, &global_ctx).await; + } + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + }); + } + async fn run_traffic_metrics_gc_routine(&self) { let mut event_receiver = self.global_ctx.subscribe(); let traffic_metrics = self.traffic_metrics.clone(); @@ -1897,6 +1944,7 @@ impl PeerManager { self.run_relay_session_gc_routine().await; self.run_recent_traffic_gc_routine().await; self.run_peer_session_gc_routine().await; + self.run_credential_gc_routine().await; self.run_traffic_metrics_gc_routine().await; self.run_foriegn_network().await; @@ -2135,6 +2183,7 @@ impl PeerManager { #[cfg(test)] mod tests { + use base64::Engine; use std::{ fmt::Debug, sync::Arc, @@ -2164,7 +2213,7 @@ mod tests { }, }, proto::{ - common::{CompressionAlgoPb, NatType}, + common::{CompressionAlgoPb, NatType, SecureModeConfig}, peer_rpc::SecureAuthLevel, }, tunnel::{ @@ -3406,6 +3455,92 @@ mod tests { // a is client, b is server } + #[tokio::test] + async fn expired_credential_peer_conn_is_closed_without_ospf() { + let (admin_ch, _admin_rx) = create_packet_recv_chan(); + let admin_ctx = get_mock_global_ctx(); + admin_ctx.config.set_network_identity(NetworkIdentity::new( + "net1".to_string(), + "secret".to_string(), + )); + set_secure_mode_cfg(&admin_ctx, true); + let admin = Arc::new(PeerManager::new( + RouteAlgoType::None, + admin_ctx.clone(), + admin_ch, + )); + admin.run().await.unwrap(); + + let (_cred_id, cred_secret) = admin_ctx.get_credential_manager().generate_credential( + vec![], + false, + vec![], + Duration::from_secs(1), + ); + let privkey_bytes: [u8; 32] = base64::engine::general_purpose::STANDARD + .decode(&cred_secret) + .unwrap() + .try_into() + .unwrap(); + let private = x25519_dalek::StaticSecret::from(privkey_bytes); + let public = x25519_dalek::PublicKey::from(&private); + let (credential_ch, _credential_rx) = create_packet_recv_chan(); + let credential_ctx = get_mock_global_ctx(); + credential_ctx + .config + .set_network_identity(NetworkIdentity::new_credential("net1".to_string())); + credential_ctx + .config + .set_secure_mode(Some(SecureModeConfig { + enabled: true, + local_private_key: Some( + base64::engine::general_purpose::STANDARD.encode(private.as_bytes()), + ), + local_public_key: Some( + base64::engine::general_purpose::STANDARD.encode(public.as_bytes()), + ), + })); + let credential = Arc::new(PeerManager::new( + RouteAlgoType::None, + credential_ctx, + credential_ch, + )); + credential.run().await.unwrap(); + let credential_peer_id = credential.my_peer_id(); + + connect_peer_manager(credential.clone(), admin.clone()).await; + + wait_for_condition( + || { + let admin = admin.clone(); + async move { + admin + .get_peer_map() + .list_peer_conns(credential_peer_id) + .await + .is_some_and(|conns| !conns.is_empty()) + } + }, + Duration::from_secs(5), + ) + .await; + + wait_for_condition( + || { + let admin = admin.clone(); + async move { + admin + .get_peer_map() + .list_peer_conns(credential_peer_id) + .await + .is_none_or(|conns| conns.is_empty()) + } + }, + Duration::from_secs(5), + ) + .await; + } + #[tokio::test] async fn close_conn_in_foreign_network_client() { let peer_mgr_server = create_mock_peer_manager_with_name("server".to_string()).await; diff --git a/easytier/src/peers/peer_ospf_route.rs b/easytier/src/peers/peer_ospf_route.rs index 4011c78b..b790aa0b 100644 --- a/easytier/src/peers/peer_ospf_route.rs +++ b/easytier/src/peers/peer_ospf_route.rs @@ -441,6 +441,9 @@ struct SyncedRouteInfo { // Tracks the currently accepted peer for non-reusable credentials. // Maps credential pubkey bytes -> peer_id. non_reusable_credential_owners: DashMap, PeerId>, + // Duplicate non-reusable credential peers are kept for OSPF sync and topology + // reachability, but excluded from forwarding until owner election selects them. + suppressed_non_reusable_credential_peers: DashMap, version: AtomicVersion, } @@ -660,6 +663,36 @@ impl SyncedRouteInfo { } } + fn replace_suppressed_non_reusable_credential_peers( + &self, + suppressed_peers: BTreeSet, + ) -> bool { + let current: BTreeSet<_> = self + .suppressed_non_reusable_credential_peers + .iter() + .map(|entry| *entry.key()) + .collect(); + if current == suppressed_peers { + return false; + } + + self.suppressed_non_reusable_credential_peers + .retain(|peer_id, _| suppressed_peers.contains(peer_id)); + + for peer_id in suppressed_peers { + self.suppressed_non_reusable_credential_peers + .insert(peer_id, ()); + } + + self.version.inc(); + true + } + + fn is_route_suppressed(&self, peer_id: PeerId) -> bool { + self.suppressed_non_reusable_credential_peers + .contains_key(&peer_id) + } + fn update_credential_groups( &self, peer_infos: &OrderedHashMap, @@ -1233,11 +1266,13 @@ impl SyncedRouteInfo { where F: FnMut(PeerId) -> bool, { - self.verify_and_update_credential_trusts_with_active_peers_protecting( - network_secret, - is_peer_active, - None, - ) + let (untrusted_peers, global_trusted_keys, _) = self + .verify_and_update_credential_trusts_with_active_peers_protecting( + network_secret, + is_peer_active, + None, + ); + (untrusted_peers, global_trusted_keys) } fn verify_and_update_credential_trusts_with_active_peers_protecting( @@ -1248,6 +1283,7 @@ impl SyncedRouteInfo { ) -> ( Vec, HashMap, crate::common::global_ctx::TrustedKeyMetadata>, + bool, ) where F: FnMut(PeerId) -> bool, @@ -1261,14 +1297,18 @@ impl SyncedRouteInfo { let (all_trusted, global_trusted_keys) = self.collect_trusted_credentials(&peer_infos, network_secret, now); let prev_trusted = self.replace_trusted_credential_pubkeys(&all_trusted); - let (active_non_reusable_owners, duplicate_untrusted_peers) = + let (active_non_reusable_owners, mut duplicate_untrusted_peers) = self.collect_non_reusable_credential_owners(&peer_infos, &all_trusted, is_peer_active); + if let Some(protected_peer_id) = protected_peer_id { + duplicate_untrusted_peers.remove(&protected_peer_id); + } self.replace_non_reusable_credential_owners(active_non_reusable_owners); + let suppressed_changed = + self.replace_suppressed_non_reusable_credential_peers(duplicate_untrusted_peers); self.update_credential_groups(&peer_infos, &all_trusted); let mut untrusted_peers = Self::collect_revoked_credential_peers(&peer_infos, &prev_trusted, &all_trusted); - untrusted_peers.extend(duplicate_untrusted_peers); if let Some(protected_peer_id) = protected_peer_id { untrusted_peers.remove(&protected_peer_id); } @@ -1282,7 +1322,11 @@ impl SyncedRouteInfo { self.remove_peers(untrusted_peers.iter().copied()); } - (untrusted_peers.into_iter().collect(), global_trusted_keys) + ( + untrusted_peers.into_iter().collect(), + global_trusted_keys, + suppressed_changed, + ) } fn is_admin_peer(&self, info: &RoutePeerInfo) -> bool { @@ -1327,6 +1371,7 @@ type NextHopMap = DashMap; struct RouteTable { peer_infos: DashMap, next_hop_map: NextHopMap, + suppressed_peer_ids: DashMap, ipv4_peer_id_map: DashMap, ipv6_peer_id_map: DashMap, cidr_peer_id_map: ArcSwap>, @@ -1339,6 +1384,7 @@ impl RouteTable { RouteTable { peer_infos: DashMap::new(), next_hop_map: DashMap::new(), + suppressed_peer_ids: DashMap::new(), ipv4_peer_id_map: DashMap::new(), ipv6_peer_id_map: DashMap::new(), cidr_peer_id_map: ArcSwap::new(Arc::new(PrefixMap::new())), @@ -1348,6 +1394,13 @@ impl RouteTable { } fn get_next_hop(&self, dst_peer_id: PeerId) -> Option { + if self.suppressed_peer_ids.contains_key(&dst_peer_id) { + return None; + } + self.get_topology_next_hop(dst_peer_id) + } + + fn get_topology_next_hop(&self, dst_peer_id: PeerId) -> Option { let cur_version = self.next_hop_map_version.get(); self.next_hop_map.get(&dst_peer_id).and_then(|x| { if x.version >= cur_version { @@ -1362,6 +1415,18 @@ impl RouteTable { self.get_next_hop(peer_id).is_some() } + fn topology_peer_reachable(&self, peer_id: PeerId) -> bool { + self.get_topology_next_hop(peer_id).is_some() + } + + fn sync_suppressed_peer_ids(&self, synced_info: &SyncedRouteInfo) { + self.suppressed_peer_ids + .retain(|peer_id, _| synced_info.is_route_suppressed(*peer_id)); + for entry in synced_info.suppressed_non_reusable_credential_peers.iter() { + self.suppressed_peer_ids.insert(*entry.key(), ()); + } + } + fn get_udp_nat_type(&self, peer_id: PeerId) -> Option { self.peer_infos .get(&peer_id) @@ -1398,21 +1463,24 @@ impl RouteTable { } for item in peer_id_to_node_index.iter() { - let src_peer_id = item.key(); + let src_peer_id = *item.key(); + if src_peer_id != my_peer_id && synced_info.is_route_suppressed(src_peer_id) { + continue; + } let src_node_idx = item.value(); let connected_peers: BTreeSet<_> = synced_info - .get_connected_peers(*src_peer_id) + .get_connected_peers(src_peer_id) .unwrap_or_default(); // if avoid relay, just set all outgoing edges to a large value: AVOID_RELAY_COST. - let peer_avoid_relay_data = synced_info.get_avoid_relay_data(*src_peer_id); + let peer_avoid_relay_data = synced_info.get_avoid_relay_data(src_peer_id); for dst_peer_id in connected_peers.iter() { let Some(dst_node_idx) = peer_id_to_node_index.get(dst_peer_id) else { continue; }; - let mut cost = cost_calc.calculate_cost(*src_peer_id, *dst_peer_id) as usize; + let mut cost = cost_calc.calculate_cost(src_peer_id, *dst_peer_id) as usize; if peer_avoid_relay_data { cost += AVOID_RELAY_COST; } @@ -1431,20 +1499,21 @@ impl RouteTable { v.version >= cur_version }); self.peer_infos.retain(|k, _| { - // remove peer info for peers we cannot reach. - self.next_hop_map.contains_key(k) + // remove peer info for peers we cannot forward to. + self.peer_reachable(*k) }); self.ipv4_peer_id_map.retain(|_, v| { - // remove ipv4 map for peers we cannot reach. - self.next_hop_map.contains_key(&v.peer_id) + // remove ipv4 map for peers we cannot forward to. + self.peer_reachable(v.peer_id) }); self.ipv6_peer_id_map.retain(|_, v| { - // remove ipv6 map for peers we cannot reach. - self.next_hop_map.contains_key(&v.peer_id) + // remove ipv6 map for peers we cannot forward to. + self.peer_reachable(v.peer_id) }); shrink_dashmap(&self.peer_infos, None); shrink_dashmap(&self.next_hop_map, None); + shrink_dashmap(&self.suppressed_peer_ids, None); shrink_dashmap(&self.ipv4_peer_id_map, None); shrink_dashmap(&self.ipv6_peer_id_map, None); } @@ -1545,6 +1614,7 @@ impl RouteTable { cost_calc: &T, ) { let version = synced_info.version.get(); + self.sync_suppressed_peer_ids(synced_info); let local_proxy_cidrs = synced_info .peer_infos @@ -1594,6 +1664,10 @@ impl RouteTable { } let peer_id = item.key(); + if !self.peer_reachable(*peer_id) { + continue; + } + let Some(info) = synced_info.peer_infos.read().get(peer_id).cloned() else { continue; }; @@ -1717,6 +1791,7 @@ impl RouteTable { cidrs_v6 = ?self.cidr_v6_peer_id_map.load(), "update peer cidr map" ); + self.clean_expired_route_info(); } fn get_peer_id_for_proxy(&self, ip: &IpAddr) -> Option { @@ -2147,6 +2222,7 @@ impl PeerRouteServiceImpl { group_trust_map_cache: DashMap::new(), trusted_credential_pubkeys: DashMap::new(), non_reusable_credential_owners: DashMap::new(), + suppressed_non_reusable_credential_peers: DashMap::new(), version: AtomicVersion::new(), }, public_ipv6_service: std::sync::Mutex::new(Weak::new()), @@ -2170,10 +2246,9 @@ impl PeerRouteServiceImpl { ni.network_secret_digest.map(|d| d.to_vec()) } + #[cfg(test)] fn is_active_non_reusable_credential_peer(&self, peer_id: PeerId) -> bool { - peer_id == self.my_peer_id - || self.sessions.contains_key(&peer_id) - || self.route_table.peer_reachable(peer_id) + peer_id == self.my_peer_id || self.route_table.topology_peer_reachable(peer_id) } fn is_credential_node(&self) -> bool { @@ -2488,7 +2563,7 @@ impl PeerRouteServiceImpl { }; for item in self.synced_route_info.conn_map.read().iter() { let src_peer_id = *item.0; - if !self.route_table.peer_reachable(src_peer_id) { + if !self.route_table.topology_peer_reachable(src_peer_id) { continue; } add_to_all_peer_ids(src_peer_id, item.1.version.get()); @@ -2555,7 +2630,7 @@ impl PeerRouteServiceImpl { } // do not send unreachable peer info to dst peer. - if !self.route_table.peer_reachable(*peer_id) { + if !self.route_table.topology_peer_reachable(*peer_id) { unreachable_peers_for_peer_info.insert(*peer_id, peer_info.version); continue; } @@ -2573,7 +2648,7 @@ impl PeerRouteServiceImpl { return false; }; - if self.route_table.peer_reachable(*peer_id) { + if self.route_table.topology_peer_reachable(*peer_id) { route_infos.push(peer_info.clone()); } @@ -2623,7 +2698,7 @@ impl PeerRouteServiceImpl { continue; } - if !self.route_table.peer_reachable(*peer_id) { + if !self.route_table.topology_peer_reachable(*peer_id) { unreachable_peers_for_conn_info.insert(*peer_id, conn_info.version.get()); continue; } @@ -2641,7 +2716,7 @@ impl PeerRouteServiceImpl { return false; }; - if self.route_table.peer_reachable(*peer_id) { + if self.route_table.topology_peer_reachable(*peer_id) { add_to_conn_peer_list(*peer_id, conn_info); } @@ -2699,7 +2774,7 @@ impl PeerRouteServiceImpl { let my_conn_info_updated = self.update_my_conn_info().await; let my_foreign_network_updated = self.update_my_foreign_network().await; let mut untrusted_changed = false; - if my_peer_info_updated { + if my_peer_info_updated || my_conn_info_updated { untrusted_changed = self.refresh_credential_trusts_and_disconnect().await; } @@ -2757,7 +2832,7 @@ impl PeerRouteServiceImpl { fn refresh_credential_trusts(&self) -> Vec { let network_identity = self.global_ctx.get_network_identity(); - let (untrusted, global_trusted_keys) = self + let (untrusted, global_trusted_keys, _) = self .synced_route_info .verify_and_update_credential_trusts_with_active_peers_protecting( network_identity.network_secret.as_deref(), @@ -2777,17 +2852,19 @@ impl PeerRouteServiceImpl { // route table from the latest synced peer/conn state before checking active peers. self.update_route_table_and_cached_local_conn_bitmap(); - let (untrusted, global_trusted_keys) = self + let (untrusted, global_trusted_keys, suppressed_changed) = self .synced_route_info .verify_and_update_credential_trusts_with_active_peers_protecting( network_identity.network_secret.as_deref(), - |peer_id| self.is_active_non_reusable_credential_peer(peer_id), + |peer_id| { + peer_id == self.my_peer_id || self.route_table.topology_peer_reachable(peer_id) + }, Some(self.my_peer_id), ); self.global_ctx .update_trusted_keys(global_trusted_keys, &network_identity.network_name); - if !untrusted.is_empty() { + if !untrusted.is_empty() || suppressed_changed { self.update_route_table_and_cached_local_conn_bitmap(); } untrusted @@ -2866,7 +2943,7 @@ impl PeerRouteServiceImpl { if let Ok(d) = now.duration_since(peer_info.last_update.unwrap().try_into().unwrap()) && (d > REMOVE_DEAD_PEER_INFO_AFTER || (d > REMOVE_UNREACHABLE_PEER_INFO_AFTER - && !self.route_table.peer_reachable(*peer_id))) + && !self.route_table.topology_peer_reachable(*peer_id))) { to_remove.push(*peer_id); } @@ -4173,7 +4250,9 @@ mod tests { time::{Duration, SystemTime}, }; - use super::{NextHopInfo, PeerRoute, REMOVE_DEAD_PEER_INFO_AFTER, RouteConnInfo}; + use super::{ + NextHopInfo, PeerRoute, REMOVE_DEAD_PEER_INFO_AFTER, RouteConnInfo, SyncRouteSession, + }; use crate::proto::common::TimestampExt; use crate::{ common::{ @@ -4203,6 +4282,8 @@ mod tests { }, tunnel::common::tests::wait_for_condition, }; + use base64::Engine as _; + use base64::prelude::BASE64_STANDARD; struct AuthOnlyInterface { my_peer_id: PeerId, @@ -4438,6 +4519,31 @@ mod tests { peer_info } + fn make_admin_route_peer_info( + peer_id: PeerId, + credential_key: &[u8], + network_secret: &str, + now: i64, + ) -> RoutePeerInfo { + let mut admin_info = RoutePeerInfo::new(); + admin_info.peer_id = peer_id; + admin_info.version = 1; + admin_info.feature_flag = Some(PeerFeatureFlag { + is_credential_peer: false, + ..Default::default() + }); + admin_info.trusted_credential_pubkeys = vec![TrustedCredentialPubkeyProof::new_signed( + TrustedCredentialPubkey { + pubkey: credential_key.to_vec(), + expiry_unix: now + 600, + reusable: Some(false), + ..Default::default() + }, + network_secret, + )]; + admin_info + } + fn make_route_conn_info(connected_peers: I, last_update: SystemTime) -> RouteConnInfo where I: IntoIterator, @@ -4887,22 +4993,7 @@ mod tests { let credential_key = vec![7; 32]; - let mut admin_info = RoutePeerInfo::new(); - admin_info.peer_id = 30; - admin_info.version = 1; - admin_info.feature_flag = Some(PeerFeatureFlag { - is_credential_peer: false, - ..Default::default() - }); - admin_info.trusted_credential_pubkeys = vec![TrustedCredentialPubkeyProof::new_signed( - TrustedCredentialPubkey { - pubkey: credential_key.clone(), - expiry_unix: now + 600, - reusable: Some(false), - ..Default::default() - }, - network_secret, - )]; + let admin_info = make_admin_route_peer_info(30, &credential_key, network_secret, now); let mut original_peer = RoutePeerInfo::new(); original_peer.peer_id = 41; @@ -4953,9 +5044,9 @@ mod tests { let (second_untrusted, _) = service_impl .synced_route_info .verify_and_update_credential_trusts(Some(network_secret)); - assert_eq!(second_untrusted, vec![41]); + assert!(second_untrusted.is_empty()); assert!( - !service_impl + service_impl .synced_route_info .peer_infos .read() @@ -4976,6 +5067,8 @@ mod tests { .map(|entry| *entry.value()), Some(39) ); + assert!(service_impl.synced_route_info.is_route_suppressed(41)); + assert!(!service_impl.synced_route_info.is_route_suppressed(39)); } #[tokio::test] @@ -4991,22 +5084,7 @@ mod tests { let stale_peer_id = 41; let replacement_peer_id = 39; - let mut admin_info = RoutePeerInfo::new(); - admin_info.peer_id = 30; - admin_info.version = 1; - admin_info.feature_flag = Some(PeerFeatureFlag { - is_credential_peer: false, - ..Default::default() - }); - admin_info.trusted_credential_pubkeys = vec![TrustedCredentialPubkeyProof::new_signed( - TrustedCredentialPubkey { - pubkey: credential_key.clone(), - expiry_unix: now + 600, - reusable: Some(false), - ..Default::default() - }, - network_secret, - )]; + let admin_info = make_admin_route_peer_info(30, &credential_key, network_secret, now); let mut stale_peer = RoutePeerInfo::new(); stale_peer.peer_id = stale_peer_id; @@ -5077,6 +5155,292 @@ mod tests { .map(|entry| *entry.value()), Some(replacement_peer_id) ); + assert!( + !service_impl + .synced_route_info + .is_route_suppressed(stale_peer_id) + ); + assert!( + !service_impl + .synced_route_info + .is_route_suppressed(replacement_peer_id) + ); + } + + #[tokio::test] + async fn suppressed_non_reusable_credential_peer_stays_synced_and_can_be_reactivated() { + const NETWORK_SECRET: &str = "sec1"; + const SELF_PEER_ID: PeerId = 1; + const ADMIN_PEER_ID: PeerId = 30; + const FIRST_PEER_ID: PeerId = 39; + const SECOND_PEER_ID: PeerId = 41; + + let service_impl = PeerRouteServiceImpl::new( + SELF_PEER_ID, + get_mock_global_ctx_with_network(Some(NetworkIdentity::new( + "test-net".to_string(), + NETWORK_SECRET.to_string(), + ))), + ); + let now_unix = SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() as i64; + let now = SystemTime::now(); + let credential_key = vec![10; 32]; + + let mut self_info = RoutePeerInfo::new(); + self_info.peer_id = SELF_PEER_ID; + self_info.version = 1; + + let admin_info = + make_admin_route_peer_info(ADMIN_PEER_ID, &credential_key, NETWORK_SECRET, now_unix); + let mut first_peer = make_credential_route_peer_info(FIRST_PEER_ID, &credential_key); + first_peer.ipv4_addr = Some(std::net::Ipv4Addr::new(10, 144, 0, 39).into()); + let mut second_peer = make_credential_route_peer_info(SECOND_PEER_ID, &credential_key); + second_peer.ipv4_addr = Some(std::net::Ipv4Addr::new(10, 144, 0, 41).into()); + second_peer.proxy_cidrs.push("10.244.41.0/24".into()); + + { + let mut peer_infos = service_impl.synced_route_info.peer_infos.write(); + peer_infos.insert(self_info.peer_id, self_info); + peer_infos.insert(admin_info.peer_id, admin_info); + peer_infos.insert(first_peer.peer_id, first_peer); + peer_infos.insert(second_peer.peer_id, second_peer); + } + { + let mut conn_map = service_impl.synced_route_info.conn_map.write(); + conn_map.insert(SELF_PEER_ID, make_route_conn_info([ADMIN_PEER_ID], now)); + conn_map.insert( + ADMIN_PEER_ID, + make_route_conn_info([SELF_PEER_ID, FIRST_PEER_ID, SECOND_PEER_ID], now), + ); + conn_map.insert(FIRST_PEER_ID, make_route_conn_info([ADMIN_PEER_ID], now)); + conn_map.insert(SECOND_PEER_ID, make_route_conn_info([ADMIN_PEER_ID], now)); + } + service_impl.synced_route_info.version.set(1); + + let first_untrusted = service_impl.refresh_credential_trusts_with_current_topology(); + assert!(first_untrusted.is_empty()); + assert_eq!( + service_impl + .synced_route_info + .non_reusable_credential_owners + .get(&credential_key) + .map(|entry| *entry.value()), + Some(FIRST_PEER_ID) + ); + assert!( + service_impl + .synced_route_info + .peer_infos + .read() + .contains_key(&SECOND_PEER_ID) + ); + assert!( + service_impl + .synced_route_info + .is_route_suppressed(SECOND_PEER_ID) + ); + assert!( + service_impl + .route_table + .topology_peer_reachable(SECOND_PEER_ID) + ); + assert!(service_impl.route_table.peer_reachable(FIRST_PEER_ID)); + assert!(!service_impl.route_table.peer_reachable(SECOND_PEER_ID)); + assert!( + service_impl + .route_table + .peer_infos + .contains_key(&FIRST_PEER_ID) + ); + assert!( + !service_impl + .route_table + .peer_infos + .contains_key(&SECOND_PEER_ID) + ); + assert_eq!( + service_impl + .route_table + .ipv4_peer_id_map + .get(&"10.144.0.41".parse().unwrap()) + .map(|entry| entry.peer_id), + None + ); + assert_eq!( + service_impl + .route_table + .get_peer_id_for_proxy(&"10.244.41.1".parse().unwrap()), + None + ); + let sync_session = SyncRouteSession::new(SELF_PEER_ID, ADMIN_PEER_ID); + let sync_peer_ids: BTreeSet<_> = service_impl + .build_route_info(&sync_session) + .unwrap() + .into_iter() + .map(|info| info.peer_id) + .collect(); + assert!(sync_peer_ids.contains(&SECOND_PEER_ID)); + + { + let mut conn_map = service_impl.synced_route_info.conn_map.write(); + conn_map.insert( + ADMIN_PEER_ID, + make_route_conn_info([SELF_PEER_ID, SECOND_PEER_ID], now), + ); + conn_map.insert(FIRST_PEER_ID, make_route_conn_info([], now)); + conn_map.insert(SECOND_PEER_ID, make_route_conn_info([ADMIN_PEER_ID], now)); + } + service_impl.synced_route_info.version.inc(); + + let second_untrusted = service_impl.refresh_credential_trusts_with_current_topology(); + assert!(second_untrusted.is_empty()); + assert_eq!( + service_impl + .synced_route_info + .non_reusable_credential_owners + .get(&credential_key) + .map(|entry| *entry.value()), + Some(SECOND_PEER_ID) + ); + assert!( + !service_impl + .synced_route_info + .is_route_suppressed(SECOND_PEER_ID) + ); + assert!( + service_impl + .route_table + .topology_peer_reachable(SECOND_PEER_ID) + ); + assert!(!service_impl.route_table.peer_reachable(FIRST_PEER_ID)); + assert!(service_impl.route_table.peer_reachable(SECOND_PEER_ID)); + assert!( + !service_impl + .route_table + .peer_infos + .contains_key(&FIRST_PEER_ID) + ); + assert!( + service_impl + .route_table + .peer_infos + .contains_key(&SECOND_PEER_ID) + ); + assert_eq!( + service_impl + .route_table + .ipv4_peer_id_map + .get(&"10.144.0.41".parse().unwrap()) + .map(|entry| entry.peer_id), + Some(SECOND_PEER_ID) + ); + assert_eq!( + service_impl + .route_table + .get_peer_id_for_proxy(&"10.244.41.1".parse().unwrap()), + Some(SECOND_PEER_ID) + ); + } + + #[tokio::test] + async fn suppressed_non_reusable_credential_peer_is_not_transit_next_hop() { + const NETWORK_SECRET: &str = "sec1"; + const SELF_PEER_ID: PeerId = 1; + const ADMIN_PEER_ID: PeerId = 30; + const FIRST_PEER_ID: PeerId = 39; + const SECOND_PEER_ID: PeerId = 41; + const DOWNSTREAM_PEER_ID: PeerId = 50; + + let service_impl = PeerRouteServiceImpl::new( + SELF_PEER_ID, + get_mock_global_ctx_with_network(Some(NetworkIdentity::new( + "test-net".to_string(), + NETWORK_SECRET.to_string(), + ))), + ); + let now_unix = SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() as i64; + let now = SystemTime::now(); + let credential_key = vec![10; 32]; + + let mut self_info = RoutePeerInfo::new(); + self_info.peer_id = SELF_PEER_ID; + self_info.version = 1; + + let admin_info = + make_admin_route_peer_info(ADMIN_PEER_ID, &credential_key, NETWORK_SECRET, now_unix); + let first_peer = make_credential_route_peer_info(FIRST_PEER_ID, &credential_key); + let second_peer = make_credential_route_peer_info(SECOND_PEER_ID, &credential_key); + let mut downstream_peer = RoutePeerInfo::new(); + downstream_peer.peer_id = DOWNSTREAM_PEER_ID; + downstream_peer.version = 1; + + { + let mut peer_infos = service_impl.synced_route_info.peer_infos.write(); + peer_infos.insert(self_info.peer_id, self_info); + peer_infos.insert(admin_info.peer_id, admin_info); + peer_infos.insert(first_peer.peer_id, first_peer); + peer_infos.insert(second_peer.peer_id, second_peer); + peer_infos.insert(downstream_peer.peer_id, downstream_peer); + } + { + let mut conn_map = service_impl.synced_route_info.conn_map.write(); + conn_map.insert(SELF_PEER_ID, make_route_conn_info([ADMIN_PEER_ID], now)); + conn_map.insert( + ADMIN_PEER_ID, + make_route_conn_info([SELF_PEER_ID, FIRST_PEER_ID, SECOND_PEER_ID], now), + ); + conn_map.insert(FIRST_PEER_ID, make_route_conn_info([ADMIN_PEER_ID], now)); + conn_map.insert( + SECOND_PEER_ID, + make_route_conn_info([ADMIN_PEER_ID, DOWNSTREAM_PEER_ID], now), + ); + conn_map.insert( + DOWNSTREAM_PEER_ID, + make_route_conn_info([SECOND_PEER_ID], now), + ); + } + service_impl.synced_route_info.version.set(1); + + let untrusted = service_impl.refresh_credential_trusts_with_current_topology(); + assert!(untrusted.is_empty()); + assert_eq!( + service_impl + .synced_route_info + .non_reusable_credential_owners + .get(&credential_key) + .map(|entry| *entry.value()), + Some(FIRST_PEER_ID) + ); + assert!( + service_impl + .synced_route_info + .is_route_suppressed(SECOND_PEER_ID) + ); + assert!( + service_impl + .route_table + .topology_peer_reachable(SECOND_PEER_ID) + ); + assert!(!service_impl.route_table.peer_reachable(SECOND_PEER_ID)); + assert!(!service_impl.route_table.peer_reachable(DOWNSTREAM_PEER_ID)); + assert!( + service_impl + .route_table + .get_next_hop(DOWNSTREAM_PEER_ID) + .is_none() + ); + assert!( + !service_impl + .route_table + .peer_infos + .contains_key(&DOWNSTREAM_PEER_ID) + ); } #[tokio::test] @@ -5106,7 +5470,7 @@ mod tests { }, ); - let (untrusted_peers, _) = service_impl + let (untrusted_peers, _, _) = service_impl .synced_route_info .verify_and_update_credential_trusts_with_active_peers_protecting( None, @@ -5157,22 +5521,8 @@ mod tests { self_info.peer_id = SELF_PEER_ID; self_info.version = 1; - let mut admin_info = RoutePeerInfo::new(); - admin_info.peer_id = admin_peer_id; - admin_info.version = 1; - admin_info.feature_flag = Some(PeerFeatureFlag { - is_credential_peer: false, - ..Default::default() - }); - admin_info.trusted_credential_pubkeys = vec![TrustedCredentialPubkeyProof::new_signed( - TrustedCredentialPubkey { - pubkey: credential_key.clone(), - expiry_unix: now + 600, - reusable: Some(false), - ..Default::default() - }, - NETWORK_SECRET, - )]; + let admin_info = + make_admin_route_peer_info(admin_peer_id, &credential_key, NETWORK_SECRET, now); let stale_peer = make_credential_route_peer_info(stale_peer_id, &credential_key); let replacement_peer = @@ -5232,6 +5582,97 @@ mod tests { ); } + #[tokio::test] + async fn update_my_infos_refreshes_non_reusable_owner_on_conn_change() { + const NETWORK_SECRET: &str = "sec1"; + const ADMIN_PEER_ID: PeerId = 30; + const FIRST_PEER_ID: PeerId = 39; + const SECOND_PEER_ID: PeerId = 41; + + let global_ctx = get_mock_global_ctx_with_network(Some(NetworkIdentity::new( + "test-net".to_string(), + NETWORK_SECRET.to_string(), + ))); + let (_credential_id, credential_secret) = global_ctx + .get_credential_manager() + .generate_credential_with_options( + vec![], + false, + vec![], + Duration::from_secs(3600), + None, + false, + ); + let credential_secret_bytes: [u8; 32] = BASE64_STANDARD + .decode(&credential_secret) + .unwrap() + .try_into() + .unwrap(); + let credential_secret = x25519_dalek::StaticSecret::from(credential_secret_bytes); + let credential_key = x25519_dalek::PublicKey::from(&credential_secret) + .as_bytes() + .to_vec(); + + let service_impl = PeerRouteServiceImpl::new(ADMIN_PEER_ID, global_ctx); + let peers = Arc::new(Mutex::new(vec![FIRST_PEER_ID, SECOND_PEER_ID])); + let peer_identity_types = Arc::new(Mutex::new(HashMap::from([ + (FIRST_PEER_ID, Some(PeerIdentityType::Credential)), + (SECOND_PEER_ID, Some(PeerIdentityType::Credential)), + ]))); + *service_impl.interface.lock().await = Some(Box::new(CountingInterface { + my_peer_id: ADMIN_PEER_ID, + peers: peers.clone(), + peer_identity_types, + list_peers_calls: Arc::new(AtomicU32::new(0)), + get_peer_identity_type_calls: Arc::new(AtomicU32::new(0)), + })); + + { + let mut peer_infos = service_impl.synced_route_info.peer_infos.write(); + peer_infos.insert( + FIRST_PEER_ID, + make_credential_route_peer_info(FIRST_PEER_ID, &credential_key), + ); + peer_infos.insert( + SECOND_PEER_ID, + make_credential_route_peer_info(SECOND_PEER_ID, &credential_key), + ); + } + let now = SystemTime::now(); + { + let mut conn_map = service_impl.synced_route_info.conn_map.write(); + conn_map.insert(FIRST_PEER_ID, make_route_conn_info([ADMIN_PEER_ID], now)); + conn_map.insert(SECOND_PEER_ID, make_route_conn_info([ADMIN_PEER_ID], now)); + } + + assert!(service_impl.update_my_infos().await); + assert_eq!( + service_impl + .synced_route_info + .non_reusable_credential_owners + .get(&credential_key) + .map(|entry| *entry.value()), + Some(FIRST_PEER_ID) + ); + assert!(service_impl.route_table.peer_reachable(FIRST_PEER_ID)); + assert!(!service_impl.route_table.peer_reachable(SECOND_PEER_ID)); + + *peers.lock() = vec![SECOND_PEER_ID]; + service_impl.handle_global_ctx_event(&GlobalCtxEvent::PeerConnRemoved(Default::default())); + + assert!(service_impl.update_my_infos().await); + assert_eq!( + service_impl + .synced_route_info + .non_reusable_credential_owners + .get(&credential_key) + .map(|entry| *entry.value()), + Some(SECOND_PEER_ID) + ); + assert!(!service_impl.route_table.peer_reachable(FIRST_PEER_ID)); + assert!(service_impl.route_table.peer_reachable(SECOND_PEER_ID)); + } + #[tokio::test] async fn sync_route_info_marks_credential_sender_and_filters_entries() { let peer_mgr = create_mock_pmgr().await; diff --git a/easytier/src/peers/tests.rs b/easytier/src/peers/tests.rs index ed626316..5987b3f5 100644 --- a/easytier/src/peers/tests.rs +++ b/easytier/src/peers/tests.rs @@ -1220,9 +1220,6 @@ async fn credential_expiry_disconnects_from_all_admins() { .await; tokio::time::sleep(Duration::from_secs(3)).await; - admin_a - .get_global_ctx() - .issue_event(crate::common::global_ctx::GlobalCtxEvent::CredentialChanged); wait_for_condition( || { diff --git a/easytier/src/proto/api_manage.proto b/easytier/src/proto/api_manage.proto index 978042ca..32a5796a 100644 --- a/easytier/src/proto/api_manage.proto +++ b/easytier/src/proto/api_manage.proto @@ -16,7 +16,7 @@ enum NetworkingMethod { enum ConfigSource { ConfigSourceUnspecified = 0; ConfigSourceUser = 1; - ConfigSourceWebhook = 2; + ConfigSourceWeb = 2; } message NetworkConfig { diff --git a/easytier/src/rpc_service/remote_client.rs b/easytier/src/rpc_service/remote_client.rs index 5cabe5f3..9efd169d 100644 --- a/easytier/src/rpc_service/remote_client.rs +++ b/easytier/src/rpc_service/remote_client.rs @@ -52,6 +52,17 @@ where identify: T, config: NetworkConfig, save: bool, + ) -> Result<(), RemoteClientError> { + self.handle_run_network_instance_with_source(identify, config, save, ConfigSource::User) + .await + } + + async fn handle_run_network_instance_with_source( + &self, + identify: T, + config: NetworkConfig, + save: bool, + source: ConfigSource, ) -> Result<(), RemoteClientError> { let client = self .get_rpc_client(identify.clone()) @@ -63,7 +74,7 @@ where inst_id: None, config: Some(config.clone()), overwrite: true, - source: ConfigSource::User.to_rpc(), + source: source.to_rpc(), }, ) .await?; @@ -74,7 +85,7 @@ where identify, resp.inst_id.unwrap_or_default().into(), config, - ConfigSource::User, + source, ) .await .map_err(RemoteClientError::PersistentError)?; @@ -273,14 +284,20 @@ where identify: T, inst_id: uuid::Uuid, config: NetworkConfig, + ) -> Result<(), RemoteClientError> { + self.handle_save_network_config_with_source(identify, inst_id, config, ConfigSource::User) + .await + } + + async fn handle_save_network_config_with_source( + &self, + identify: T, + inst_id: uuid::Uuid, + config: NetworkConfig, + source: ConfigSource, ) -> Result<(), RemoteClientError> { self.get_storage() - .insert_or_update_user_network_config( - identify.clone(), - inst_id, - config, - ConfigSource::User, - ) + .insert_or_update_user_network_config(identify.clone(), inst_id, config, source) .await .map_err(RemoteClientError::PersistentError)?; self.get_storage()