Update kernel peers on peer modification (#4622)

* Include wireguard in gw config

* Support nym node first

* Create wg keypair

* Activate wg feature in gw dep

* Move key paths to separate structure

* Use client reg

* Generate and use own private key

* Rename network to ip for wg gw

* Propagate wireguard setup error message

* Remove logs

* Bump gateway version number

* Remove upgrade code

* Init wireguard on migration

* Upgrade code for nym-node too

* Wireguard paths upgrade

* Init wg keys on upgrade

* Simplify pub key translatations

* Fix clippy

* Undo comment change

* Fix tests

* Don't bump version just yet

* Remove redundant source attr

* Remove unused wg details

* Rename wg device

* Init for mixnode migration as well

* Add upgrade for gw wireguard deleted field

* Move interface removal to Drop trait impl

* Fix clippy

* Wgapi could be included on other platforms

* Handle peer control msgs

* Send add peer msg

* Handle double registration

* Fix deadlock

* Fix clippy

* More clippy

* Use same defguard

* Fix unit test

* Fix wasm build

* Move implementations from types to wireguard crate

* Fix linux

common/wireguard-types/src/error.rs

@@ -32,4 +32,7 @@ pub enum Error {
         #[source]
         source: hmac::digest::MacError,
     },
+
+    #[error("peers can't be modified anymore")]
+    PeerModifyStopped,
 }

common/wireguard-types/src/lib.rs

@@ -1,10 +1,6 @@
 // Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use dashmap::DashMap;
-use nym_crypto::asymmetric::encryption::KeyPair;
-use std::sync::Arc;
-
 pub mod config;
 pub mod error;
 pub mod public_key;
