diff --git a/easytier/src/instance/shared_virtual_nic.rs b/easytier/src/instance/shared_virtual_nic.rs index f04c128d..aaa0009c 100644 --- a/easytier/src/instance/shared_virtual_nic.rs +++ b/easytier/src/instance/shared_virtual_nic.rs @@ -17,6 +17,9 @@ use crate::{ use super::virtual_nic::{VirtualNic, VirtualNicConfig}; +#[cfg(not(target_os = "linux"))] +use crate::common::ifcfg::IfConfiger; + mod dispatcher; use dispatcher::{SharedVirtualNicDispatcher, SharedVirtualNicMemberTunnelTable}; @@ -239,6 +242,13 @@ impl SharedIfConfig { effective_mtu: self.effective_mtu(), } } + + fn claims_of(&self, member_id: SharedVirtualNicMemberId) -> SharedIfConfigClaims { + self.member_claims + .get(&member_id) + .cloned() + .unwrap_or_default() + } } pub struct SharedVirtualNic { @@ -280,6 +290,97 @@ impl SharedVirtualNic { self.nic.clone() } + #[cfg(not(target_os = "linux"))] + async fn ifcfg_and_ifname(&self) -> Result<(IfConfiger, String), Error> { + self.ensure_valid()?; + let nic = self.nic.lock().await; + Ok((nic.get_ifcfg(), nic.ifname().to_owned())) + } + + async fn link_up(&self) -> Result<(), Error> { + self.ensure_valid()?; + self.nic.lock().await.link_up().await + } + + async fn apply_member_claims( + &mut self, + member_id: SharedVirtualNicMemberId, + claims: SharedIfConfigClaims, + ) -> Result<(), Error> { + self.ensure_valid()?; + + let mut next_ifcfg = self.ifcfg.clone(); + let delta = next_ifcfg.apply_member_claims(member_id, claims); + self.apply_ifcfg_delta(&delta).await?; + self.ifcfg = next_ifcfg; + + Ok(()) + } + + async fn remove_member_claims( + &mut self, + member_id: SharedVirtualNicMemberId, + ) -> Result<(), Error> { + self.ensure_valid()?; + + let mut next_ifcfg = self.ifcfg.clone(); + let Some(delta) = next_ifcfg.remove_member(member_id) else { + return Ok(()); + }; + self.apply_ifcfg_delta(&delta).await?; + self.ifcfg = next_ifcfg; + + Ok(()) + } + + async fn apply_ifcfg_delta(&self, delta: &SharedIfConfigDelta) -> Result<(), Error> { + let nic = self.nic.lock().await; + + for route in &delta.ipv4_routes.removed { + nic.remove_route(route.address, route.prefix).await?; + } + for route in &delta.ipv6_routes.removed { + nic.remove_ipv6_route(route.address, route.prefix).await?; + } + for ip in &delta.ipv4_addresses.removed { + nic.remove_ip(Some(*ip)).await?; + } + for ip in &delta.ipv6_addresses.removed { + nic.remove_ipv6(Some(*ip)).await?; + } + + for ip in &delta.ipv4_addresses.added { + nic.add_ip(ip.address(), ip.network_length() as i32).await?; + } + for ip in &delta.ipv6_addresses.added { + nic.add_ipv6(ip.address(), ip.network_length() as i32) + .await?; + } + for route in &delta.ipv4_routes.added { + nic.add_route_with_cost(route.address, route.prefix, route.cost) + .await?; + } + for route in &delta.ipv6_routes.added { + nic.add_ipv6_route_with_cost(route.address, route.prefix, route.cost) + .await?; + } + + if let Some(mtu) = &delta.mtu { + nic.set_mtu(mtu.new.unwrap_or_else(|| nic.configured_mtu())) + .await?; + } + + Ok(()) + } + + fn ensure_valid(&self) -> Result<(), Error> { + if self.is_valid() { + return Ok(()); + } + + Err(anyhow::anyhow!("shared virtual nic is invalid").into()) + } + fn member_tunnel_table(&self) -> SharedVirtualNicMemberTunnelTable { self.member_tunnel_table.clone() } @@ -289,9 +390,7 @@ impl SharedVirtualNic { } async fn ensure_dispatcher(&mut self) -> Result<(), Error> { - if !self.is_valid() { - return Err(anyhow::anyhow!("shared virtual nic is invalid").into()); - } + self.ensure_valid()?; if self.dispatcher.is_some() { return Ok(()); @@ -309,6 +408,7 @@ impl SharedVirtualNic { struct SharedVirtualNicMemberRegistration { member_id: SharedVirtualNicMemberId, + shared_nic: Arc>, member_tunnel_table: SharedVirtualNicMemberTunnelTable, } @@ -326,6 +426,27 @@ impl SharedVirtualNicMemberRegistration { impl Drop for SharedVirtualNicMemberRegistration { fn drop(&mut self) { self.member_tunnel_table.unregister(self.member_id); + let shared_nic = self.shared_nic.clone(); + let member_id = self.member_id; + + let Ok(handle) = tokio::runtime::Handle::try_current() else { + tracing::warn!( + ?member_id, + "skip shared virtual nic member claim cleanup without tokio runtime" + ); + return; + }; + + handle.spawn(async move { + let mut shared_nic = shared_nic.lock().await; + if let Err(err) = shared_nic.remove_member_claims(member_id).await { + tracing::warn!( + ?member_id, + ?err, + "failed to clean shared virtual nic member claims" + ); + } + }); } } @@ -346,10 +467,11 @@ impl SharedVirtualNicMember { ) -> Self { Self { member_id, - shared_nic, + shared_nic: shared_nic.clone(), close_notifier, registration: Arc::new(SharedVirtualNicMemberRegistration { member_id, + shared_nic: shared_nic.clone(), member_tunnel_table, }), } @@ -377,6 +499,119 @@ impl SharedVirtualNicMember { .register_tunnel(shared_tunnel, self.close_notifier.clone())?; Ok(member_tunnel) } + + #[cfg(not(target_os = "linux"))] + pub async fn ifcfg_and_ifname(&self) -> Result<(IfConfiger, String), Error> { + self.shared_nic.lock().await.ifcfg_and_ifname().await + } + + pub async fn link_up(&self) -> Result<(), Error> { + self.shared_nic.lock().await.link_up().await + } + + pub async fn add_ip(&self, ip: Ipv4Addr, cidr: i32) -> Result<(), Error> { + let ip = ipv4_inet(ip, cidr)?; + self.update_claims(|claims| { + claims.ipv4_addresses.insert(ip); + }) + .await + } + + pub async fn remove_ip(&self, ip: Option) -> Result<(), Error> { + self.update_claims(|claims| match ip { + Some(ip) => { + claims.ipv4_addresses.remove(&ip); + } + None => { + claims.ipv4_addresses.clear(); + } + }) + .await + } + + pub async fn add_ipv6(&self, ip: Ipv6Addr, cidr: i32) -> Result<(), Error> { + let ip = ipv6_inet(ip, cidr)?; + self.update_claims(|claims| { + claims.ipv6_addresses.insert(ip); + }) + .await + } + + pub async fn remove_ipv6(&self, ip: Option) -> Result<(), Error> { + self.update_claims(|claims| match ip { + Some(ip) => { + claims.ipv6_addresses.remove(&ip); + } + None => { + claims.ipv6_addresses.clear(); + } + }) + .await + } + + pub async fn add_route(&self, address: Ipv4Addr, cidr: u8) -> Result<(), Error> { + self.add_route_with_cost(address, cidr, None).await + } + + pub async fn add_route_with_cost( + &self, + address: Ipv4Addr, + cidr: u8, + cost: Option, + ) -> Result<(), Error> { + self.update_claims(|claims| { + claims + .ipv4_routes + .insert(SharedIpv4Route::new(address, cidr, cost)); + }) + .await + } + + pub async fn remove_route(&self, address: Ipv4Addr, cidr: u8) -> Result<(), Error> { + self.update_claims(|claims| { + claims + .ipv4_routes + .retain(|route| route.address != address || route.prefix != cidr); + }) + .await + } + + pub async fn add_ipv6_route(&self, address: Ipv6Addr, cidr: u8) -> Result<(), Error> { + self.add_ipv6_route_with_cost(address, cidr, None).await + } + + pub async fn add_ipv6_route_with_cost( + &self, + address: Ipv6Addr, + cidr: u8, + cost: Option, + ) -> Result<(), Error> { + self.update_claims(|claims| { + claims + .ipv6_routes + .insert(SharedIpv6Route::new(address, cidr, cost)); + }) + .await + } + + pub async fn remove_ipv6_route(&self, address: Ipv6Addr, cidr: u8) -> Result<(), Error> { + self.update_claims(|claims| { + claims + .ipv6_routes + .retain(|route| route.address != address || route.prefix != cidr); + }) + .await + } + + async fn update_claims(&self, update: F) -> Result<(), Error> + where + F: FnOnce(&mut SharedIfConfigClaims) + Send, + { + let mut shared_nic = self.shared_nic.lock().await; + let mut claims = shared_nic.ifcfg.claims_of(self.member_id); + update(&mut claims); + shared_nic.apply_member_claims(self.member_id, claims).await + } } #[derive(Default)] @@ -577,6 +812,22 @@ where owners.get(item).cloned().unwrap_or_default() } +fn ipv4_inet(address: Ipv4Addr, prefix: i32) -> Result { + let prefix = u8::try_from(prefix) + .map_err(|_| anyhow::anyhow!("invalid IPv4 prefix length {}", prefix))?; + Ipv4Inet::new(address, prefix).map_err(|err| { + anyhow::anyhow!("invalid IPv4 address {}/{}: {:?}", address, prefix, err).into() + }) +} + +fn ipv6_inet(address: Ipv6Addr, prefix: i32) -> Result { + let prefix = u8::try_from(prefix) + .map_err(|_| anyhow::anyhow!("invalid IPv6 prefix length {}", prefix))?; + Ipv6Inet::new(address, prefix).map_err(|err| { + anyhow::anyhow!("invalid IPv6 address {}/{}: {:?}", address, prefix, err).into() + }) +} + #[cfg(test)] mod tests { use std::str::FromStr as _; diff --git a/easytier/src/instance/virtual_nic.rs b/easytier/src/instance/virtual_nic.rs index e01b8cff..b0cc00ca 100644 --- a/easytier/src/instance/virtual_nic.rs +++ b/easytier/src/instance/virtual_nic.rs @@ -742,9 +742,26 @@ impl VirtualNic { } pub async fn add_route(&self, address: Ipv4Addr, cidr: u8) -> Result<(), Error> { + self.add_route_with_cost(address, cidr, None).await + } + + pub async fn add_route_with_cost( + &self, + address: Ipv4Addr, + cidr: u8, + cost: Option, + ) -> Result<(), Error> { let _g = self.config.net_ns.guard(); self.ifcfg - .add_ipv4_route(self.ifname(), address, cidr, None) + .add_ipv4_route(self.ifname(), address, cidr, cost) + .await?; + Ok(()) + } + + pub async fn remove_route(&self, address: Ipv4Addr, cidr: u8) -> Result<(), Error> { + let _g = self.config.net_ns.guard(); + self.ifcfg + .remove_ipv4_route(self.ifname(), address, cidr) .await?; Ok(()) } @@ -802,6 +819,16 @@ impl VirtualNic { Ok(()) } + pub async fn set_mtu(&self, mtu: u32) -> Result<(), Error> { + let _g = self.config.net_ns.guard(); + self.ifcfg.set_mtu(self.ifname(), mtu).await?; + Ok(()) + } + + pub fn configured_mtu(&self) -> u32 { + self.config.mtu + } + pub fn get_ifcfg(&self) -> IfConfiger { IfConfiger {} } @@ -814,10 +841,6 @@ pub enum NicBackend { } impl NicBackend { - fn shared_not_implemented(operation: &str) -> Error { - anyhow::anyhow!("shared virtual nic {} is not implemented", operation).into() - } - pub fn dedicated(nic: Arc>) -> Self { Self::Dedicated(nic) } @@ -867,34 +890,46 @@ impl NicBackend { } } + #[cfg(not(target_os = "linux"))] + /// Returns a raw ifcfg handle and interface name for platform cleanup. + /// + /// This does not carry `VirtualNic`'s netns guard. Use the typed + /// `NicBackend` methods for normal IP and route configuration. pub async fn ifcfg_and_ifname(&self) -> Result<(IfConfiger, String), Error> { match self { Self::Dedicated(nic) => { let nic = nic.lock().await; Ok((nic.get_ifcfg(), nic.ifname().to_owned())) } - Self::Shared(_) => Err(Self::shared_not_implemented("ifcfg")), + Self::Shared(member) => member.ifcfg_and_ifname().await, } } pub async fn link_up(&self) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.link_up().await, - Self::Shared(_) => Err(Self::shared_not_implemented("link_up")), + Self::Shared(member) => member.link_up().await, } } pub async fn add_route(&self, address: Ipv4Addr, cidr: u8) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.add_route(address, cidr).await, - Self::Shared(_) => Err(Self::shared_not_implemented("add_route")), + Self::Shared(member) => member.add_route(address, cidr).await, + } + } + + pub async fn remove_route(&self, address: Ipv4Addr, cidr: u8) -> Result<(), Error> { + match self { + Self::Dedicated(nic) => nic.lock().await.remove_route(address, cidr).await, + Self::Shared(member) => member.remove_route(address, cidr).await, } } pub async fn add_ipv6_route(&self, address: Ipv6Addr, cidr: u8) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.add_ipv6_route(address, cidr).await, - Self::Shared(_) => Err(Self::shared_not_implemented("add_ipv6_route")), + Self::Shared(member) => member.add_ipv6_route(address, cidr).await, } } @@ -911,42 +946,42 @@ impl NicBackend { .add_ipv6_route_with_cost(address, cidr, cost) .await } - Self::Shared(_) => Err(Self::shared_not_implemented("add_ipv6_route_with_cost")), + Self::Shared(member) => member.add_ipv6_route_with_cost(address, cidr, cost).await, } } pub async fn remove_ipv6_route(&self, address: Ipv6Addr, cidr: u8) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.remove_ipv6_route(address, cidr).await, - Self::Shared(_) => Err(Self::shared_not_implemented("remove_ipv6_route")), + Self::Shared(member) => member.remove_ipv6_route(address, cidr).await, } } pub async fn remove_ip(&self, ip: Option) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.remove_ip(ip).await, - Self::Shared(_) => Err(Self::shared_not_implemented("remove_ip")), + Self::Shared(member) => member.remove_ip(ip).await, } } pub async fn remove_ipv6(&self, ip: Option) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.remove_ipv6(ip).await, - Self::Shared(_) => Err(Self::shared_not_implemented("remove_ipv6")), + Self::Shared(member) => member.remove_ipv6(ip).await, } } pub async fn add_ip(&self, ip: Ipv4Addr, cidr: i32) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.add_ip(ip, cidr).await, - Self::Shared(_) => Err(Self::shared_not_implemented("add_ip")), + Self::Shared(member) => member.add_ip(ip, cidr).await, } } pub async fn add_ipv6(&self, ip: Ipv6Addr, cidr: i32) -> Result<(), Error> { match self { Self::Dedicated(nic) => nic.lock().await.add_ipv6(ip, cidr).await, - Self::Shared(_) => Err(Self::shared_not_implemented("add_ipv6")), + Self::Shared(member) => member.add_ipv6(ip, cidr).await, } } } @@ -1280,9 +1315,7 @@ impl NicCtx { } async fn apply_route_changes( - ifcfg: &impl IfConfiguerTrait, - ifname: &str, - net_ns: &crate::common::netns::NetNS, + backend: &NicBackend, cur_proxy_cidrs: &mut BTreeSet, added: Vec, removed: Vec, @@ -1294,9 +1327,8 @@ impl NicCtx { if !cur_proxy_cidrs.contains(&cidr) { continue; } - let _g = net_ns.guard(); - let ret = ifcfg - .remove_ipv4_route(ifname, cidr.first_address(), cidr.network_length()) + let ret = backend + .remove_route(cidr.first_address(), cidr.network_length()) .await; if ret.is_err() { @@ -1314,9 +1346,8 @@ impl NicCtx { if cur_proxy_cidrs.contains(&cidr) { continue; } - let _g = net_ns.guard(); - let ret = ifcfg - .add_ipv4_route(ifname, cidr.first_address(), cidr.network_length(), None) + let ret = backend + .add_route(cidr.first_address(), cidr.network_length()) .await; if ret.is_err() { @@ -1331,9 +1362,7 @@ impl NicCtx { } async fn apply_public_ipv6_route_changes( - ifcfg: &impl IfConfiguerTrait, - ifname: &str, - net_ns: &crate::common::netns::NetNS, + backend: &NicBackend, cur_routes: &mut BTreeSet, added: Vec, removed: Vec, @@ -1342,9 +1371,8 @@ impl NicCtx { if !cur_routes.contains(&route) { continue; } - let _g = net_ns.guard(); - let ret = ifcfg - .remove_ipv6_route(ifname, route.address(), route.network_length()) + let ret = backend + .remove_ipv6_route(route.address(), route.network_length()) .await; if ret.is_err() { tracing::trace!(route = ?route, err = ?ret, "remove public ipv6 route failed"); @@ -1356,9 +1384,8 @@ impl NicCtx { if cur_routes.contains(&route) { continue; } - let _g = net_ns.guard(); - let ret = ifcfg - .add_ipv6_route(ifname, route.address(), route.network_length(), None) + let ret = backend + .add_ipv6_route(route.address(), route.network_length()) .await; if ret.is_err() { tracing::trace!(route = ?route, err = ?ret, "add public ipv6 route failed"); @@ -1373,8 +1400,7 @@ impl NicCtx { return Err(anyhow::anyhow!("peer manager not available").into()); }; let global_ctx = self.global_ctx.clone(); - let net_ns = self.global_ctx.net_ns.clone(); - let (ifcfg, ifname) = self.backend.ifcfg_and_ifname().await?; + let backend = self.backend.clone(); let mut event_receiver = global_ctx.subscribe(); self.tasks.spawn(async move { @@ -1387,15 +1413,7 @@ impl NicCtx { &cur_proxy_cidrs, ) .await; - Self::apply_route_changes( - &ifcfg, - &ifname, - &net_ns, - &mut cur_proxy_cidrs, - added, - removed, - ) - .await; + Self::apply_route_changes(&backend, &mut cur_proxy_cidrs, added, removed).await; loop { let event = match event_receiver.recv().await { @@ -1426,15 +1444,7 @@ impl NicCtx { _ => continue, }; - Self::apply_route_changes( - &ifcfg, - &ifname, - &net_ns, - &mut cur_proxy_cidrs, - added, - removed, - ) - .await; + Self::apply_route_changes(&backend, &mut cur_proxy_cidrs, added, removed).await; } }); @@ -1446,8 +1456,7 @@ impl NicCtx { return Err(anyhow::anyhow!("peer manager not available").into()); }; let global_ctx = self.global_ctx.clone(); - let net_ns = self.global_ctx.net_ns.clone(); - let (ifcfg, ifname) = self.backend.ifcfg_and_ifname().await?; + let backend = self.backend.clone(); let mut event_receiver = global_ctx.subscribe(); self.tasks.spawn(async move { @@ -1455,9 +1464,7 @@ impl NicCtx { let initial_routes = peer_mgr.list_public_ipv6_routes().await; let initial_added = initial_routes.iter().copied().collect::>(); Self::apply_public_ipv6_route_changes( - &ifcfg, - &ifname, - &net_ns, + &backend, &mut cur_routes, initial_added, Vec::new(), @@ -1482,15 +1489,8 @@ impl NicCtx { _ => continue, }; - Self::apply_public_ipv6_route_changes( - &ifcfg, - &ifname, - &net_ns, - &mut cur_routes, - added, - removed, - ) - .await; + Self::apply_public_ipv6_route_changes(&backend, &mut cur_routes, added, removed) + .await; } });