@@ -20,32 +16,3 @@ pub use registration::{
 
 #[cfg(feature = "verify")]
 pub use registration::HmacSha256;
-
-#[derive(Clone)]
-pub struct WireguardGatewayData {
-    config: Config,
-    keypair: Arc<KeyPair>,
-    client_registry: Arc<GatewayClientRegistry>,
-}
-
-impl WireguardGatewayData {
-    pub fn new(config: Config, keypair: Arc<KeyPair>) -> Self {
-        WireguardGatewayData {
-            config,
-            keypair,
-            client_registry: Arc::new(DashMap::default()),
-        }
-    }
-
-    pub fn config(&self) -> Config {
-        self.config
-    }
-
-    pub fn keypair(&self) -> &Arc<KeyPair> {
-        &self.keypair
-    }
-
-    pub fn client_registry(&self) -> &Arc<GatewayClientRegistry> {
-        &self.client_registry
-    }
-}

common/wireguard-types/src/registration.rs

@@ -17,7 +17,7 @@ use nym_crypto::asymmetric::encryption::PrivateKey;
 use sha2::Sha256;
 
 pub type GatewayClientRegistry = DashMap<PeerPublicKey, GatewayClient>;
-pub type PendingRegistrations = DashMap<PeerPublicKey, Nonce>;
+pub type PendingRegistrations = DashMap<PeerPublicKey, RegistrationData>;
 pub type PrivateIPs = DashMap<IpAddr, Free>;
 
 #[cfg(feature = "verify")]
@@ -56,14 +56,16 @@ impl InitMessage {
 #[serde(tag = "type", rename_all = "camelCase")]
 #[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
 pub enum ClientRegistrationResponse {
-    PendingRegistration {
-        nonce: u64,
-        gateway_data: GatewayClient,
-        wg_port: u16,
-    },
-    Registered {
-        success: bool,
-    },
+    PendingRegistration(RegistrationData),
+    Registered,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
+pub struct RegistrationData {
+    pub nonce: u64,
+    pub gateway_data: GatewayClient,
+    pub wg_port: u16,
 }
 
 /// Client that wants to register sends its PublicKey bytes mac digest encrypted with a DH shared secret.

common/wireguard/Cargo.toml

@@ -12,13 +12,15 @@ license.workspace = true
 
 [dependencies]
 base64 = "0.21.3"
+dashmap = { workspace = true }
 defguard_wireguard_rs = { workspace = true }
 # The latest version on crates.io at the time of writing this (6.0.0) has a
 # version mismatch with x25519-dalek/curve25519-dalek that is resolved in the
 # latest commit. So pick that for now.
 x25519-dalek = "2.0.0"
 ip_network = { workspace = true }
 log.workspace = true
+nym-crypto = { path = "../crypto", features = ["asymmetric"] }
 nym-network-defaults = { path = "../network-defaults" }
 nym-task = { path = "../task" }
 nym-wireguard-types = { path = "../wireguard-types" }

common/wireguard/src/lib.rs

@@ -3,64 +3,130 @@
 // #![warn(clippy::expect_used)]
 // #![warn(clippy::unwrap_used)]
 
-use defguard_wireguard_rs::WGApi;
+use dashmap::DashMap;
+use defguard_wireguard_rs::{host::Peer, key::Key, net::IpAddrMask, WGApi};
+use nym_crypto::asymmetric::encryption::KeyPair;
+use nym_wireguard_types::{Config, Error, GatewayClient, GatewayClientRegistry};
+use peer_controller::PeerControlMessage;
+use std::sync::Arc;
+use tokio::sync::mpsc::{self, UnboundedReceiver};
 
 const WG_TUN_NAME: &str = "nymwg";
 
+pub mod peer_controller;
+
 pub struct WgApiWrapper {
-    wg_api: WGApi,
+    inner: WGApi,
 }
 
 impl WgApiWrapper {
     pub fn new(wg_api: WGApi) -> Self {
-        WgApiWrapper { wg_api }
+        WgApiWrapper { inner: wg_api }
     }
 }
 
 impl Drop for WgApiWrapper {
     fn drop(&mut self) {
-        if let Err(e) = defguard_wireguard_rs::WireguardInterfaceApi::remove_interface(&self.wg_api)
+        if let Err(e) = defguard_wireguard_rs::WireguardInterfaceApi::remove_interface(&self.inner)
         {
             log::error!("Could not remove the wireguard interface: {:?}", e);
         }
     }
 }
 
+#[derive(Clone)]
+pub struct WireguardGatewayData {
+    config: Config,
+    keypair: Arc<KeyPair>,
+    client_registry: Arc<GatewayClientRegistry>,
+    peer_tx: mpsc::UnboundedSender<PeerControlMessage>,
+}
+
+impl WireguardGatewayData {
+    pub fn new(
+        config: Config,
+        keypair: Arc<KeyPair>,
+    ) -> (Self, mpsc::UnboundedReceiver<PeerControlMessage>) {
+        let (peer_tx, peer_rx) = mpsc::unbounded_channel();
+        (
+            WireguardGatewayData {
+                config,
+                keypair,
+                client_registry: Arc::new(DashMap::default()),
+                peer_tx,
+            },
+            peer_rx,
+        )
+    }
+
+    pub fn config(&self) -> Config {
+        self.config
+    }
+
+    pub fn keypair(&self) -> &Arc<KeyPair> {
+        &self.keypair
+    }
+
+    pub fn client_registry(&self) -> &Arc<GatewayClientRegistry> {
+        &self.client_registry
+    }
+
+    pub fn add_peer(&self, client: &GatewayClient) -> Result<(), Error> {
+        let mut peer = Peer::new(Key::new(client.pub_key.to_bytes()));
+        peer.allowed_ips
+            .push(IpAddrMask::new(client.private_ip, 32));
+        let msg = PeerControlMessage::AddPeer(peer);
+        self.peer_tx.send(msg).map_err(|_| Error::PeerModifyStopped)
+    }
+
+    pub fn remove_peer(&self, client: &GatewayClient) -> Result<(), Error> {
+        let key = Key::new(client.pub_key().to_bytes());
+        let msg = PeerControlMessage::RemovePeer(key);
+        self.peer_tx.send(msg).map_err(|_| Error::PeerModifyStopped)
+    }
+}
+
+pub struct WireguardData {
+    pub inner: WireguardGatewayData,
+    pub peer_rx: UnboundedReceiver<PeerControlMessage>,
+}
+
 /// Start wireguard device
 #[cfg(target_os = "linux")]
 pub async fn start_wireguard(
-    mut task_client: nym_task::TaskClient,
-    wireguard_data: std::sync::Arc<nym_wireguard_types::WireguardGatewayData>,
-) -> Result<WgApiWrapper, Box<dyn std::error::Error + Send + Sync + 'static>> {
+    task_client: nym_task::TaskClient,
+    wireguard_data: WireguardData,
+) -> Result<std::sync::Arc<WgApiWrapper>, Box<dyn std::error::Error + Send + Sync + 'static>> {
     use base64::{prelude::BASE64_STANDARD, Engine};
-    use defguard_wireguard_rs::{
-        host::Peer, key::Key, net::IpAddrMask, InterfaceConfiguration, WireguardInterfaceApi,
-    };
+    use defguard_wireguard_rs::{InterfaceConfiguration, WireguardInterfaceApi};
+    use peer_controller::PeerController;
 
     let mut peers = vec![];
-    for peer_client in wireguard_data.client_registry().iter() {
+    for peer_client in wireguard_data.inner.client_registry().iter() {
         let mut peer = Peer::new(Key::new(peer_client.pub_key.to_bytes()));
         let peer_ip_mask = IpAddrMask::new(peer_client.private_ip, 32);
         peer.set_allowed_ips(vec![peer_ip_mask]);
         peers.push(peer);
     }
 
     let ifname = String::from(WG_TUN_NAME);
-    let wgapi = WGApi::new(ifname.clone(), false)?;
-    wgapi.create_interface()?;
+    let wg_api = defguard_wireguard_rs::WGApi::new(ifname.clone(), false)?;
+    wg_api.create_interface()?;
     let interface_config = InterfaceConfiguration {
         name: ifname.clone(),
-        prvkey: BASE64_STANDARD.encode(wireguard_data.keypair().private_key().to_bytes()),
-        address: wireguard_data.config().private_ip.to_string(),
-        port: wireguard_data.config().announced_port as u32,
+        prvkey: BASE64_STANDARD.encode(wireguard_data.inner.keypair().private_key().to_bytes()),
+        address: wireguard_data.inner.config().private_ip.to_string(),
+        port: wireguard_data.inner.config().announced_port as u32,
         peers,
     };
-    wgapi.configure_interface(&interface_config)?;
+    wg_api.configure_interface(&interface_config)?;
     // wgapi.configure_peer_routing(&peers)?;
 
-    tokio::spawn(async move { task_client.recv().await });
+    let wg_api = std::sync::Arc::new(WgApiWrapper::new(wg_api));
+    let mut controller = PeerController::new(wg_api.clone(), wireguard_data.peer_rx);
+    tokio::spawn(async move { controller.run(task_client).await });
 
-    Ok(WgApiWrapper::new(wgapi))
+    Ok(wg_api)
 }
 
 #[cfg(not(target_os = "linux"))]

common/wireguard/src/peer_controller.rs

@@ -0,0 +1,58 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use std::sync::Arc;
+
+use defguard_wireguard_rs::{host::Peer, key::Key, WireguardInterfaceApi};
+use tokio::sync::mpsc;
+
+use crate::WgApiWrapper;
+
+pub enum PeerControlMessage {
+    AddPeer(Peer),
+    RemovePeer(Key),
+}
+
+pub struct PeerController {
+    peer_rx: mpsc::UnboundedReceiver<PeerControlMessage>,
+    wg_api: Arc<WgApiWrapper>,
+}
+
+impl PeerController {
+    pub fn new(
+        wg_api: Arc<WgApiWrapper>,
+        peer_rx: mpsc::UnboundedReceiver<PeerControlMessage>,
+    ) -> Self {
+        PeerController { wg_api, peer_rx }
+    }
+
+    pub async fn run(&mut self, mut task_client: nym_task::TaskClient) {
+        loop {
+            tokio::select! {
+                _ = task_client.recv() => {
+                    log::trace!("PeerController handler: Received shutdown");
+                    break;
+                }
+                msg = self.peer_rx.recv() => {
+                    match msg {
+                        Some(PeerControlMessage::AddPeer(peer)) => {
+                            if self.wg_api.inner.configure_peer(&peer).is_err() {
+                                log::error!("Could not configure peer {:?}", peer);
+                            }
+                        }
+                        Some(PeerControlMessage::RemovePeer(peer_pubkey)) => {
+                            if self.wg_api.inner.remove_peer(&peer_pubkey).is_err() {
+                                log::error!("Could not remove peer with key {:?}", peer_pubkey);
+                            }
+                        }
+                        None => {
+                            log::trace!("PeerController [main loop]: stopping since channel closed");
+                            break;
+
+                        }
+                    }
+                }
+            }
+        }
+    }
+}

gateway/src/http/mod.rs

@@ -14,7 +14,6 @@ use nym_node_http_api::api::api_requests::SignedHostInformation;
 use nym_node_http_api::NymNodeHttpError;
 use nym_sphinx::addressing::clients::Recipient;
 use nym_task::TaskClient;
-use nym_wireguard_types::WireguardGatewayData;
 use std::sync::Arc;
 
 fn load_gateway_details(
@@ -141,8 +140,6 @@ pub(crate) struct HttpApiBuilder<'a> {
     identity_keypair: &'a identity::KeyPair,
     // TODO: this should be a wg specific key and not re-used sphinx
     sphinx_keypair: Arc<encryption::KeyPair>,
-
-    wireguard_data: Option<Arc<WireguardGatewayData>>,
 }
 
 impl<'a> HttpApiBuilder<'a> {
@@ -158,7 +155,6 @@ impl<'a> HttpApiBuilder<'a> {
             exit_policy: None,
             identity_keypair,
             sphinx_keypair,
-            wireguard_data: None,
         }
     }
 
@@ -222,15 +218,6 @@ impl<'a> HttpApiBuilder<'a> {
         self
     }
 
-    #[must_use]
-    pub(crate) fn with_wireguard_data(
-        mut self,
-        wireguard_data: Option<Arc<WireguardGatewayData>>,
-    ) -> Self {
-        self.wireguard_data = wireguard_data;
-        self
-    }
-
     pub(crate) fn start(self, task_client: TaskClient) -> Result<(), GatewayError> {
         debug!("starting http API");