Enforce verifying signed requests

* Move client type to the client code

* Check both version and type in header

Cargo.lock

@@ -4228,10 +4228,15 @@ dependencies = [
  "nym-bin-common",
  "nym-client-core",
  "nym-config",
+ "nym-credential-verification",
+ "nym-credentials-interface",
  "nym-crypto",
+ "nym-gateway-requests",
+ "nym-gateway-storage",
  "nym-id",
  "nym-network-defaults",
  "nym-sdk",
+ "nym-service-provider-requests-common",
  "nym-service-providers-common",
  "nym-sphinx",
  "nym-task",
@@ -4252,11 +4257,19 @@ dependencies = [
 name = "nym-authenticator-requests"
 version = "0.1.0"
 dependencies = [
+ "base64 0.22.1",
  "bincode",
+ "hmac",
+ "nym-credentials-interface",
+ "nym-crypto",
+ "nym-service-provider-requests-common",
  "nym-sphinx",
  "nym-wireguard-types",
  "rand",
  "serde",
+ "sha2 0.10.8",
+ "thiserror",
+ "x25519-dalek",
 ]
 
 [[package]]
@@ -5483,7 +5496,6 @@ dependencies = [
  "nym-node-requests",
  "nym-task",
  "nym-wireguard",
- "nym-wireguard-types",
  "rand",
  "serde_json",
  "thiserror",
@@ -5678,6 +5690,13 @@ dependencies = [
  "time",
 ]
 
+[[package]]
+name = "nym-service-provider-requests-common"
+version = "0.1.0"
+dependencies = [
+ "serde",
+]
+
 [[package]]
 name = "nym-service-providers-common"
 version = "0.1.0"
@@ -6192,8 +6211,11 @@ dependencies = [
  "chrono",
  "dashmap",
  "defguard_wireguard_rs",
+ "futures",
  "ip_network",
  "log",
+ "nym-authenticator-requests",
+ "nym-credential-verification",
  "nym-crypto",
  "nym-gateway-storage",
  "nym-network-defaults",
@@ -6210,17 +6232,13 @@ name = "nym-wireguard-types"
 version = "0.1.0"
 dependencies = [
  "base64 0.22.1",
- "hmac",
  "log",
  "nym-config",
  "nym-crypto",
  "nym-network-defaults",
  "rand",
  "serde",
- "serde_json",
- "sha2 0.10.8",
  "thiserror",
- "utoipa",
  "x25519-dalek",
 ]
 

Cargo.toml

@@ -81,6 +81,7 @@ members = [
     "common/nyxd-scraper",
     "common/pemstore",
     "common/serde-helpers",
+    "common/service-provider-requests-common",
     "common/socks5-client-core",
     "common/socks5/proxy-helpers",
     "common/socks5/requests",

common/authenticator-requests/Cargo.toml

@@ -9,9 +9,24 @@ edition.workspace = true
 license.workspace = true
 
 [dependencies]
+base64 = { workspace = true }
 bincode = { workspace = true }
 rand = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
+thiserror = { workspace = true }
 
+nym-credentials-interface = { path = "../credentials-interface" }
+nym-crypto = { path = "../crypto", features = ["asymmetric"] }
+nym-service-provider-requests-common = { path = "../service-provider-requests-common" }
 nym-sphinx = { path = "../nymsphinx" }
 nym-wireguard-types = { path = "../wireguard-types" }
+
+## verify:
+hmac = { workspace = true, optional = true }
+sha2 = { workspace = true, optional = true }
+x25519-dalek = { workspace = true, features = ["static_secrets"] }
+
+[features]
+default = ["verify"]
+# this is moved to a separate feature as we really need clients to import it (especially, *cough*, wasm)
+verify = ["hmac", "sha2"]

common/authenticator-requests/src/error.rs

@@ -0,0 +1,22 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("the provided base64-encoded client MAC ('{mac}') was malformed: {source}")]
+    MalformedClientMac {
+        mac: String,
+        #[source]
+        source: base64::DecodeError,
+    },
+
+    #[cfg(feature = "verify")]
+    #[error("failed to verify mac provided by '{client}': {source}")]
+    FailedClientMacVerification {
+        client: String,
+        #[source]
+        source: hmac::digest::MacError,
+    },
+}

common/authenticator-requests/src/lib.rs

@@ -2,8 +2,13 @@
 // SPDX-License-Identifier: Apache-2.0
 
 pub mod v1;
+pub mod v2;
 
-pub const CURRENT_VERSION: u8 = 1;
+mod error;
+
+pub use v2 as latest;
+
+pub const CURRENT_VERSION: u8 = 2;
 
 fn make_bincode_serializer() -> impl bincode::Options {
     use bincode::Options;

common/authenticator-requests/src/v1/mod.rs

@@ -1,7 +1,13 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+pub mod registration;
 pub mod request;
 pub mod response;
 
+pub use registration::{ClientMac, GatewayClient, InitMessage, Nonce};
+
+#[cfg(feature = "verify")]
+pub use registration::HmacSha256;
+
 const VERSION: u8 = 1;

common/authenticator-requests/src/v1/registration.rs

@@ -0,0 +1,218 @@
+// Copyright 2023-2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::error::Error;
+use base64::{engine::general_purpose, Engine};
+use nym_wireguard_types::PeerPublicKey;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::net::IpAddr;
+use std::time::SystemTime;
+use std::{fmt, ops::Deref, str::FromStr};
+
+#[cfg(feature = "verify")]
+use hmac::{Hmac, Mac};
+#[cfg(feature = "verify")]
+use nym_crypto::asymmetric::encryption::PrivateKey;
+#[cfg(feature = "verify")]
+use sha2::Sha256;
+
+pub type PendingRegistrations = HashMap<PeerPublicKey, RegistrationData>;
+pub type PrivateIPs = HashMap<IpAddr, Taken>;
+
+#[cfg(feature = "verify")]
+pub type HmacSha256 = Hmac<Sha256>;
+
+pub type Nonce = u64;
+pub type Taken = Option<SystemTime>;
+
+pub const BANDWIDTH_CAP_PER_DAY: u64 = 1024 * 1024 * 1024; // 1 GB
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct InitMessage {
+    /// Base64 encoded x25519 public key
+    pub pub_key: PeerPublicKey,
+}
+
+impl InitMessage {
+    pub fn new(pub_key: PeerPublicKey) -> Self {
+        InitMessage { pub_key }
+    }
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct RegistrationData {
+    pub nonce: u64,
+    pub gateway_data: GatewayClient,
+    pub wg_port: u16,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct RegistredData {
+    pub pub_key: PeerPublicKey,
+    pub private_ip: IpAddr,
+    pub wg_port: u16,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct RemainingBandwidthData {
+    pub available_bandwidth: u64,
+    pub suspended: bool,
+}
+
+/// Client that wants to register sends its PublicKey bytes mac digest encrypted with a DH shared secret.
+/// Gateway/Nym node can then verify pub_key payload using the same process
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct GatewayClient {
+    /// Base64 encoded x25519 public key
+    pub pub_key: PeerPublicKey,
+
+    /// Assigned private IP
+    pub private_ip: IpAddr,
+
+    /// Sha256 hmac on the data (alongside the prior nonce)
+    pub mac: ClientMac,
+}
+
+impl GatewayClient {
+    #[cfg(feature = "verify")]
+    pub fn new(
+        local_secret: &PrivateKey,
+        remote_public: x25519_dalek::PublicKey,
+        private_ip: IpAddr,
+        nonce: u64,
+    ) -> Self {
+        // convert from 1.0 x25519-dalek private key into 2.0 x25519-dalek
+        #[allow(clippy::expect_used)]
+        let static_secret = x25519_dalek::StaticSecret::from(local_secret.to_bytes());
+        let local_public: x25519_dalek::PublicKey = (&static_secret).into();
+
+        let dh = static_secret.diffie_hellman(&remote_public);
+
+        // TODO: change that to use our nym_crypto::hmac module instead
+        #[allow(clippy::expect_used)]
+        let mut mac = HmacSha256::new_from_slice(dh.as_bytes())
+            .expect("x25519 shared secret is always 32 bytes long");
+
+        mac.update(local_public.as_bytes());
+        mac.update(private_ip.to_string().as_bytes());
+        mac.update(&nonce.to_le_bytes());
+
+        GatewayClient {
+            pub_key: PeerPublicKey::new(local_public),
+            private_ip,
+            mac: ClientMac(mac.finalize().into_bytes().to_vec()),
+        }
+    }
+
+    // Reusable secret should be gateways Wireguard PK
+    // Client should perform this step when generating its payload, using its own WG PK
+    #[cfg(feature = "verify")]
+    pub fn verify(&self, gateway_key: &PrivateKey, nonce: u64) -> Result<(), Error> {
+        // convert from 1.0 x25519-dalek private key into 2.0 x25519-dalek
+        #[allow(clippy::expect_used)]
+        let static_secret = x25519_dalek::StaticSecret::from(gateway_key.to_bytes());
+
+        let dh = static_secret.diffie_hellman(&self.pub_key);
+
+        // TODO: change that to use our nym_crypto::hmac module instead
+        #[allow(clippy::expect_used)]
+        let mut mac = HmacSha256::new_from_slice(dh.as_bytes())
+            .expect("x25519 shared secret is always 32 bytes long");
+
+        mac.update(self.pub_key.as_bytes());
+        mac.update(self.private_ip.to_string().as_bytes());
+        mac.update(&nonce.to_le_bytes());
+
+        mac.verify_slice(&self.mac)
+            .map_err(|source| Error::FailedClientMacVerification {
+                client: self.pub_key.to_string(),
+                source,
+            })
+    }
+
+    pub fn pub_key(&self) -> PeerPublicKey {
+        self.pub_key
+    }
+}
+
+// TODO: change the inner type into generic array of size HmacSha256::OutputSize
+// TODO2: rely on our internal crypto/hmac
+#[derive(Debug, Clone)]
+pub struct ClientMac(Vec<u8>);
+
+impl fmt::Display for ClientMac {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", general_purpose::STANDARD.encode(&self.0))
+    }
+}
+
+impl ClientMac {
+    #[allow(dead_code)]
+    pub fn new(mac: Vec<u8>) -> Self {
+        ClientMac(mac)
+    }
+}
+
+impl Deref for ClientMac {
+    type Target = Vec<u8>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl FromStr for ClientMac {
+    type Err = Error;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let mac_bytes: Vec<u8> =
+            general_purpose::STANDARD
+                .decode(s)
+                .map_err(|source| Error::MalformedClientMac {
+                    mac: s.to_string(),
+                    source,
+                })?;
+
+        Ok(ClientMac(mac_bytes))
+    }
+}
+
+impl Serialize for ClientMac {
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        let encoded_key = general_purpose::STANDARD.encode(self.0.clone());
+        serializer.serialize_str(&encoded_key)
+    }
+}
+
+impl<'de> Deserialize<'de> for ClientMac {
+    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
+        let encoded_key = String::deserialize(deserializer)?;
+        ClientMac::from_str(&encoded_key).map_err(serde::de::Error::custom)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use nym_crypto::asymmetric::encryption;
+
+    #[test]
+    #[cfg(feature = "verify")]
+    fn client_request_roundtrip() {
+        let mut rng = rand::thread_rng();
+
+        let gateway_key_pair = encryption::KeyPair::new(&mut rng);
+        let client_key_pair = encryption::KeyPair::new(&mut rng);
+
+        let nonce = 1234567890;
+
+        let client = GatewayClient::new(
+            client_key_pair.private_key(),
+            x25519_dalek::PublicKey::from(gateway_key_pair.public_key().to_bytes()),
+            "10.0.0.42".parse().unwrap(),
+            nonce,
+        );
+        assert!(client.verify(gateway_key_pair.private_key(), nonce).is_ok())
+    }
+}

common/authenticator-requests/src/v1/request.rs

@@ -1,8 +1,9 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use super::registration::{GatewayClient, InitMessage};
 use nym_sphinx::addressing::Recipient;
-use nym_wireguard_types::{GatewayClient, InitMessage, PeerPublicKey};
+use nym_wireguard_types::PeerPublicKey;
 use serde::{Deserialize, Serialize};
 
 use crate::make_bincode_serializer;
@@ -82,3 +83,24 @@ pub enum AuthenticatorRequestData {
     Final(GatewayClient),
     QueryBandwidth(PeerPublicKey),
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::str::FromStr;
+
+    #[test]
+    fn check_first_byte_version() {
+        let version = 2;
+        let data = AuthenticatorRequest {
+            version,
+            data: AuthenticatorRequestData::Initial(InitMessage::new(
+                PeerPublicKey::from_str("yvNUDpT5l7W/xDhiu6HkqTHDQwbs/B3J5UrLmORl1EQ=").unwrap(),
+            )),
+            reply_to: Recipient::try_from_base58_string("D1rrpsysCGCYXy9saP8y3kmNpGtJZUXN9SvFoUcqAsM9.9Ssso1ea5NfkbMASdiseDSjTN1fSWda5SgEVjdSN4CvV@GJqd3ZxpXWSNxTfx7B1pPtswpetH4LnJdFeLeuY5KUuN").unwrap(),
+            request_id: 1,
+        };
+        let bytes = data.to_bytes().unwrap();
+        assert_eq!(*bytes.first().unwrap(), version);
+    }
+}

common/authenticator-requests/src/v1/response.rs

@@ -1,8 +1,8 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use super::registration::{RegistrationData, RegistredData, RemainingBandwidthData};
 use nym_sphinx::addressing::Recipient;
-use nym_wireguard_types::registration::{RegistrationData, RegistredData, RemainingBandwidthData};
 use serde::{Deserialize, Serialize};
 
 use crate::make_bincode_serializer;

common/authenticator-requests/src/v2/conversion.rs

@@ -0,0 +1,69 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use nym_service_provider_requests_common::{Protocol, ServiceProviderType};
+
+use crate::{v1, v2};
+
+impl From<v1::request::AuthenticatorRequest> for v2::request::AuthenticatorRequest {
+    fn from(authenticator_request: v1::request::AuthenticatorRequest) -> Self {
+        Self {
+            protocol: Protocol {
+                version: 2,
+                service_provider_type: ServiceProviderType::Authenticator,
+            },
+            data: authenticator_request.data.into(),
+            reply_to: authenticator_request.reply_to,
+            request_id: authenticator_request.request_id,
+        }
+    }
+}
+
+impl From<v1::request::AuthenticatorRequestData> for v2::request::AuthenticatorRequestData {
+    fn from(authenticator_request_data: v1::request::AuthenticatorRequestData) -> Self {
+        match authenticator_request_data {
+            v1::request::AuthenticatorRequestData::Initial(init_msg) => {
+                v2::request::AuthenticatorRequestData::Initial(init_msg.into())
+            }
+            v1::request::AuthenticatorRequestData::Final(gw_client) => {
+                v2::request::AuthenticatorRequestData::Final(gw_client.into())
+            }
+            v1::request::AuthenticatorRequestData::QueryBandwidth(pub_key) => {
+                v2::request::AuthenticatorRequestData::QueryBandwidth(pub_key)
+            }
+        }
+    }
+}
+
+impl From<v1::registration::InitMessage> for v2::registration::InitMessage {
+    fn from(init_msg: v1::registration::InitMessage) -> Self {
+        Self {
+            pub_key: init_msg.pub_key,
+        }
+    }
+}
+
+impl From<v1::registration::GatewayClient> for Box<v2::registration::FinalMessage> {
+    fn from(gw_client: v1::registration::GatewayClient) -> Self {
+        Box::new(v2::registration::FinalMessage {
+            gateway_client: gw_client.into(),
+            credential: None,
+        })
+    }
+}
+
+impl From<v1::registration::GatewayClient> for v2::registration::GatewayClient {
+    fn from(gw_client: v1::registration::GatewayClient) -> Self {
+        Self {
+            pub_key: gw_client.pub_key,
+            private_ip: gw_client.private_ip,
+            mac: gw_client.mac.into(),
+        }
+    }
+}
+
+impl From<v1::registration::ClientMac> for v2::registration::ClientMac {
+    fn from(mac: v1::registration::ClientMac) -> Self {
+        Self::new(mac.to_vec())
+    }
+}

common/authenticator-requests/src/v2/mod.rs

@@ -0,0 +1,9 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+pub mod conversion;
+pub mod registration;
+pub mod request;
+pub mod response;
+
+const VERSION: u8 = 2;

common/authenticator-requests/src/v2/registration.rs

@@ -1,9 +1,10 @@
-// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2023-2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::error::Error;
-use crate::PeerPublicKey;
 use base64::{engine::general_purpose, Engine};
+use nym_credentials_interface::CredentialSpendingData;
+use nym_wireguard_types::PeerPublicKey;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::net::IpAddr;
@@ -29,32 +30,26 @@ pub type Taken = Option<SystemTime>;
 pub const BANDWIDTH_CAP_PER_DAY: u64 = 1024 * 1024 * 1024; // 1 GB
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
-#[serde(tag = "type", rename_all = "camelCase")]
-#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
-pub enum ClientMessage {
-    Initial(InitMessage),
-    Final(GatewayClient),
-    Query(PeerPublicKey),
-}
-
-#[derive(Serialize, Deserialize, Debug, Clone)]
-#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
 pub struct InitMessage {
     /// Base64 encoded x25519 public key
-    #[cfg_attr(feature = "openapi", schema(value_type = String, format = Byte))]
     pub pub_key: PeerPublicKey,
 }
 
 impl InitMessage {
-    pub fn pub_key(&self) -> PeerPublicKey {
-        self.pub_key
-    }
-
     pub fn new(pub_key: PeerPublicKey) -> Self {
         InitMessage { pub_key }
     }
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct FinalMessage {
+    /// Gateway client data
+    pub gateway_client: GatewayClient,
+
+    /// Ecash credential
+    pub credential: Option<CredentialSpendingData>,
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct RegistrationData {
     pub nonce: u64,
@@ -72,23 +67,19 @@ pub struct RegistredData {
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct RemainingBandwidthData {
     pub available_bandwidth: u64,
-    pub suspended: bool,
 }
 
 /// Client that wants to register sends its PublicKey bytes mac digest encrypted with a DH shared secret.
 /// Gateway/Nym node can then verify pub_key payload using the same process
 #[derive(Serialize, Deserialize, Debug, Clone)]
-#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
 pub struct GatewayClient {
     /// Base64 encoded x25519 public key
-    #[cfg_attr(feature = "openapi", schema(value_type = String, format = Byte))]
     pub pub_key: PeerPublicKey,
 
     /// Assigned private IP
     pub private_ip: IpAddr,
 
     /// Sha256 hmac on the data (alongside the prior nonce)
-    #[cfg_attr(feature = "openapi", schema(value_type = String, format = Byte))]
     pub mac: ClientMac,
 }
 

common/authenticator-requests/src/v2/request.rs

@@ -0,0 +1,116 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use super::registration::{FinalMessage, InitMessage};
+use nym_service_provider_requests_common::{Protocol, ServiceProviderType};
+use nym_sphinx::addressing::Recipient;
+use nym_wireguard_types::PeerPublicKey;
+use serde::{Deserialize, Serialize};
+
+use crate::make_bincode_serializer;
+
+use super::VERSION;
+
+fn generate_random() -> u64 {
+    use rand::RngCore;
+    let mut rng = rand::rngs::OsRng;
+    rng.next_u64()
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct AuthenticatorRequest {
+    pub protocol: Protocol,
+    pub data: AuthenticatorRequestData,
+    pub reply_to: Recipient,
+    pub request_id: u64,
+}
+
+impl AuthenticatorRequest {
+    pub fn from_reconstructed_message(
+        message: &nym_sphinx::receiver::ReconstructedMessage,
+    ) -> Result<Self, bincode::Error> {
+        use bincode::Options;
+        make_bincode_serializer().deserialize(&message.message)
+    }
+
+    pub fn new_initial_request(init_message: InitMessage, reply_to: Recipient) -> (Self, u64) {
+        let request_id = generate_random();
+        (
+            Self {
+                protocol: Protocol {
+                    service_provider_type: ServiceProviderType::Authenticator,
+                    version: VERSION,
+                },
+                data: AuthenticatorRequestData::Initial(init_message),
+                reply_to,
+                request_id,
+            },
+            request_id,
+        )
+    }
+
+    pub fn new_final_request(final_message: FinalMessage, reply_to: Recipient) -> (Self, u64) {
+        let request_id = generate_random();
+        (
+            Self {
+                protocol: Protocol {
+                    service_provider_type: ServiceProviderType::Authenticator,
+                    version: VERSION,
+                },
+                data: AuthenticatorRequestData::Final(Box::new(final_message)),
+                reply_to,
+                request_id,
+            },
+            request_id,
+        )
+    }
+
+    pub fn new_query_request(peer_public_key: PeerPublicKey, reply_to: Recipient) -> (Self, u64) {
+        let request_id = generate_random();
+        (
+            Self {
+                protocol: Protocol {
+                    service_provider_type: ServiceProviderType::Authenticator,
+                    version: VERSION,
+                },
+                data: AuthenticatorRequestData::QueryBandwidth(peer_public_key),
+                reply_to,
+                request_id,
+            },
+            request_id,
+        )
+    }
+
+    pub fn to_bytes(&self) -> Result<Vec<u8>, bincode::Error> {
+        use bincode::Options;
+        make_bincode_serializer().serialize(self)
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub enum AuthenticatorRequestData {
+    Initial(InitMessage),
+    Final(Box<FinalMessage>),
+    QueryBandwidth(PeerPublicKey),
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::str::FromStr;
+
+    #[test]
+    fn check_first_bytes_protocol() {
+        let version = 2;
+        let data = AuthenticatorRequest {
+            protocol: Protocol { version, service_provider_type: ServiceProviderType::Authenticator },
+            data: AuthenticatorRequestData::Initial(InitMessage::new(
+                PeerPublicKey::from_str("yvNUDpT5l7W/xDhiu6HkqTHDQwbs/B3J5UrLmORl1EQ=").unwrap(),
+            )),
+            reply_to: Recipient::try_from_base58_string("D1rrpsysCGCYXy9saP8y3kmNpGtJZUXN9SvFoUcqAsM9.9Ssso1ea5NfkbMASdiseDSjTN1fSWda5SgEVjdSN4CvV@GJqd3ZxpXWSNxTfx7B1pPtswpetH4LnJdFeLeuY5KUuN").unwrap(),
+            request_id: 1,
+        };
+        let bytes = *data.to_bytes().unwrap().first_chunk::<2>().unwrap();
+        assert_eq!(bytes, [version, ServiceProviderType::Authenticator as u8]);
+    }
+}

common/authenticator-requests/src/v2/response.rs

@@ -0,0 +1,129 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use super::registration::{RegistrationData, RegistredData, RemainingBandwidthData};
+use nym_service_provider_requests_common::{Protocol, ServiceProviderType};
+use nym_sphinx::addressing::Recipient;
+use serde::{Deserialize, Serialize};
+
+use crate::make_bincode_serializer;
+
+use super::VERSION;
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct AuthenticatorResponse {
+    pub protocol: Protocol,
+    pub data: AuthenticatorResponseData,
+    pub reply_to: Recipient,
+}
+
+impl AuthenticatorResponse {
+    pub fn new_pending_registration_success(
+        registration_data: RegistrationData,
+        request_id: u64,
+        reply_to: Recipient,
+    ) -> Self {
+        Self {
+            protocol: Protocol {
+                service_provider_type: ServiceProviderType::Authenticator,
+                version: VERSION,
+            },
+            data: AuthenticatorResponseData::PendingRegistration(PendingRegistrationResponse {
+                reply: registration_data,
+                reply_to,
+                request_id,
+            }),
+            reply_to,
+        }
+    }
+
+    pub fn new_registered(
+        registred_data: RegistredData,
+        reply_to: Recipient,
+        request_id: u64,
+    ) -> Self {
+        Self {
+            protocol: Protocol {
+                service_provider_type: ServiceProviderType::Authenticator,
+                version: VERSION,
+            },
+            data: AuthenticatorResponseData::Registered(RegisteredResponse {
+                reply: registred_data,
+                reply_to,
+                request_id,
+            }),
+            reply_to,
+        }
+    }
+
+    pub fn new_remaining_bandwidth(
+        remaining_bandwidth_data: Option<RemainingBandwidthData>,
+        reply_to: Recipient,
+        request_id: u64,
+    ) -> Self {
+        Self {
+            protocol: Protocol {
+                service_provider_type: ServiceProviderType::Authenticator,
+                version: VERSION,
+            },
+            data: AuthenticatorResponseData::RemainingBandwidth(RemainingBandwidthResponse {
+                reply: remaining_bandwidth_data,
+                reply_to,
+                request_id,
+            }),
+            reply_to,
+        }
+    }
+
+    pub fn recipient(&self) -> Recipient {
+        self.reply_to
+    }
+
+    pub fn to_bytes(&self) -> Result<Vec<u8>, bincode::Error> {
+        use bincode::Options;
+        make_bincode_serializer().serialize(self)
+    }
+
+    pub fn from_reconstructed_message(
+        message: &nym_sphinx::receiver::ReconstructedMessage,
+    ) -> Result<Self, bincode::Error> {
+        use bincode::Options;
+        make_bincode_serializer().deserialize(&message.message)
+    }
+
+    pub fn id(&self) -> Option<u64> {
+        match &self.data {
+            AuthenticatorResponseData::PendingRegistration(response) => Some(response.request_id),
+            AuthenticatorResponseData::Registered(response) => Some(response.request_id),
+            AuthenticatorResponseData::RemainingBandwidth(response) => Some(response.request_id),
+        }
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub enum AuthenticatorResponseData {
+    PendingRegistration(PendingRegistrationResponse),
+    Registered(RegisteredResponse),
+    RemainingBandwidth(RemainingBandwidthResponse),
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct PendingRegistrationResponse {
+    pub request_id: u64,
+    pub reply_to: Recipient,
+    pub reply: RegistrationData,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct RegisteredResponse {
+    pub request_id: u64,
+    pub reply_to: Recipient,
+    pub reply: RegistredData,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct RemainingBandwidthResponse {
+    pub request_id: u64,
+    pub reply_to: Recipient,
+    pub reply: Option<RemainingBandwidthData>,
+}

common/credential-verification/src/bandwidth_storage_manager.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use nym_credentials::ecash::utils::ecash_today;
-use nym_credentials_interface::Bandwidth;
+use nym_credentials_interface::{AvailableBandwidth, Bandwidth};
 use nym_gateway_requests::ServerResponse;
 use nym_gateway_storage::Storage;
 use si_scale::helpers::bibytes2;
@@ -41,6 +41,10 @@ impl<S: Storage + Clone + 'static> BandwidthStorageManager<S> {
         }
     }
 
+    pub fn available_bandwidth(&self) -> AvailableBandwidth {
+        self.client_bandwidth.bandwidth
+    }
+
     async fn sync_expiration(&mut self) -> Result<()> {
         self.storage
             .set_expiration(self.client_id, self.client_bandwidth.bandwidth.expiration)

common/credential-verification/src/client_bandwidth.rs

@@ -6,6 +6,9 @@ use std::time::Duration;
 use nym_credentials_interface::AvailableBandwidth;
 use time::OffsetDateTime;
 
+const DEFAULT_CLIENT_BANDWIDTH_MAX_FLUSHING_RATE: Duration = Duration::from_millis(5);
+const DEFAULT_CLIENT_BANDWIDTH_MAX_DELTA_FLUSHING_AMOUNT: i64 = 512 * 1024; // 512kB
+
 #[derive(Debug, Clone, Copy)]
 pub struct BandwidthFlushingBehaviourConfig {
     /// Defines maximum delay between client bandwidth information being flushed to the persistent storage.
@@ -15,6 +18,16 @@ pub struct BandwidthFlushingBehaviourConfig {
     pub client_bandwidth_max_delta_flushing_amount: i64,
 }
 
+impl Default for BandwidthFlushingBehaviourConfig {
+    fn default() -> Self {
+        Self {
+            client_bandwidth_max_flushing_rate: DEFAULT_CLIENT_BANDWIDTH_MAX_FLUSHING_RATE,
+            client_bandwidth_max_delta_flushing_amount:
+                DEFAULT_CLIENT_BANDWIDTH_MAX_DELTA_FLUSHING_AMOUNT,
+        }
+    }
+}
+
 #[derive(Debug, Clone, Copy)]
 pub struct ClientBandwidth {
     pub(crate) bandwidth: AvailableBandwidth,

common/credential-verification/src/ecash/mod.rs

@@ -73,6 +73,10 @@ where
         self.shared_state.verification_key(epoch_id).await
     }
 
+    pub fn storage(&self) -> &S {
+        &self.shared_state.storage
+    }
+
     //Check for duplicate pay_info, then check the payment, then insert pay_info if everything succeeded
     pub async fn check_payment(
         &self,

common/gateway-storage/migrations/20240920120000_wireguard_client_id.sql

@@ -0,0 +1,10 @@
+/*
+ * Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+ALTER TABLE wireguard_peer
+ADD COLUMN client_id INTEGER REFERENCES clients(id) DEFAULT NULL;
+
+ALTER TABLE wireguard_peer
+DROP COLUMN suspended;

common/gateway-storage/src/lib.rs

@@ -227,12 +227,14 @@ pub trait Storage: Send + Sync {
     /// # Arguments
     ///
     /// * `peer`: wireguard peer data to be stored
-    /// * `suspended`: if peer exists, but it's currently suspended
+    /// * `with_client_id`: if the peer should have a corresponding client_id
+    ///   (created with entry wireguard ticket) or live without one (or with an
+    ///   exiting one), for temporary backwards compatibility.
     async fn insert_wireguard_peer(
         &self,
         peer: &defguard_wireguard_rs::host::Peer,
-        suspended: bool,
-    ) -> Result<(), StorageError>;
+        with_client_id: bool,
+    ) -> Result<Option<i64>, StorageError>;
 
     /// Tries to retrieve available bandwidth for the particular peer.
     ///
@@ -334,14 +336,23 @@ impl Storage for PersistentStorage {
         client_address: DestinationAddressBytes,
         shared_keys: &SharedGatewayKey,
     ) -> Result<i64, StorageError> {
-        let client_id = self
-            .client_manager
-            .insert_client(ClientType::EntryMixnet)
-            .await?;
+        let client_address_bs58 = client_address.as_base58_string();
+        let client_id = match self
+            .shared_key_manager
+            .client_id(&client_address_bs58)
+            .await
+        {
+            Ok(client_id) => client_id,
+            _ => {
+                self.client_manager
+                    .insert_client(ClientType::EntryMixnet)
+                    .await?
+            }
+        };
         self.shared_key_manager
             .insert_shared_keys(
                 client_id,
-                client_address.as_base58_string(),
+                client_address_bs58,
                 shared_keys.aes128_ctr_hmac_bs58().as_deref(),
                 shared_keys.aes256_gcm_siv().as_deref(),
             )
@@ -640,12 +651,30 @@ impl Storage for PersistentStorage {
     async fn insert_wireguard_peer(
         &self,
         peer: &defguard_wireguard_rs::host::Peer,
-        suspended: bool,
-    ) -> Result<(), StorageError> {
+        with_client_id: bool,
+    ) -> Result<Option<i64>, StorageError> {
+        let client_id = match self
+            .wireguard_peer_manager
+            .retrieve_peer(&peer.public_key.to_string())
+            .await?
+        {
+            Some(peer) => peer.client_id,
+            _ => {
+                if with_client_id {
+                    Some(
+                        self.client_manager
+                            .insert_client(ClientType::EntryWireguard)
+                            .await?,
+                    )
+                } else {
+                    None
+                }
+            }
+        };
         let mut peer = WireguardPeer::from(peer.clone());
-        peer.suspended = suspended;
+        peer.client_id = client_id;
         self.wireguard_peer_manager.insert_peer(&peer).await?;
-        Ok(())
+        Ok(client_id)
     }
 
     async fn get_wireguard_peer(

common/gateway-storage/src/models.rs

@@ -116,7 +116,7 @@ pub struct WireguardPeer {
     pub rx_bytes: i64,
     pub persistent_keepalive_interval: Option<i64>,
     pub allowed_ips: Vec<u8>,
-    pub suspended: bool,
+    pub client_id: Option<i64>,
 }
 
 impl From<defguard_wireguard_rs::host::Peer> for WireguardPeer {
@@ -146,7 +146,7 @@ impl From<defguard_wireguard_rs::host::Peer> for WireguardPeer {
                 &value.allowed_ips,
             )
             .unwrap_or_default(),
-            suspended: false,
+            client_id: None,
         }
     }
 }

common/gateway-storage/src/wireguard_peers.rs

@@ -27,16 +27,16 @@ impl WgPeerManager {
     pub(crate) async fn insert_peer(&self, peer: &WireguardPeer) -> Result<(), sqlx::Error> {
         sqlx::query!(
             r#"
-                INSERT OR IGNORE INTO wireguard_peer(public_key, preshared_key, protocol_version, endpoint, last_handshake, tx_bytes, rx_bytes, persistent_keepalive_interval, allowed_ips, suspended)
+                INSERT OR IGNORE INTO wireguard_peer(public_key, preshared_key, protocol_version, endpoint, last_handshake, tx_bytes, rx_bytes, persistent_keepalive_interval, allowed_ips, client_id)
                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
 
                 UPDATE wireguard_peer 
-                SET preshared_key = ?, protocol_version = ?, endpoint = ?, last_handshake = ?, tx_bytes = ?, rx_bytes = ?, persistent_keepalive_interval = ?, allowed_ips = ?, suspended = ?
+                SET preshared_key = ?, protocol_version = ?, endpoint = ?, last_handshake = ?, tx_bytes = ?, rx_bytes = ?, persistent_keepalive_interval = ?, allowed_ips = ?, client_id = ?
                 WHERE public_key = ?
             "#,
-            peer.public_key, peer.preshared_key, peer.protocol_version, peer.endpoint, peer.last_handshake, peer.tx_bytes, peer.rx_bytes, peer.persistent_keepalive_interval, peer.allowed_ips, peer.suspended,
-
-            peer.preshared_key, peer.protocol_version, peer.endpoint, peer.last_handshake, peer.tx_bytes, peer.rx_bytes, peer.persistent_keepalive_interval, peer.allowed_ips, peer.suspended,peer.public_key,
+            peer.public_key, peer.preshared_key, peer.protocol_version, peer.endpoint, peer.last_handshake, peer.tx_bytes, peer.rx_bytes, peer.persistent_keepalive_interval, peer.allowed_ips, peer.client_id,
+            peer.preshared_key, peer.protocol_version, peer.endpoint, peer.last_handshake, peer.tx_bytes, peer.rx_bytes, peer.persistent_keepalive_interval, peer.allowed_ips, peer.client_id,
+            peer.public_key,
         )
         .execute(&self.connection_pool)
         .await?;
@@ -78,7 +78,7 @@ impl WgPeerManager {
         .await
     }
 
-    /// Retrieve the wireguard peer with the provided public key from the storage.
+    /// Remove the wireguard peer with the provided public key from the storage.
     ///
     /// # Arguments
     ///

common/http-api-client/src/lib.rs

@@ -18,7 +18,9 @@ pub use user_agent::UserAgent;
 
 mod user_agent;
 
-pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
+// The timeout is relatively high as we are often making requests over the mixnet, where latency is
+// high and chatty protocols take a while to complete.
+pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
 
 pub type PathSegments<'a> = &'a [&'a str];
 pub type Params<'a, K, V> = &'a [(K, V)];

common/ip-packet-requests/src/v6/conversion.rs

@@ -67,3 +67,19 @@ impl From<v7::response::InfoLevel> for v6::response::InfoLevel {
         }
     }
 }
+
+impl From<v7::response::UnrequestedDisconnectReason> for v6::response::UnrequestedDisconnectReason {
+    fn from(reason: v7::response::UnrequestedDisconnectReason) -> Self {
+        match reason {
+            v7::response::UnrequestedDisconnectReason::ClientMixnetTrafficTimeout => {
+                v6::response::UnrequestedDisconnectReason::ClientMixnetTrafficTimeout
+            }
+            v7::response::UnrequestedDisconnectReason::ClientTunTrafficTimeout => {
+                v6::response::UnrequestedDisconnectReason::ClientTunTrafficTimeout
+            }
+            v7::response::UnrequestedDisconnectReason::Other(reason) => {
+                v6::response::UnrequestedDisconnectReason::Other(reason)
+            }
+        }
+    }
+}

common/service-provider-requests-common/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "nym-service-provider-requests-common"
+version = "0.1.0"
+authors.workspace = true
+repository.workspace = true
+homepage.workspace = true
+documentation.workspace = true
+edition.workspace = true
+license.workspace = true
+rust-version.workspace = true
+readme.workspace = true
+
+[dependencies]
+serde = { workspace = true, features = ["derive"] }

common/service-provider-requests-common/src/lib.rs

@@ -0,0 +1,18 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[repr(u8)]
+pub enum ServiceProviderType {
+    NetworkRequester = 0,
+    IpPacketRouter = 1,
+    Authenticator = 2,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct Protocol {
+    pub version: u8,
+    pub service_provider_type: ServiceProviderType,
+}

common/wireguard-types/Cargo.toml

@@ -17,28 +17,10 @@ serde = { workspace = true, features = ["derive"] }
 thiserror = { workspace = true }
 
 nym-config = { path = "../config" }
-nym-crypto = { path = "../crypto", features = ["asymmetric"] }
 nym-network-defaults = { path = "../network-defaults" }
 
-# feature-specific dependencies:
-
-## verify:
-hmac = { workspace = true, optional = true }
-sha2 = { workspace = true, optional = true }
-
-## openapi:
-utoipa = { workspace = true, optional = true }
-serde_json = { workspace = true, optional = true }
-
 x25519-dalek = { workspace = true, features = ["static_secrets"] }
 
 [dev-dependencies]
 rand = { workspace = true }
 nym-crypto = { path = "../crypto", features = ["rand"] }
-
-
-[features]
-default = ["verify"]
-openapi = ["utoipa", "serde_json"]
-# this is moved to a separate feature as we really need clients to import it (especially, *cough*, wasm)
-verify = ["hmac", "sha2"]

common/wireguard-types/src/error.rs

@@ -5,13 +5,6 @@ use thiserror::Error;
 
 #[derive(Debug, Error)]
 pub enum Error {
-    #[error("the provided base64-encoded client MAC ('{mac}') was malformed: {source}")]
-    MalformedClientMac {
-        mac: String,
-        #[source]
-        source: base64::DecodeError,
-    },
-
     #[error("the provided base64-encoded client x25519 public key ('{pub_key}') was malformed: {source}")]
     MalformedPeerPublicKeyEncoding {
         pub_key: String,
@@ -24,12 +17,4 @@ pub enum Error {
         pub_key: String,
         decoded_length: usize,
     },
-
-    #[cfg(feature = "verify")]
-    #[error("failed to verify mac provided by '{client}': {source}")]
-    FailedClientMacVerification {
-        client: String,
-        #[source]
-        source: hmac::digest::MacError,
-    },
 }

common/wireguard-types/src/lib.rs

@@ -4,19 +4,11 @@
 pub mod config;
 pub mod error;
 pub mod public_key;
-pub mod registration;
 
 use std::time::Duration;
 
 pub use config::Config;
 pub use error::Error;
 pub use public_key::PeerPublicKey;
-pub use registration::{ClientMac, ClientMessage, GatewayClient, InitMessage, Nonce};
 
-// To avoid any problems, keep this stale check time bigger (>2x) then the bandwidth cap
-// reset time (currently that one is 24h, at UTC midnight)
-pub const DEFAULT_PEER_TIMEOUT: Duration = Duration::from_secs(60 * 60 * 24 * 3); // 3 days
 pub const DEFAULT_PEER_TIMEOUT_CHECK: Duration = Duration::from_secs(5); // 5 seconds
-
-#[cfg(feature = "verify")]
-pub use registration::HmacSha256;

common/wireguard/Cargo.toml

@@ -16,17 +16,21 @@ bincode = { workspace = true }
 chrono = { workspace = true }
 dashmap = { workspace = true }
 defguard_wireguard_rs = { workspace = true }
+futures = { 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 = { workspace = true }
 ip_network = { workspace = true }
 log.workspace = true
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "net", "io-util"] }
+tokio-stream = { workspace = true }
+
+nym-authenticator-requests = { path = "../authenticator-requests" }
+nym-credential-verification = { path = "../credential-verification" }
 nym-crypto = { path = "../crypto", features = ["asymmetric"] }
 nym-gateway-storage = { path = "../gateway-storage" }
 nym-network-defaults = { path = "../network-defaults" }
 nym-task = { path = "../task" }
 nym-wireguard-types = { path = "../wireguard-types" }
-thiserror = { workspace = true }
-tokio = { workspace = true, features = ["rt-multi-thread", "net", "io-util"] }
-tokio-stream = { workspace = true }

common/wireguard/src/error.rs

@@ -6,9 +6,18 @@ pub enum Error {
     #[error("peers in wireguard don't match with in-memory ")]
     PeerMismatch,
 
+    #[error("traffic byte data needs to be increasing")]
+    InconsistentConsumedBytes,
+
     #[error("{0}")]
     Defguard(#[from] defguard_wireguard_rs::error::WireguardInterfaceError),
 
+    #[error("internal {0}")]
+    Internal(String),
+
+    #[error("storage should have the requested bandwidht entry")]
+    MissingClientBandwidthEntry,
+
     #[error("{0}")]
     GatewayStorage(#[from] nym_gateway_storage::error::StorageError),
 }

common/wireguard/src/lib.rs

@@ -13,12 +13,13 @@ use nym_crypto::asymmetric::encryption::KeyPair;
 use nym_wireguard_types::Config;
 use peer_controller::PeerControlRequest;
 use std::sync::Arc;
-use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
+use tokio::sync::mpsc::{self, Receiver, Sender};
 
 const WG_TUN_NAME: &str = "nymwg";
 
 pub(crate) mod error;
 pub mod peer_controller;
+pub mod peer_handle;
 
 pub struct WgApiWrapper {
     inner: WGApi,
@@ -43,15 +44,12 @@ impl Drop for WgApiWrapper {
 pub struct WireguardGatewayData {
     config: Config,
     keypair: Arc<KeyPair>,
-    peer_tx: UnboundedSender<PeerControlRequest>,
+    peer_tx: Sender<PeerControlRequest>,
 }
 
 impl WireguardGatewayData {
-    pub fn new(
-        config: Config,
-        keypair: Arc<KeyPair>,
-    ) -> (Self, UnboundedReceiver<PeerControlRequest>) {
-        let (peer_tx, peer_rx) = mpsc::unbounded_channel();
+    pub fn new(config: Config, keypair: Arc<KeyPair>) -> (Self, Receiver<PeerControlRequest>) {
+        let (peer_tx, peer_rx) = mpsc::channel(1);
         (
             WireguardGatewayData {
                 config,
@@ -70,44 +68,45 @@ impl WireguardGatewayData {
         &self.keypair
     }
 
-    pub fn peer_tx(&self) -> &UnboundedSender<PeerControlRequest> {
+    pub fn peer_tx(&self) -> &Sender<PeerControlRequest> {
         &self.peer_tx
     }
 }
 
 pub struct WireguardData {
     pub inner: WireguardGatewayData,
-    pub peer_rx: UnboundedReceiver<PeerControlRequest>,
+    pub peer_rx: Receiver<PeerControlRequest>,
 }
 
 /// Start wireguard device
 #[cfg(target_os = "linux")]
-pub async fn start_wireguard<St: nym_gateway_storage::Storage + 'static>(
+pub async fn start_wireguard<St: nym_gateway_storage::Storage + Clone + 'static>(
     storage: St,
     all_peers: Vec<nym_gateway_storage::models::WireguardPeer>,
     task_client: nym_task::TaskClient,
     wireguard_data: WireguardData,
-    control_tx: UnboundedSender<peer_controller::PeerControlResponse>,
 ) -> Result<std::sync::Arc<WgApiWrapper>, Box<dyn std::error::Error + Send + Sync + 'static>> {
     use base64::{prelude::BASE64_STANDARD, Engine};
     use defguard_wireguard_rs::{InterfaceConfiguration, WireguardInterfaceApi};
     use ip_network::IpNetwork;
     use peer_controller::PeerController;
-
-    let mut peers = vec![];
-    let mut suspended_peers = vec![];
-    for storage_peer in all_peers {
-        let suspended = storage_peer.suspended;
-        let peer = Peer::try_from(storage_peer)?;
-        if suspended {
-            suspended_peers.push(peer);
-        } else {
-            peers.push(peer);
-        }
-    }
+    use std::collections::HashMap;
+    use tokio::sync::RwLock;
 
     let ifname = String::from(WG_TUN_NAME);
     let wg_api = defguard_wireguard_rs::WGApi::new(ifname.clone(), false)?;
+    let mut peer_bandwidth_managers = HashMap::with_capacity(all_peers.len());
+    let peers = all_peers
+        .into_iter()
+        .map(Peer::try_from)
+        .collect::<Result<Vec<_>, _>>()?;
+    for peer in peers.iter() {
+        let bandwidth_manager =
+            PeerController::generate_bandwidth_manager(storage.clone(), &peer.public_key)
+                .await?
+                .map(|bw_m| Arc::new(RwLock::new(bw_m)));
+        peer_bandwidth_managers.insert(peer.public_key.clone(), bandwidth_manager);
+    }
     wg_api.create_interface()?;
     let interface_config = InterfaceConfiguration {
         name: ifname.clone(),
@@ -130,16 +129,18 @@ pub async fn start_wireguard<St: nym_gateway_storage::Storage + 'static>(
     )]);
     wg_api.configure_peer_routing(&[catch_all_peer])?;
 
+    let host = wg_api.read_interface_data()?;
     let wg_api = std::sync::Arc::new(WgApiWrapper::new(wg_api));
     let mut controller = PeerController::new(
         storage,
         wg_api.clone(),
-        interface_config.peers,
-        suspended_peers,
+        host,
+        peer_bandwidth_managers,
+        wireguard_data.inner.peer_tx.clone(),
         wireguard_data.peer_rx,
-        control_tx,
+        task_client,
     );
-    tokio::spawn(async move { controller.run(task_client).await });
+    tokio::spawn(async move { controller.run().await });
 
     Ok(wg_api)
 }

common/wireguard/src/peer_controller.rs

@@ -1,259 +1,292 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use chrono::{Timelike, Utc};
-use defguard_wireguard_rs::{host::Peer, key::Key, WireguardInterfaceApi};
+use defguard_wireguard_rs::{
+    host::{Host, Peer},
+    key::Key,
+    WireguardInterfaceApi,
+};
+use futures::channel::oneshot;
+use nym_authenticator_requests::{
+    v1::registration::BANDWIDTH_CAP_PER_DAY, v2::registration::RemainingBandwidthData,
+};
+use nym_credential_verification::{
+    bandwidth_storage_manager::BandwidthStorageManager, BandwidthFlushingBehaviourConfig,
+    ClientBandwidth,
+};
 use nym_gateway_storage::Storage;
-use nym_wireguard_types::registration::{RemainingBandwidthData, BANDWIDTH_CAP_PER_DAY};
-use nym_wireguard_types::{DEFAULT_PEER_TIMEOUT, DEFAULT_PEER_TIMEOUT_CHECK};
-use std::time::SystemTime;
+use nym_wireguard_types::DEFAULT_PEER_TIMEOUT_CHECK;
 use std::{collections::HashMap, sync::Arc};
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, RwLock};
 use tokio_stream::{wrappers::IntervalStream, StreamExt};
 
-use crate::error::Error;
+use crate::peer_handle::PeerHandle;
 use crate::WgApiWrapper;
+use crate::{error::Error, peer_handle::SharedBandwidthStorageManager};
 
 pub enum PeerControlRequest {
-    AddPeer(Peer),
-    RemovePeer(Key),
-    QueryPeer(Key),
-    QueryBandwidth(Key),
-}
-
-pub enum PeerControlResponse {
     AddPeer {
-        success: bool,
+        peer: Peer,
+        ticket_validation: bool,
+        response_tx: oneshot::Sender<AddPeerControlResponse>,
     },
     RemovePeer {
-        success: bool,
+        key: Key,
+        response_tx: oneshot::Sender<RemovePeerControlResponse>,
     },
     QueryPeer {
-        success: bool,
-        peer: Option<Peer>,
+        key: Key,
+        response_tx: oneshot::Sender<QueryPeerControlResponse>,
     },
     QueryBandwidth {
-        bandwidth_data: Option<RemainingBandwidthData>,
+        key: Key,
+        response_tx: oneshot::Sender<QueryBandwidthControlResponse>,
     },
 }
 
-pub struct PeerController<St: Storage> {
-    storage: St,
-    request_rx: mpsc::UnboundedReceiver<PeerControlRequest>,
-    response_tx: mpsc::UnboundedSender<PeerControlResponse>,
-    wg_api: Arc<WgApiWrapper>,
-    timeout_check_interval: IntervalStream,
-    active_peers: HashMap<Key, Peer>,
-    suspended_peers: HashMap<Key, Peer>,
-    last_seen_bandwidth: HashMap<Key, u64>,
-    timeout_count: u8,
+pub struct AddPeerControlResponse {
+    pub success: bool,
+    pub client_id: Option<i64>,
 }
 
-impl<St: Storage> PeerController<St> {
+pub struct RemovePeerControlResponse {
+    pub success: bool,
+}
+
+pub struct QueryPeerControlResponse {
+    pub success: bool,
+    pub peer: Option<Peer>,
+}
+
+pub struct QueryBandwidthControlResponse {
+    pub success: bool,
+    pub bandwidth_data: Option<RemainingBandwidthData>,
+}
+
+pub struct PeerController<St: Storage + Clone + 'static> {
+    storage: St,
+    // used to receive commands from individual handles too
+    request_tx: mpsc::Sender<PeerControlRequest>,
+    request_rx: mpsc::Receiver<PeerControlRequest>,
+    wg_api: Arc<WgApiWrapper>,
+    host_information: Arc<RwLock<Host>>,
+    bw_storage_managers: HashMap<Key, Option<SharedBandwidthStorageManager<St>>>,
+    timeout_check_interval: IntervalStream,
+    task_client: nym_task::TaskClient,
+}
+
+impl<St: Storage + Clone + 'static> PeerController<St> {
     pub fn new(
         storage: St,
         wg_api: Arc<WgApiWrapper>,
-        peers: Vec<Peer>,
-        suspended_peers: Vec<Peer>,
-        request_rx: mpsc::UnboundedReceiver<PeerControlRequest>,
-        response_tx: mpsc::UnboundedSender<PeerControlResponse>,
+        initial_host_information: Host,
+        bw_storage_managers: HashMap<Key, Option<SharedBandwidthStorageManager<St>>>,
+        request_tx: mpsc::Sender<PeerControlRequest>,
+        request_rx: mpsc::Receiver<PeerControlRequest>,
+        task_client: nym_task::TaskClient,
     ) -> Self {
         let timeout_check_interval = tokio_stream::wrappers::IntervalStream::new(
             tokio::time::interval(DEFAULT_PEER_TIMEOUT_CHECK),
         );
-        let active_peers: HashMap<Key, Peer> = peers
-            .into_iter()
-            .map(|peer| (peer.public_key.clone(), peer))
-            .collect();
-        let suspended_peers: HashMap<Key, Peer> = suspended_peers
-            .into_iter()
-            .map(|peer| (peer.public_key.clone(), peer))
-            .collect();
-        let last_seen_bandwidth = active_peers
-            .iter()
-            .map(|(k, p)| (k.clone(), p.rx_bytes + p.tx_bytes))
-            .chain(suspended_peers.keys().map(|k| (k.clone(), 0)))
-            .collect();
+        let host_information = Arc::new(RwLock::new(initial_host_information));
+        for (public_key, bandwidth_storage_manager) in bw_storage_managers.iter() {
+            let mut handle = PeerHandle::new(
+                storage.clone(),
+                public_key.clone(),
+                host_information.clone(),
+                bandwidth_storage_manager.clone(),
+                request_tx.clone(),
+                &task_client,
+            );
+            tokio::spawn(async move {
+                if let Err(e) = handle.run().await {
+                    log::error!("Peer handle shut down ungracefully - {e}");
+                }
+            });
+        }
 
         PeerController {
             storage,
             wg_api,
+            host_information,
+            bw_storage_managers,
+            request_tx,
             request_rx,
-            response_tx,
             timeout_check_interval,
-            active_peers,
-            suspended_peers,
-            last_seen_bandwidth,
-            timeout_count: 0,
+            task_client,
         }
     }
 
-    async fn check_stale_peer(
-        &self,
-        peer: &Peer,
-        current_timestamp: SystemTime,
-    ) -> Result<bool, Error> {
-        if let Some(timestamp) = peer.last_handshake {
-            if let Ok(duration_since_handshake) = current_timestamp.duration_since(timestamp) {
-                if duration_since_handshake > DEFAULT_PEER_TIMEOUT {
-                    self.storage
-                        .remove_wireguard_peer(&peer.public_key.to_string())
-                        .await?;
-                    self.wg_api.inner.remove_peer(&peer.public_key)?;
-                    return Ok(true);
-                }
-            }
-        }
-
-        Ok(false)
-    }
-
-    async fn check_suspend_peer(&mut self, peer: &Peer) -> Result<(), Error> {
-        let prev_peer = self
-            .active_peers
-            .get(&peer.public_key)
-            .ok_or(Error::PeerMismatch)?;
-        let data_usage =
-            (peer.rx_bytes + peer.tx_bytes).saturating_sub(prev_peer.rx_bytes + prev_peer.tx_bytes);
-        if data_usage > BANDWIDTH_CAP_PER_DAY {
-            self.storage.insert_wireguard_peer(peer, true).await?;
-            self.wg_api.inner.remove_peer(&peer.public_key)?;
-            self.active_peers
-                .remove_entry(&peer.public_key)
-                .ok_or(Error::PeerMismatch)?;
-            self.suspended_peers
-                .insert(peer.public_key.clone(), peer.clone());
-        } else {
-            // Update peer stored data
-            self.storage.insert_wireguard_peer(peer, false).await?;
-        }
-        Ok(())
-    }
-
-    async fn check_peers(&mut self) -> Result<(), Error> {
-        // Add 10 seconds to cover edge cases. At worst, we give ten free seconds worth of bandwidth
-        // by resetting the bandwidth twice
-        let reset = Utc::now().num_seconds_from_midnight() as u64
-            <= DEFAULT_PEER_TIMEOUT_CHECK.as_secs() + 10;
-
-        if reset {
-            for (_, peer) in self.suspended_peers.drain() {
-                self.wg_api.inner.configure_peer(&peer)?;
-            }
-        }
-        let host = self.wg_api.inner.read_interface_data()?;
-        self.last_seen_bandwidth = host
-            .peers
-            .iter()
-            .map(|(key, peer)| (key.clone(), peer.rx_bytes + peer.tx_bytes))
-            .collect();
-
-        // Do in-memory updates of bandwidth every DEFAULT_PEER_TIMEOUT_CHECK
-        // and storage updates every 5 * DEFAULT_PEER_TIMEOUT_CHECK, because in-memory
-        // is more important for client query preciseness
-        self.timeout_count = self.timeout_count % 5 + 1;
-        if !reset && self.timeout_count < 5 {
-            return Ok(());
-        }
-
-        if reset {
-            self.active_peers = host.peers;
-            for peer in self.active_peers.values() {
-                self.storage.insert_wireguard_peer(peer, false).await?;
-            }
-        } else {
-            let peers = self
+    // Function that should be used for peer insertion, to handle both storage and kernel interaction
+    pub async fn add_peer(&self, peer: &Peer, with_client_id: bool) -> Result<Option<i64>, Error> {
+        let client_id = self
+            .storage
+            .insert_wireguard_peer(peer, with_client_id)
+            .await?;
+        let ret = self.wg_api.inner.configure_peer(peer);
+        if ret.is_err() {
+            // Try to revert the insertion in storage
+            if self
                 .storage
-                .get_all_wireguard_peers()
-                .await?
-                .into_iter()
-                .map(Peer::try_from)
-                .collect::<Result<Vec<_>, _>>()?;
-            let current_timestamp = SystemTime::now();
-            for peer in peers {
-                if !self.check_stale_peer(&peer, current_timestamp).await? {
-                    self.check_suspend_peer(&peer).await?;
-                }
+                .remove_wireguard_peer(&peer.public_key.to_string())
+                .await
+                .is_err()
+            {
+                log::error!("The storage has been corrupted. Wireguard peer {} will persist in storage indefinitely.", peer.public_key);
             }
         }
-
-        Ok(())
+        ret?;
+        Ok(client_id)
     }
 
-    pub async fn run(&mut self, mut task_client: nym_task::TaskClient) {
+    // Function that should be used for peer removal, to handle both storage and kernel interaction
+    pub async fn remove_peer(&mut self, key: &Key) -> Result<(), Error> {
+        self.storage.remove_wireguard_peer(&key.to_string()).await?;
+        self.bw_storage_managers.remove(key);
+        let ret = self.wg_api.inner.remove_peer(key);
+        if ret.is_err() {
+            log::error!("Wireguard peer could not be removed from wireguard kernel module. Process should be restarted so that the interface is reset.");
+        }
+        Ok(ret?)
+    }
+
+    pub async fn generate_bandwidth_manager(
+        storage: St,
+        public_key: &Key,
+    ) -> Result<Option<BandwidthStorageManager<St>>, Error> {
+        if let Some(client_id) = storage
+            .get_wireguard_peer(&public_key.to_string())
+            .await?
+            .ok_or(Error::MissingClientBandwidthEntry)?
+            .client_id
+        {
+            let bandwidth = storage
+                .get_available_bandwidth(client_id)
+                .await?
+                .ok_or(Error::MissingClientBandwidthEntry)?;
+            Ok(Some(BandwidthStorageManager::new(
+                storage,
+                ClientBandwidth::new(bandwidth.into()),
+                client_id,
+                BandwidthFlushingBehaviourConfig::default(),
+                true,
+            )))
+        } else {
+            Ok(None)
+        }
+    }
+
+    async fn handle_add_request(
+        &mut self,
+        peer: &Peer,
+        with_client_id: bool,
+    ) -> Result<Option<i64>, Error> {
+        let client_id = self.add_peer(peer, with_client_id).await?;
+        let bandwidth_storage_manager =
+            Self::generate_bandwidth_manager(self.storage.clone(), &peer.public_key)
+                .await?
+                .map(|bw_m| Arc::new(RwLock::new(bw_m)));
+        let mut handle = PeerHandle::new(
+            self.storage.clone(),
+            peer.public_key.clone(),
+            self.host_information.clone(),
+            bandwidth_storage_manager.clone(),
+            self.request_tx.clone(),
+            &self.task_client,
+        );
+        self.bw_storage_managers
+            .insert(peer.public_key.clone(), bandwidth_storage_manager);
+        tokio::spawn(async move {
+            if let Err(e) = handle.run().await {
+                log::error!("Peer handle shut down ungracefully - {e}");
+            }
+        });
+        Ok(client_id)
+    }
+
+    async fn handle_query_peer(&self, key: &Key) -> Result<Option<Peer>, Error> {
+        Ok(self
+            .storage
+            .get_wireguard_peer(&key.to_string())
+            .await?
+            .map(Peer::try_from)
+            .transpose()?)
+    }
+
+    async fn handle_query_bandwidth(
+        &self,
+        key: &Key,
+    ) -> Result<Option<RemainingBandwidthData>, Error> {
+        let Some(bandwidth_storage_manager) = self.bw_storage_managers.get(key) else {
+            return Ok(None);
+        };
+        let available_bandwidth = if let Some(bandwidth_storage_manager) = bandwidth_storage_manager
+        {
+            bandwidth_storage_manager
+                .read()
+                .await
+                .available_bandwidth()
+                .bytes as u64
+        } else {
+            let peer = self
+                .host_information
+                .read()
+                .await
+                .peers
+                .get(key)
+                .ok_or(Error::PeerMismatch)?
+                .clone();
+            BANDWIDTH_CAP_PER_DAY.saturating_sub(peer.rx_bytes + peer.tx_bytes)
+        };
+
+        Ok(Some(RemainingBandwidthData {
+            available_bandwidth,
+        }))
+    }
+
+    pub async fn run(&mut self) {
         loop {
             tokio::select! {
                 _ = self.timeout_check_interval.next() => {
-                    if let Err(e) = self.check_peers().await {
-                        log::error!("Error while periodically checking peers: {:?}", e);
-                    }
+                    let Ok(host) = self.wg_api.inner.read_interface_data() else {
+                        log::error!("Can't read wireguard kernel data");
+                        continue;
+                    };
+                    *self.host_information.write().await = host;
                 }
-                _ = task_client.recv() => {
+                _ = self.task_client.recv() => {
                     log::trace!("PeerController handler: Received shutdown");
                     break;
                 }
                 msg = self.request_rx.recv() => {
                     match msg {
-                        Some(PeerControlRequest::AddPeer(peer)) => {
-                            if let Err(e) = self.storage.insert_wireguard_peer(&peer, false).await {
-                                log::error!("Could not insert peer into storage: {:?}", e);
-                                self.response_tx.send(PeerControlResponse::AddPeer { success: false }).ok();
-                                continue;
+                        Some(PeerControlRequest::AddPeer { peer, ticket_validation, response_tx }) => {
+                            let ret = self.handle_add_request(&peer, ticket_validation).await;
+                            if let Ok(client_id) = ret {
+                                response_tx.send(AddPeerControlResponse { success: true, client_id }).ok();
+                            } else {
+                                response_tx.send(AddPeerControlResponse { success: false, client_id: None }).ok();
                             }
-                            let success = if let Err(e) = self.wg_api.inner.configure_peer(&peer) {
-                                log::error!("Could not configure peer: {:?}", e);
-                                false
-                            } else {
-                                self.last_seen_bandwidth.insert(peer.public_key.clone(), peer.rx_bytes + peer.tx_bytes);
-                                self.active_peers.insert(peer.public_key.clone(), peer);
-                                true
-                            };
-                            self.response_tx.send(PeerControlResponse::AddPeer { success }).ok();
                         }
-                        Some(PeerControlRequest::RemovePeer(peer_pubkey)) => {
-                            if let Err(e) = self.storage.remove_wireguard_peer(&peer_pubkey.to_string()).await {
-                                log::error!("Could not remove peer from storage: {:?}", e);
-                                self.response_tx.send(PeerControlResponse::RemovePeer { success: false }).ok();
-                                continue;
+                        Some(PeerControlRequest::RemovePeer { key, response_tx }) => {
+                            let success = self.remove_peer(&key).await.is_ok();
+                            response_tx.send(RemovePeerControlResponse { success }).ok();
+                        }
+                        Some(PeerControlRequest::QueryPeer { key, response_tx }) => {
+                            let ret = self.handle_query_peer(&key).await;
+                            if let Ok(peer) = ret {
+                                response_tx.send(QueryPeerControlResponse { success: true, peer }).ok();
+                            } else {
+                                response_tx.send(QueryPeerControlResponse { success: false, peer: None }).ok();
                             }
-                            let success = if let Err(e) = self.wg_api.inner.remove_peer(&peer_pubkey) {
-                                log::error!("Could not remove peer: {:?}", e);
-                                false
-                            } else {
-                                self.active_peers.remove(&peer_pubkey);
-                                self.suspended_peers.remove(&peer_pubkey);
-                                true
-                            };
-                            self.response_tx.send(PeerControlResponse::RemovePeer { success }).ok();
                         }
-                        Some(PeerControlRequest::QueryPeer(peer_pubkey)) => {
-                            let (success, peer) = match self.storage.get_wireguard_peer(&peer_pubkey.to_string()).await {
-                                Err(e) => {
-                                    log::error!("Could not query peer storage {e}");
-                                    (false, None)
-                                },
-                                Ok(None) => (true, None),
-                                Ok(Some(storage_peer)) => {
-                                    match Peer::try_from(storage_peer) {
-                                        Ok(peer) => (true, Some(peer)),
-                                        Err(e) => {
-                                            log::error!("Could not parse storage peer {e}");
-                                            (false, None)
-                                        }
-                                    }
-                                },
-                            };
-                            self.response_tx.send(PeerControlResponse::QueryPeer { success, peer }).ok();
-                        }
-                        Some(PeerControlRequest::QueryBandwidth(peer_pubkey)) => {
-                            let msg = if self.suspended_peers.contains_key(&peer_pubkey) {
-                                PeerControlResponse::QueryBandwidth { bandwidth_data: Some(RemainingBandwidthData{ available_bandwidth: 0, suspended: true }) }
-                            } else if let Some(&consumed_bandwidth) = self.last_seen_bandwidth.get(&peer_pubkey) {
-                                PeerControlResponse::QueryBandwidth { bandwidth_data: Some(RemainingBandwidthData{ available_bandwidth: BANDWIDTH_CAP_PER_DAY - consumed_bandwidth, suspended: false })}
+                        Some(PeerControlRequest::QueryBandwidth { key, response_tx }) => {
+                            let ret = self.handle_query_bandwidth(&key).await;
+                            if let Ok(bandwidth_data) = ret {
+                                response_tx.send(QueryBandwidthControlResponse { success: true, bandwidth_data }).ok();
                             } else {
-                                PeerControlResponse::QueryBandwidth { bandwidth_data: None }
-                            };
-                            self.response_tx.send(msg).ok();
+                                response_tx.send(QueryBandwidthControlResponse { success: false, bandwidth_data: None }).ok();
+                            }
                         }
                         None => {
                             log::trace!("PeerController [main loop]: stopping since channel closed");

common/wireguard/src/peer_handle.rs

@@ -0,0 +1,131 @@
+// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::error::Error;
+use crate::peer_controller::PeerControlRequest;
+use defguard_wireguard_rs::{host::Host, key::Key};
+use futures::channel::oneshot;
+use nym_authenticator_requests::v2::registration::BANDWIDTH_CAP_PER_DAY;
+use nym_credential_verification::bandwidth_storage_manager::BandwidthStorageManager;
+use nym_gateway_storage::models::WireguardPeer;
+use nym_gateway_storage::Storage;
+use nym_task::TaskClient;
+use nym_wireguard_types::DEFAULT_PEER_TIMEOUT_CHECK;
+use std::sync::Arc;
+use tokio::sync::{mpsc, RwLock};
+use tokio_stream::{wrappers::IntervalStream, StreamExt};
+
+pub(crate) type SharedBandwidthStorageManager<St> = Arc<RwLock<BandwidthStorageManager<St>>>;
+
+pub struct PeerHandle<St> {
+    storage: St,
+    public_key: Key,
+    host_information: Arc<RwLock<Host>>,
+    bandwidth_storage_manager: Option<SharedBandwidthStorageManager<St>>,
+    request_tx: mpsc::Sender<PeerControlRequest>,
+    timeout_check_interval: IntervalStream,
+    task_client: TaskClient,
+}
+
+impl<St: Storage + Clone + 'static> PeerHandle<St> {
+    pub fn new(
+        storage: St,
+        public_key: Key,
+        host_information: Arc<RwLock<Host>>,
+        bandwidth_storage_manager: Option<SharedBandwidthStorageManager<St>>,
+        request_tx: mpsc::Sender<PeerControlRequest>,
+        task_client: &TaskClient,
+    ) -> Self {
+        let timeout_check_interval = tokio_stream::wrappers::IntervalStream::new(
+            tokio::time::interval(DEFAULT_PEER_TIMEOUT_CHECK),
+        );
+        let task_client = task_client.fork(format!("peer{public_key}"));
+        PeerHandle {
+            storage,
+            public_key,
+            host_information,
+            bandwidth_storage_manager,
+            request_tx,
+            timeout_check_interval,
+            task_client,
+        }
+    }
+
+    async fn remove_depleted_peer(&self) -> Result<bool, Error> {
+        log::debug!(
+            "Peer {} doesn't have bandwidth anymore, removing it",
+            self.public_key.to_string()
+        );
+        let (response_tx, response_rx) = oneshot::channel();
+        self.request_tx
+            .send(PeerControlRequest::RemovePeer {
+                key: self.public_key.clone(),
+                response_tx,
+            })
+            .await
+            .map_err(|_| Error::Internal("peer controller shut down".to_string()))?;
+        let success = response_rx
+            .await
+            .map_err(|_| Error::Internal("peer controller didn't respond".to_string()))?
+            .success;
+        Ok(success)
+    }
+
+    async fn active_peer(&mut self, storage_peer: WireguardPeer) -> Result<bool, Error> {
+        let kernel_peer = self
+            .host_information
+            .read()
+            .await
+            .peers
+            .get(&self.public_key)
+            .ok_or(Error::PeerMismatch)?
+            .clone();
+        if let Some(bandwidth_manager) = &self.bandwidth_storage_manager {
+            let spent_bandwidth = (kernel_peer.rx_bytes + kernel_peer.tx_bytes)
+                .checked_sub(storage_peer.rx_bytes as u64 + storage_peer.tx_bytes as u64)
+                .ok_or(Error::InconsistentConsumedBytes)?
+                .try_into()
+                .map_err(|_| Error::InconsistentConsumedBytes)?;
+            if bandwidth_manager
+                .write()
+                .await
+                .try_use_bandwidth(spent_bandwidth)
+                .await
+                .is_err()
+            {
+                let success = self.remove_depleted_peer().await?;
+                return Ok(!success);
+            }
+        } else {
+            let spent_bandwidth = kernel_peer.rx_bytes + kernel_peer.tx_bytes;
+            if spent_bandwidth >= BANDWIDTH_CAP_PER_DAY {
+                let success = self.remove_depleted_peer().await?;
+                return Ok(!success);
+            }
+        }
+
+        Ok(true)
+    }
+
+    pub async fn run(&mut self) -> Result<(), Error> {
+        while !self.task_client.is_shutdown() {
+            tokio::select! {
+                _ = self.timeout_check_interval.next() => {
+                    let Some(peer) = self.storage.get_wireguard_peer(&self.public_key.to_string()).await? else {
+                        log::debug!("Peer {:?} not in storage anymore, shutting down handle", self.public_key);
+                        return Ok(());
+                    };
+                    if !self.active_peer(peer).await? {
+                        log::debug!("Peer {:?} doesn't have bandwidth anymore, shutting down handle", self.public_key);
+                        return Ok(());
+                    }
+                }
+
+                _ = self.task_client.recv() => {
+                    log::trace!("PeerHandle: Received shutdown");
+                }
+            }
+        }
+        Ok(())
+    }
+}

gateway/src/node/mod.rs

@@ -246,6 +246,7 @@ impl<St> Gateway<St> {
         &mut self,
         forwarding_channel: MixForwardingSender,
         shutdown: TaskClient,
+        ecash_verifier: Arc<EcashManager<St>>,
     ) -> Result<StartedAuthenticator, Box<dyn std::error::Error + Send + Sync>>
     where
         St: Storage + Clone + 'static,
@@ -256,7 +257,6 @@ impl<St> Gateway<St> {
             .ok_or(GatewayError::UnspecifiedAuthenticatorConfig)?;
         let (router_tx, mut router_rx) = oneshot::channel();
         let (auth_mix_sender, auth_mix_receiver) = mpsc::unbounded();
-        let (peer_response_tx, peer_response_rx) = tokio::sync::mpsc::unbounded_channel();
         let router_shutdown = shutdown.fork("message_router");
         let transceiver = LocalGateway::new(
             *self.identity_keypair.public_key(),
@@ -286,8 +286,8 @@ impl<St> Gateway<St> {
                 opts.config.clone(),
                 wireguard_data.inner.clone(),
                 used_private_network_ips,
-                peer_response_rx,
             )
+            .with_ecash_verifier(ecash_verifier)
             .with_custom_gateway_transceiver(Box::new(transceiver))
             .with_shutdown(shutdown.fork("authenticator"))
             .with_wait_for_gateway(true)
@@ -322,7 +322,6 @@ impl<St> Gateway<St> {
                 all_peers,
                 shutdown,
                 wireguard_data,
-                peer_response_tx,
             )
             .await?;
 
@@ -342,6 +341,7 @@ impl<St> Gateway<St> {
         &self,
         _forwarding_channel: MixForwardingSender,
         _shutdown: TaskClient,
+        _ecash_verifier: Arc<EcashManager<St>>,
     ) -> Result<StartedAuthenticator, Box<dyn std::error::Error + Send + Sync>> {
         todo!("Authenticator is currently only supported on Linux");
     }
@@ -616,14 +616,16 @@ impl<St> Gateway<St> {
                 .maximum_time_between_redemption,
         };
 
-        let ecash_manager = EcashManager::new(
-            handler_config,
-            nyxd_client,
-            self.identity_keypair.public_key().to_bytes(),
-            shutdown.fork("EcashVerifier"),
-            self.storage.clone(),
-        )
-        .await?;
+        let ecash_verifier = Arc::new(
+            EcashManager::new(
+                handler_config,
+                nyxd_client,
+                self.identity_keypair.public_key().to_bytes(),
+                shutdown.fork("EcashVerifier"),
+                self.storage.clone(),
+            )
+            .await?,
+        );
 
         let mix_forwarding_channel = self.start_packet_forwarder(shutdown.fork("PacketForwarder"));
 
@@ -638,7 +640,7 @@ impl<St> Gateway<St> {
             mix_forwarding_channel.clone(),
             active_clients_store.clone(),
             shutdown.fork("websocket::Listener"),
-            Arc::new(ecash_manager),
+            ecash_verifier.clone(),
         );
 
         let nr_request_filter = if self.config.network_requester.enabled {
@@ -670,7 +672,11 @@ impl<St> Gateway<St> {
 
         let _wg_api = if self.wireguard_data.is_some() {
             let embedded_auth = self
-                .start_authenticator(mix_forwarding_channel, shutdown.fork("authenticator"))
+                .start_authenticator(
+                    mix_forwarding_channel,
+                    shutdown.fork("authenticator"),
+                    ecash_verifier,
+                )
                 .await
                 .map_err(|source| GatewayError::AuthenticatorStartError { source })?;
             active_clients_store.insert_embedded(embedded_auth.handle);

nym-node/nym-node-http-api/Cargo.toml

@@ -30,12 +30,13 @@ fastrand = { workspace = true }
 
 nym-crypto = { path = "../../common/crypto", features = ["asymmetric", "rand"] }
 nym-http-api-common = { path = "../../common/http-api-common" }
-nym-node-requests = { path = "../nym-node-requests", default-features = false, features = ["openapi"] }
+nym-node-requests = { path = "../nym-node-requests", default-features = false, features = [
+    "openapi",
+] }
 nym-task = { path = "../../common/task" }
 
 nym-metrics = { path = "../../common/nym-metrics" }
 nym-wireguard = { path = "../../common/wireguard" }
-nym-wireguard-types = { path = "../../common/wireguard-types", features = ["verify"] }
 
 [dev-dependencies]
 base64 = { workspace = true }

nym-node/nym-node-http-api/src/router/api/v1/openapi.rs

@@ -60,9 +60,6 @@ use utoipa_swagger_ui::SwaggerUi;
             api_requests::v1::gateway::models::Wireguard,
             api_requests::v1::gateway::models::ClientInterfaces,
             api_requests::v1::gateway::models::WebSockets,
-            api_requests::v1::gateway::client_interfaces::wireguard::models::ClientMessage,
-            api_requests::v1::gateway::client_interfaces::wireguard::models::InitMessage,
-            api_requests::v1::gateway::client_interfaces::wireguard::models::GatewayClient,
             api_requests::v1::mixnode::models::Mixnode,
             api_requests::v1::network_requester::models::NetworkRequester,
             api_requests::v1::network_requester::exit_policy::models::AddressPolicy,

nym-node/nym-node-requests/Cargo.toml

@@ -12,7 +12,7 @@ license.workspace = true
 
 [dependencies]
 base64 = { workspace = true }
-celes = { workspace = true } # country codes
+celes = { workspace = true }                                               # country codes
 humantime = { workspace = true }
 humantime-serde = { workspace = true }
 schemars = { workspace = true, features = ["preserve_order"] }
@@ -21,7 +21,10 @@ serde_json = { workspace = true }
 time = { workspace = true, features = ["serde", "formatting", "parsing"] }
 thiserror = { workspace = true }
 
-nym-crypto = { path = "../../common/crypto", features = ["asymmetric", "serde"] }
+nym-crypto = { path = "../../common/crypto", features = [
+    "asymmetric",
+    "serde",
+] }
 nym-exit-policy = { path = "../../common/exit-policy" }
 nym-wireguard-types = { path = "../../common/wireguard-types", default-features = false }
 
@@ -33,7 +36,9 @@ nym-http-api-client = { path = "../../common/http-api-client", optional = true }
 
 ## openapi:
 utoipa = { workspace = true, optional = true }
-nym-bin-common = { path = "../../common/bin-common", features = ["bin_info_schema"] }
+nym-bin-common = { path = "../../common/bin-common", features = [
+    "bin_info_schema",
+] }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["full"] }
@@ -44,4 +49,4 @@ nym-crypto = { path = "../../common/crypto", features = ["rand"] }
 [features]
 default = ["client"]
 client = ["nym-http-api-client", "async-trait"]
-openapi = ["utoipa", "nym-bin-common/openapi", "nym-wireguard-types/openapi", "nym-exit-policy/openapi"]
+openapi = ["utoipa", "nym-bin-common/openapi", "nym-exit-policy/openapi"]

nym-node/nym-node-requests/src/api/v1/gateway/client_interfaces/wireguard/models.rs

@@ -1,6 +1,4 @@
 // Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-pub use nym_wireguard_types::{
-    ClientMac, ClientMessage, GatewayClient, InitMessage, Nonce, PeerPublicKey,
-};
+pub use nym_wireguard_types::PeerPublicKey;

nym-node/src/node/mod.rs

@@ -37,7 +37,7 @@ use rand::rngs::OsRng;
 use rand::{CryptoRng, RngCore};
 use std::path::Path;
 use std::sync::Arc;
-use tokio::sync::mpsc::UnboundedReceiver;
+use tokio::sync::mpsc;
 use tracing::{debug, error, info, trace};
 use zeroize::Zeroizing;
 
@@ -274,7 +274,7 @@ impl ExitGatewayData {
 
 pub struct WireguardData {
     inner: WireguardGatewayData,
-    peer_rx: UnboundedReceiver<PeerControlRequest>,
+    peer_rx: mpsc::Receiver<PeerControlRequest>,
 }
 
 impl WireguardData {

service-providers/authenticator/Cargo.toml

@@ -35,11 +35,16 @@ nym-bin-common = { path = "../../common/bin-common", features = [
 ] }
 nym-client-core = { path = "../../common/client-core", features = ["cli"] }
 nym-config = { path = "../../common/config" }
+nym-credentials-interface = { path = "../../common/credentials-interface" }
+nym-credential-verification = { path = "../../common/credential-verification" }
 nym-crypto = { path = "../../common/crypto" }
+nym-gateway-requests = { path = "../../common/gateway-requests" }
+nym-gateway-storage = { path = "../../common/gateway-storage" }
 nym-id = { path = "../../common/nym-id" }
 nym-network-defaults = { path = "../../common/network-defaults" }
 nym-sdk = { path = "../../sdk/rust/nym-sdk" }
 nym-service-providers-common = { path = "../common" }
+nym-service-provider-requests-common = { path = "../../common/service-provider-requests-common" }
 nym-sphinx = { path = "../../common/nymsphinx" }
 nym-task = { path = "../../common/task" }
 nym-types = { path = "../../common/types" }

service-providers/authenticator/src/authenticator.rs

@@ -1,15 +1,16 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use std::{net::IpAddr, path::Path, time::SystemTime};
+use std::{net::IpAddr, path::Path, sync::Arc, time::SystemTime};
 
 use futures::channel::oneshot;
 use ipnetwork::IpNetwork;
 use nym_client_core::{HardcodedTopologyProvider, TopologyProvider};
+use nym_credential_verification::ecash::EcashManager;
+use nym_gateway_storage::Storage;
 use nym_sdk::{mixnet::Recipient, GatewayTransceiver};
 use nym_task::{TaskClient, TaskHandle};
-use nym_wireguard::{peer_controller::PeerControlResponse, WireguardGatewayData};
-use tokio::sync::mpsc::UnboundedReceiver;
+use nym_wireguard::WireguardGatewayData;
 
 use crate::{config::Config, error::AuthenticatorError};
 
@@ -24,39 +25,45 @@ impl OnStartData {
     }
 }
 
-pub struct Authenticator {
+pub struct Authenticator<S> {
     #[allow(unused)]
     config: Config,
     wait_for_gateway: bool,
     custom_topology_provider: Option<Box<dyn TopologyProvider + Send + Sync>>,
     custom_gateway_transceiver: Option<Box<dyn GatewayTransceiver + Send + Sync>>,
     wireguard_gateway_data: WireguardGatewayData,
+    ecash_verifier: Option<Arc<EcashManager<S>>>,
     used_private_network_ips: Vec<IpAddr>,
-    response_rx: UnboundedReceiver<PeerControlResponse>,
     shutdown: Option<TaskClient>,
     on_start: Option<oneshot::Sender<OnStartData>>,
 }
 
-impl Authenticator {
+impl<S: Storage + Clone + 'static> Authenticator<S> {
     pub fn new(
         config: Config,
         wireguard_gateway_data: WireguardGatewayData,
         used_private_network_ips: Vec<IpAddr>,
-        response_rx: UnboundedReceiver<PeerControlResponse>,
     ) -> Self {
         Self {
             config,
             wait_for_gateway: false,
             custom_topology_provider: None,
             custom_gateway_transceiver: None,
+            ecash_verifier: None,
             wireguard_gateway_data,
             used_private_network_ips,
-            response_rx,
             shutdown: None,
             on_start: None,
         }
     }
 
+    #[must_use]
+    #[allow(unused)]
+    pub fn with_ecash_verifier(mut self, ecash_verifier: Arc<EcashManager<S>>) -> Self {
+        self.ecash_verifier = Some(ecash_verifier);
+        self
+    }
+
     #[must_use]
     #[allow(unused)]
     pub fn with_shutdown(mut self, shutdown: TaskClient) -> Self {
@@ -152,9 +159,9 @@ impl Authenticator {
             self.config,
             free_private_network_ips,
             self.wireguard_gateway_data,
-            self.response_rx,
             mixnet_client,
             task_handle,
+            self.ecash_verifier,
         );
 
         log::info!("The address of this client is: {self_address}");

service-providers/authenticator/src/cli/peer_handler.rs

@@ -2,24 +2,21 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use nym_sdk::TaskClient;
-use nym_wireguard::peer_controller::{PeerControlRequest, PeerControlResponse};
+use nym_wireguard::peer_controller::{
+    AddPeerControlResponse, PeerControlRequest, QueryBandwidthControlResponse,
+    QueryPeerControlResponse, RemovePeerControlResponse,
+};
 use tokio::sync::mpsc;
 
 pub struct DummyHandler {
-    peer_rx: mpsc::UnboundedReceiver<PeerControlRequest>,
-    response_tx: mpsc::UnboundedSender<PeerControlResponse>,
+    peer_rx: mpsc::Receiver<PeerControlRequest>,
     task_client: TaskClient,
 }
 
 impl DummyHandler {
-    pub fn new(
-        peer_rx: mpsc::UnboundedReceiver<PeerControlRequest>,
-        response_tx: mpsc::UnboundedSender<PeerControlResponse>,
-        task_client: TaskClient,
-    ) -> Self {
+    pub fn new(peer_rx: mpsc::Receiver<PeerControlRequest>, task_client: TaskClient) -> Self {
         DummyHandler {
             peer_rx,
-            response_tx,
             task_client,
         }
     }
@@ -30,21 +27,21 @@ impl DummyHandler {
                 msg = self.peer_rx.recv() => {
                     if let Some(msg) = msg {
                         match msg {
-                            PeerControlRequest::AddPeer(peer) => {
-                                log::info!("[DUMMY] Adding peer {:?}", peer);
-                                self.response_tx.send(PeerControlResponse::AddPeer { success: true }).ok();
+                            PeerControlRequest::AddPeer { peer, ticket_validation, response_tx } => {
+                                log::info!("[DUMMY] Adding peer {:?} with ticket validation {}", peer, ticket_validation);
+                                response_tx.send(AddPeerControlResponse { success: true, client_id: None }).ok();
                             }
-                            PeerControlRequest::RemovePeer(key) => {
+                            PeerControlRequest::RemovePeer { key, response_tx } => {
                                 log::info!("[DUMMY] Removing peer {:?}", key);
-                                self.response_tx.send(PeerControlResponse::RemovePeer { success: true }).ok();
+                                response_tx.send(RemovePeerControlResponse { success: true }).ok();
                             }
-                            PeerControlRequest::QueryPeer(key) => {
+                            PeerControlRequest::QueryPeer{key, response_tx} => {
                                 log::info!("[DUMMY] Querying peer {:?}", key);
-                                self.response_tx.send(PeerControlResponse::QueryPeer { success: false, peer: None }).ok();
+                                response_tx.send(QueryPeerControlResponse { success: true, peer: None }).ok();
                             }
-                            PeerControlRequest::QueryBandwidth(key) => {
+                            PeerControlRequest::QueryBandwidth{key, response_tx} => {
                                 log::info!("[DUMMY] Querying bandwidth for peer {:?}", key);
-                                self.response_tx.send(PeerControlResponse::QueryBandwidth { bandwidth_data: None }).ok();
+                                response_tx.send(QueryBandwidthControlResponse { success: true, bandwidth_data: None }).ok();
                             }
                         }
                     } else {

service-providers/authenticator/src/cli/run.rs

@@ -11,10 +11,10 @@ use log::error;
 use nym_authenticator::error::AuthenticatorError;
 use nym_client_core::cli_helpers::client_run::CommonClientRunArgs;
 use nym_crypto::asymmetric::x25519::KeyPair;
+use nym_gateway_storage::PersistentStorage;
 use nym_task::TaskHandle;
 use nym_wireguard::WireguardGatewayData;
 use rand::rngs::OsRng;
-use tokio::sync::mpsc::unbounded_channel;
 
 #[allow(clippy::struct_excessive_bools)]
 #[derive(Args, Clone)]
@@ -49,13 +49,16 @@ pub(crate) async fn execute(args: &Run) -> Result<(), AuthenticatorError> {
         Arc::new(KeyPair::new(&mut OsRng)),
     );
     let task_handler = TaskHandle::default();
-    let (response_tx, response_rx) = unbounded_channel();
-    let handler = DummyHandler::new(peer_rx, response_tx, task_handler.fork("peer-handler"));
+    let handler = DummyHandler::new(peer_rx, task_handler.fork("peer-handler"));
     tokio::spawn(async move {
         handler.run().await;
     });
-    let mut server =
-        nym_authenticator::Authenticator::new(config, wireguard_gateway_data, vec![], response_rx);
+
+    let mut server = nym_authenticator::Authenticator::<PersistentStorage>::new(
+        config,
+        wireguard_gateway_data,
+        vec![],
+    );
     if let Some(custom_mixnet) = &args.common_args.custom_mixnet {
         server = server.with_stored_topology(custom_mixnet)?
     }

service-providers/authenticator/src/error.rs

@@ -14,11 +14,14 @@ pub enum AuthenticatorError {
     #[error("failed to validate the loaded config")]
     ConfigValidationFailure,
 
+    #[error("{0}")]
+    CredentialVerificationError(#[from] nym_credential_verification::Error),
+
     #[error("the entity wrapping the network requester has disconnected")]
     DisconnectedParent,
 
-    #[error("received empty packet")]
-    EmptyPacket,
+    #[error("received too short packet")]
+    ShortPacket,
 
     #[error("failed local version check, client and config mismatch")]
     FailedLocalVersionCheck,
@@ -41,9 +44,15 @@ pub enum AuthenticatorError {
     #[error("failed to setup mixnet client: {source}")]
     FailedToSetupMixnetClient { source: nym_sdk::Error },
 
+    #[error("{0}")]
+    GatewayStorageError(#[from] nym_gateway_storage::error::StorageError),
+
     #[error("internal error: {0}")]
     InternalError(String),
 
+    #[error("received packet has an invalid type: {0}")]
+    InvalidPacketType(u8),
+
     #[error("received packet has an invalid version: {0}")]
     InvalidPacketVersion(u8),
 

service-providers/authenticator/src/mixnet_listener.rs

@@ -8,22 +8,37 @@ use std::{
 
 use crate::{error::AuthenticatorError, peer_manager::PeerManager};
 use futures::StreamExt;
-use nym_authenticator_requests::v1::{
+use log::warn;
+use nym_authenticator_requests::v2::{
     self,
-    request::{AuthenticatorRequest, AuthenticatorRequestData},
-    response::AuthenticatorResponse,
+    registration::{
+        FinalMessage, GatewayClient, InitMessage, PendingRegistrations, PrivateIPs,
+        RegistrationData, RegistredData,
+    },
 };
+use nym_authenticator_requests::{
+    v1,
+    v2::{
+        request::{AuthenticatorRequest, AuthenticatorRequestData},
+        response::AuthenticatorResponse,
+    },
+};
+use nym_credential_verification::{
+    bandwidth_storage_manager::BandwidthStorageManager, ecash::EcashManager,
+    BandwidthFlushingBehaviourConfig, ClientBandwidth, CredentialVerifier,
+};
+use nym_credentials_interface::CredentialSpendingData;
 use nym_crypto::asymmetric::x25519::KeyPair;
+use nym_gateway_requests::models::CredentialSpendingRequest;
+use nym_gateway_storage::Storage;
 use nym_sdk::mixnet::{InputMessage, MixnetMessageSender, Recipient, TransmissionLane};
+use nym_service_provider_requests_common::ServiceProviderType;
 use nym_sphinx::receiver::ReconstructedMessage;
 use nym_task::TaskHandle;
-use nym_wireguard::{peer_controller::PeerControlResponse, WireguardGatewayData};
-use nym_wireguard_types::{
-    registration::{PendingRegistrations, PrivateIPs, RegistrationData, RegistredData},
-    GatewayClient, InitMessage, PeerPublicKey,
-};
+use nym_wireguard::WireguardGatewayData;
+use nym_wireguard_types::PeerPublicKey;
 use rand::{prelude::IteratorRandom, thread_rng};
-use tokio::sync::{mpsc::UnboundedReceiver, RwLock};
+use tokio::sync::RwLock;
 use tokio_stream::wrappers::IntervalStream;
 
 use crate::{config::Config, error::*};
@@ -45,7 +60,7 @@ impl RegistredAndFree {
     }
 }
 
-pub(crate) struct MixnetListener {
+pub(crate) struct MixnetListener<S> {
     // The configuration for the mixnet listener
     pub(crate) config: Config,
 
@@ -60,17 +75,19 @@ pub(crate) struct MixnetListener {
 
     pub(crate) peer_manager: PeerManager,
 
+    pub(crate) ecash_verifier: Option<Arc<EcashManager<S>>>,
+
     pub(crate) timeout_check_interval: IntervalStream,
 }
 
-impl MixnetListener {
+impl<S: Storage + Clone + 'static> MixnetListener<S> {
     pub fn new(
         config: Config,
         free_private_network_ips: PrivateIPs,
         wireguard_gateway_data: WireguardGatewayData,
-        response_rx: UnboundedReceiver<PeerControlResponse>,
         mixnet_client: nym_sdk::mixnet::MixnetClient,
         task_handle: TaskHandle,
+        ecash_verifier: Option<Arc<EcashManager<S>>>,
     ) -> Self {
         let timeout_check_interval =
             IntervalStream::new(tokio::time::interval(DEFAULT_REGISTRATION_TIMEOUT_CHECK));
@@ -79,7 +96,8 @@ impl MixnetListener {
             mixnet_client,
             task_handle,
             registred_and_free: RwLock::new(RegistredAndFree::new(free_private_network_ips)),
-            peer_manager: PeerManager::new(wireguard_gateway_data, response_rx),
+            peer_manager: PeerManager::new(wireguard_gateway_data),
+            ecash_verifier,
             timeout_check_interval,
         }
     }
@@ -132,7 +150,7 @@ impl MixnetListener {
         request_id: u64,
         reply_to: Recipient,
     ) -> AuthenticatorHandleResult {
-        let remote_public = init_message.pub_key();
+        let remote_public = init_message.pub_key;
         let nonce: u64 = fastrand::u64(..);
         if let Some(registration_data) = self
             .registred_and_free
@@ -165,6 +183,7 @@ impl MixnetListener {
                 request_id,
             ));
         }
+
         let mut registred_and_free = self.registred_and_free.write().await;
         let private_ip_ref = registred_and_free
             .free_private_network_ips
@@ -198,38 +217,102 @@ impl MixnetListener {
 
     async fn on_final_request(
         &mut self,
-        gateway_client: GatewayClient,
+        final_message: FinalMessage,
         request_id: u64,
         reply_to: Recipient,
     ) -> AuthenticatorHandleResult {
         let mut registred_and_free = self.registred_and_free.write().await;
         let registration_data = registred_and_free
             .registration_in_progres
-            .get(&gateway_client.pub_key())
+            .get(&final_message.gateway_client.pub_key())
             .ok_or(AuthenticatorError::RegistrationNotInProgress)?
             .clone();
 
-        if gateway_client
+        if final_message
+            .gateway_client
             .verify(self.keypair().private_key(), registration_data.nonce)
-            .is_ok()
+            .is_err()
         {
-            self.peer_manager.add_peer(&gateway_client).await?;
-            registred_and_free
-                .registration_in_progres
-                .remove(&gateway_client.pub_key());
-
-            Ok(AuthenticatorResponse::new_registered(
-                RegistredData {
-                    pub_key: registration_data.gateway_data.pub_key,
-                    private_ip: registration_data.gateway_data.private_ip,
-                    wg_port: registration_data.wg_port,
-                },
-                reply_to,
-                request_id,
-            ))
-        } else {
-            Err(AuthenticatorError::MacVerificationFailure)
+            return Err(AuthenticatorError::MacVerificationFailure);
         }
+
+        // If gateway does ecash verification and client sends a credential, we do the additional
+        // credential verification. Later this will become mandatory.
+        if let (Some(ecash_verifier), Some(credential)) = (
+            self.ecash_verifier.clone(),
+            final_message.credential.clone(),
+        ) {
+            let client_id = self
+                .peer_manager
+                .add_peer(&final_message.gateway_client, true)
+                .await?
+                .ok_or(AuthenticatorError::InternalError(
+                    "peer with ticket shouldn't have been used before without a ticket".to_string(),
+                ))?;
+
+            if let Err(e) =
+                Self::credential_verification(ecash_verifier, credential, client_id).await
+            {
+                self.peer_manager
+                    .remove_peer(&final_message.gateway_client)
+                    .await
+                    .inspect_err(|err| {
+                        warn!(
+                            "Could not revert adding peer {} on credential verification {err}",
+                            final_message.gateway_client.pub_key()
+                        )
+                    })?;
+                return Err(e);
+            }
+        } else {
+            self.peer_manager
+                .add_peer(&final_message.gateway_client, false)
+                .await?;
+        }
+        registred_and_free
+            .registration_in_progres
+            .remove(&final_message.gateway_client.pub_key());
+
+        Ok(AuthenticatorResponse::new_registered(
+            RegistredData {
+                pub_key: registration_data.gateway_data.pub_key,
+                private_ip: registration_data.gateway_data.private_ip,
+                wg_port: registration_data.wg_port,
+            },
+            reply_to,
+            request_id,
+        ))
+    }
+
+    async fn credential_verification(
+        ecash_verifier: Arc<EcashManager<S>>,
+        credential: CredentialSpendingData,
+        client_id: i64,
+    ) -> Result<i64> {
+        ecash_verifier
+            .storage()
+            .create_bandwidth_entry(client_id)
+            .await?;
+        let bandwidth = ecash_verifier
+            .storage()
+            .get_available_bandwidth(client_id)
+            .await?
+            .ok_or(AuthenticatorError::InternalError(
+                "bandwidth entry should have just been created".to_string(),
+            ))?;
+        let client_bandwidth = ClientBandwidth::new(bandwidth.into());
+        let mut verifier = CredentialVerifier::new(
+            CredentialSpendingRequest::new(credential),
+            ecash_verifier.clone(),
+            BandwidthStorageManager::new(
+                ecash_verifier.storage().clone(),
+                client_bandwidth,
+                client_id,
+                BandwidthFlushingBehaviourConfig::default(),
+                true,
+            ),
+        );
+        Ok(verifier.verify().await?)
     }
 
     async fn on_query_bandwidth_request(
@@ -267,8 +350,8 @@ impl MixnetListener {
                 self.on_initial_request(init_msg, request.request_id, request.reply_to)
                     .await
             }
-            AuthenticatorRequestData::Final(client) => {
-                self.on_final_request(client, request.request_id, request.reply_to)
+            AuthenticatorRequestData::Final(final_msg) => {
+                self.on_final_request(*final_msg, request.request_id, request.reply_to)
                     .await
             }
             AuthenticatorRequestData::QueryBandwidth(peer_public_key) => {
@@ -350,16 +433,28 @@ impl MixnetListener {
 fn deserialize_request(reconstructed: &ReconstructedMessage) -> Result<AuthenticatorRequest> {
     let request_version = *reconstructed
         .message
-        .first()
-        .ok_or(AuthenticatorError::EmptyPacket)?;
+        .first_chunk::<2>()
+        .ok_or(AuthenticatorError::ShortPacket)?;
 
     // Check version of the request and convert to the latest version if necessary
     match request_version {
-        1 => v1::request::AuthenticatorRequest::from_reconstructed_message(reconstructed)
-            .map_err(|err| AuthenticatorError::FailedToDeserializeTaggedPacket { source: err }),
-        _ => {
-            log::info!("Received packet with invalid version: v{request_version}");
-            Err(AuthenticatorError::InvalidPacketVersion(request_version))
+        [1, _] => v1::request::AuthenticatorRequest::from_reconstructed_message(reconstructed)
+            .map_err(|err| AuthenticatorError::FailedToDeserializeTaggedPacket { source: err })
+            .map(Into::into),
+        [2, request_type] => {
+            if request_type == ServiceProviderType::Authenticator as u8 {
+                v2::request::AuthenticatorRequest::from_reconstructed_message(reconstructed)
+                    .map_err(|err| AuthenticatorError::FailedToDeserializeTaggedPacket {
+                        source: err,
+                    })
+                    .map(Into::into)
+            } else {
+                Err(AuthenticatorError::InvalidPacketType(request_type))
+            }
+        }
+        [version, _] => {
+            log::info!("Received packet with invalid version: v{version}");
+            Err(AuthenticatorError::InvalidPacketVersion(version))
         }
     }
 }

service-providers/authenticator/src/peer_manager.rs

@@ -3,82 +3,74 @@
 
 use crate::error::*;
 use defguard_wireguard_rs::{host::Peer, key::Key, net::IpAddrMask};
+use futures::channel::oneshot;
+use nym_authenticator_requests::v2::registration::{GatewayClient, RemainingBandwidthData};
 use nym_wireguard::{
-    peer_controller::{PeerControlRequest, PeerControlResponse},
+    peer_controller::{
+        AddPeerControlResponse, PeerControlRequest, QueryBandwidthControlResponse,
+        QueryPeerControlResponse, RemovePeerControlResponse,
+    },
     WireguardGatewayData,
 };
-use nym_wireguard_types::{registration::RemainingBandwidthData, GatewayClient, PeerPublicKey};
-use tokio::sync::mpsc::UnboundedReceiver;
+use nym_wireguard_types::PeerPublicKey;
 
 pub struct PeerManager {
     pub(crate) wireguard_gateway_data: WireguardGatewayData,
-
-    pub(crate) response_rx: UnboundedReceiver<PeerControlResponse>,
 }
 
 impl PeerManager {
-    pub fn new(
-        wireguard_gateway_data: WireguardGatewayData,
-        response_rx: UnboundedReceiver<PeerControlResponse>,
-    ) -> Self {
+    pub fn new(wireguard_gateway_data: WireguardGatewayData) -> Self {
         PeerManager {
             wireguard_gateway_data,
-            response_rx,
         }
     }
-    pub async fn add_peer(&mut self, client: &GatewayClient) -> Result<()> {
+    pub async fn add_peer(
+        &mut self,
+        client: &GatewayClient,
+        ticket_validation: bool,
+    ) -> Result<Option<i64>> {
         let mut peer = Peer::new(Key::new(client.pub_key.to_bytes()));
+        let (response_tx, response_rx) = oneshot::channel();
         peer.allowed_ips
             .push(IpAddrMask::new(client.private_ip, 32));
-        let msg = PeerControlRequest::AddPeer(peer);
+        let msg = PeerControlRequest::AddPeer {
+            peer,
+            ticket_validation,
+            response_tx,
+        };
         self.wireguard_gateway_data
             .peer_tx()
             .send(msg)
+            .await
             .map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
 
-        let PeerControlResponse::AddPeer { success } =
-            self.response_rx
-                .recv()
-                .await
-                .ok_or(AuthenticatorError::InternalError(
-                    "no response for add peer".to_string(),
-                ))?
-        else {
-            return Err(AuthenticatorError::InternalError(
-                "unexpected response type".to_string(),
-            ));
-        };
+        let AddPeerControlResponse { success, client_id } = response_rx.await.map_err(|_| {
+            AuthenticatorError::InternalError("no response for add peer".to_string())
+        })?;
         if !success {
             return Err(AuthenticatorError::InternalError(
                 "adding peer could not be performed".to_string(),
             ));
         }
-        Ok(())
+        Ok(client_id)
     }
 
-    pub async fn _remove_peer(&mut self, client: &GatewayClient) -> Result<()> {
+    pub async fn remove_peer(&mut self, client: &GatewayClient) -> Result<()> {
         let key = Key::new(client.pub_key().to_bytes());
-        let msg = PeerControlRequest::RemovePeer(key);
+        let (response_tx, response_rx) = oneshot::channel();
+        let msg = PeerControlRequest::RemovePeer { key, response_tx };
         self.wireguard_gateway_data
             .peer_tx()
             .send(msg)
+            .await
             .map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
 
-        let PeerControlResponse::RemovePeer { success } =
-            self.response_rx
-                .recv()
-                .await
-                .ok_or(AuthenticatorError::InternalError(
-                    "no response for add peer".to_string(),
-                ))?
-        else {
-            return Err(AuthenticatorError::InternalError(
-                "unexpected response type".to_string(),
-            ));
-        };
+        let RemovePeerControlResponse { success } = response_rx.await.map_err(|_| {
+            AuthenticatorError::InternalError("no response for add peer".to_string())
+        })?;
         if !success {
             return Err(AuthenticatorError::InternalError(
-                "adding peer could not be performed".to_string(),
+                "removing peer could not be performed".to_string(),
             ));
         }
         Ok(())
@@ -86,24 +78,17 @@ impl PeerManager {
 
     pub async fn query_peer(&mut self, public_key: PeerPublicKey) -> Result<Option<Peer>> {
         let key = Key::new(public_key.to_bytes());
-        let msg = PeerControlRequest::QueryPeer(key);
+        let (response_tx, response_rx) = oneshot::channel();
+        let msg = PeerControlRequest::QueryPeer { key, response_tx };
         self.wireguard_gateway_data
             .peer_tx()
             .send(msg)
+            .await
             .map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
 
-        let PeerControlResponse::QueryPeer { success, peer } = self
-            .response_rx
-            .recv()
-            .await
-            .ok_or(AuthenticatorError::InternalError(
-                "no response for query peer".to_string(),
-            ))?
-        else {
-            return Err(AuthenticatorError::InternalError(
-                "unexpected response type".to_string(),
-            ));
-        };
+        let QueryPeerControlResponse { success, peer } = response_rx.await.map_err(|_| {
+            AuthenticatorError::InternalError("no response for query peer".to_string())
+        })?;
         if !success {
             return Err(AuthenticatorError::InternalError(
                 "querying peer could not be performed".to_string(),
@@ -117,24 +102,25 @@ impl PeerManager {
         peer_public_key: PeerPublicKey,
     ) -> Result<Option<RemainingBandwidthData>> {
         let key = Key::new(peer_public_key.to_bytes());
-        let msg = PeerControlRequest::QueryBandwidth(key);
+        let (response_tx, response_rx) = oneshot::channel();
+        let msg = PeerControlRequest::QueryBandwidth { key, response_tx };
         self.wireguard_gateway_data
             .peer_tx()
             .send(msg)
+            .await
             .map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
 
-        let PeerControlResponse::QueryBandwidth { bandwidth_data } = self
-            .response_rx
-            .recv()
+        let QueryBandwidthControlResponse {
+            success,
+            bandwidth_data,
+        } = response_rx
             .await
-            .ok_or(AuthenticatorError::InternalError(
-                "no response for query".to_string(),
-            ))?
-        else {
+            .map_err(|_| AuthenticatorError::InternalError("no response for query".to_string()))?;
+        if !success {
             return Err(AuthenticatorError::InternalError(
-                "unexpected response type".to_string(),
+                "querying bandwidth could not be performed".to_string(),
             ));
-        };
+        }
         Ok(bandwidth_data)
     }
 }

service-providers/ip-packet-router/src/connected_client_handler.rs

@@ -22,9 +22,6 @@ pub(crate) struct ConnectedClientHandler {
     // The address of the client that this handler is connected to
     nym_address: Recipient,
 
-    // The number of hops the packet should take before reaching the client
-    mix_hops: Option<u8>,
-
     // Channel to receive packets from the tun_listener
     forward_from_tun_rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
 
@@ -47,7 +44,6 @@ pub(crate) struct ConnectedClientHandler {
 impl ConnectedClientHandler {
     pub(crate) fn start(
         reply_to: Recipient,
-        reply_to_hops: Option<u8>,
         buffer_timeout: std::time::Duration,
         client_version: SupportedClientVersion,
         mixnet_client_sender: nym_sdk::mixnet::MixnetClientSender,
@@ -67,7 +63,6 @@ impl ConnectedClientHandler {
 
         let connected_client_handler = ConnectedClientHandler {
             nym_address: reply_to,
-            mix_hops: reply_to_hops,
             forward_from_tun_rx,
             mixnet_client_sender,
             close_rx,
@@ -98,7 +93,7 @@ impl ConnectedClientHandler {
         }
         .map_err(|err| IpPacketRouterError::FailedToSerializeResponsePacket { source: err })?;
 
-        let input_message = create_input_message(self.nym_address, response_packet, self.mix_hops);
+        let input_message = create_input_message(self.nym_address, response_packet);
 
         self.mixnet_client_sender
             .send(input_message)

service-providers/ip-packet-router/src/mixnet_listener.rs

@@ -4,6 +4,7 @@ use std::{collections::HashMap, net::SocketAddr};
 
 use bytes::{Bytes, BytesMut};
 use futures::StreamExt;
+use nym_ip_packet_requests::v7::request::{HealthRequest, PingRequest};
 use nym_ip_packet_requests::v7::response::{
     DynamicConnectFailureReason, InfoLevel, InfoResponseReply, StaticConnectFailureReason,
 };
@@ -95,6 +96,33 @@ impl ConnectedClients {
             })
     }
 
+    fn remove_client_with_nym_address(&self, nym_address: &Recipient) {
+        // Remove the client from both the ipv4 and ipv6 maps
+        let ipv4 = self.clients_ipv4_mapping.iter().find_map(|(ip, client)| {
+            if client.nym_address == *nym_address {
+                Some(ip)
+            } else {
+                None
+            }
+        });
+        let client_ipv4 = self.clients_ipv4_mapping.remove(ipv4);
+
+        let ipv6 = self.clients_ipv6_mapping.iter().find_map(|(ip, client)| {
+            if client.nym_address == *nym_address {
+                Some(ip)
+            } else {
+                None
+            }
+        });
+        let client_ipv6 = self.clients_ipv6_mapping.remove(ipv6);
+
+        // These two should be the same
+        if let Some(client) = client_ipv4 {
+            // client.update_activity()
+        }
+    }
+
+    #[allow(dead_code)]
     fn lookup_client_from_nym_address(&self, nym_address: &Recipient) -> Option<&ConnectedClient> {
         self.clients_ipv4_mapping
             .iter()
@@ -111,7 +139,7 @@ impl ConnectedClients {
         &mut self,
         ips: IpPair,
         nym_address: Recipient,
-        mix_hops: Option<u8>,
+        client_version: SupportedClientVersion,
         forward_from_tun_tx: tokio::sync::mpsc::UnboundedSender<Vec<u8>>,
         close_tx: tokio::sync::oneshot::Sender<()>,
         handle: tokio::task::JoinHandle<()>,
@@ -121,8 +149,8 @@ impl ConnectedClients {
         let client = ConnectedClient {
             nym_address,
             ipv6: ips.ipv6,
-            mix_hops,
             last_activity: Arc::new(RwLock::new(std::time::Instant::now())),
+            client_version,
             _close_tx: Arc::new(CloseTx {
                 nym_address,
                 inner: Some(close_tx),
@@ -155,7 +183,7 @@ impl ConnectedClients {
     }
 
     // Identify connected client handlers that have stopped without being told to stop
-    fn get_finished_client_handlers(&mut self) -> Vec<(IpPair, Recipient)> {
+    fn get_finished_client_handlers(&mut self) -> Vec<(IpPair, Recipient, SupportedClientVersion)> {
         self.clients_ipv4_mapping
             .iter_mut()
             .filter_map(|(ip, connected_client)| {
@@ -163,6 +191,7 @@ impl ConnectedClients {
                     Some((
                         IpPair::new(*ip, connected_client.ipv6),
                         connected_client.nym_address,
+                        connected_client.client_version,
                     ))
                 } else {
                     None
@@ -171,7 +200,7 @@ impl ConnectedClients {
             .collect()
     }
 
-    async fn get_inactive_clients(&mut self) -> Vec<(IpPair, Recipient)> {
+    async fn get_inactive_clients(&mut self) -> Vec<(IpPair, Recipient, SupportedClientVersion)> {
         let now = std::time::Instant::now();
         let mut ret = vec![];
         for (ip, connected_client) in self.clients_ipv4_mapping.iter() {
@@ -181,37 +210,41 @@ impl ConnectedClients {
                 ret.push((
                     IpPair::new(*ip, connected_client.ipv6),
                     connected_client.nym_address,
+                    connected_client.client_version,
                 ))
             }
         }
         ret
     }
 
-    fn disconnect_stopped_client_handlers(&mut self, stopped_clients: Vec<(IpPair, Recipient)>) {
-        for (ips, _) in &stopped_clients {
+    fn disconnect(&mut self, ips: &IpPair) {
+        self.clients_ipv4_mapping.remove(&ips.ipv4);
+        self.clients_ipv6_mapping.remove(&ips.ipv6);
+        self.tun_listener_connected_client_tx
+            .send(ConnectedClientEvent::Disconnect(DisconnectEvent(*ips)))
+            .inspect_err(|err| {
+                log::error!("Failed to send disconnect event: {err}");
+            })
+            .ok();
+    }
+
+    fn disconnect_stopped_client_handlers(
+        &mut self,
+        stopped_clients: Vec<(IpPair, Recipient, SupportedClientVersion)>,
+    ) {
+        for (ips, _, _) in &stopped_clients {
             log::info!("Disconnect stopped client: {ips}");
-            self.clients_ipv4_mapping.remove(&ips.ipv4);
-            self.clients_ipv6_mapping.remove(&ips.ipv6);
-            self.tun_listener_connected_client_tx
-                .send(ConnectedClientEvent::Disconnect(DisconnectEvent(*ips)))
-                .tap_err(|err| {
-                    log::error!("Failed to send disconnect event: {err}");
-                })
-                .ok();
+            self.disconnect(ips);
         }
     }
 
-    fn disconnect_inactive_clients(&mut self, inactive_clients: Vec<(IpPair, Recipient)>) {
-        for (ips, _) in &inactive_clients {
+    fn disconnect_inactive_clients(
+        &mut self,
+        inactive_clients: Vec<(IpPair, Recipient, SupportedClientVersion)>,
+    ) {
+        for (ips, _, _) in &inactive_clients {
             log::info!("Disconnect inactive client: {ips}");
-            self.clients_ipv4_mapping.remove(&ips.ipv4);
-            self.clients_ipv6_mapping.remove(&ips.ipv6);
-            self.tun_listener_connected_client_tx
-                .send(ConnectedClientEvent::Disconnect(DisconnectEvent(*ips)))
-                .tap_err(|err| {
-                    log::error!("Failed to send disconnect event: {err}");
-                })
-                .ok();
+            self.disconnect(ips);
         }
     }
 
@@ -236,12 +269,12 @@ pub(crate) struct ConnectedClient {
     // The assigned IPv6 address of this client
     pub(crate) ipv6: Ipv6Addr,
 
-    // Number of mix node hops that the client has requested to use
-    pub(crate) mix_hops: Option<u8>,
-
     // Keep track of last activity so we can disconnect inactive clients
     pub(crate) last_activity: Arc<RwLock<std::time::Instant>>,
 
+    // The version of the client, since we need to know this to send the correct response
+    pub(crate) client_version: SupportedClientVersion,
+
     pub(crate) _close_tx: Arc<CloseTx>,
 
     // Handle for the connected client handler
@@ -378,6 +411,56 @@ impl Response {
         }
     }
 
+    fn new_pong(
+        request_id: u64,
+        reply_to: Recipient,
+        client_version: SupportedClientVersion,
+    ) -> Self {
+        match client_version {
+            SupportedClientVersion::V6 => Response::V6(v6::response::IpPacketResponse::new_pong(
+                request_id, reply_to,
+            )),
+            SupportedClientVersion::V7 => Response::V7(v7::response::IpPacketResponse::new_pong(
+                request_id, reply_to,
+            )),
+        }
+    }
+
+    fn new_health_response(
+        request_id: u64,
+        reply_to: Recipient,
+        build_info: nym_bin_common::build_information::BinaryBuildInformationOwned,
+        client_version: SupportedClientVersion,
+    ) -> Self {
+        match client_version {
+            SupportedClientVersion::V6 => {
+                Response::V6(v6::response::IpPacketResponse::new_health_response(
+                    request_id, reply_to, build_info, None,
+                ))
+            }
+            SupportedClientVersion::V7 => {
+                Response::V7(v7::response::IpPacketResponse::new_health_response(
+                    request_id, reply_to, build_info, None,
+                ))
+            }
+        }
+    }
+
+    fn new_unrequested_disconnect(
+        reply_to: Recipient,
+        reason: v7::response::UnrequestedDisconnectReason,
+        client_version: SupportedClientVersion,
+    ) -> Self {
+        match client_version {
+            SupportedClientVersion::V6 => Response::V6(
+                v6::response::IpPacketResponse::new_unrequested_disconnect(reply_to, reason.into()),
+            ),
+            SupportedClientVersion::V7 => Response::V7(
+                v7::response::IpPacketResponse::new_unrequested_disconnect(reply_to, reason),
+            ),
+        }
+    }
+
     fn to_bytes(&self) -> Result<Vec<u8>> {
         match self {
             Response::V6(response) => response.to_bytes(),
@@ -429,10 +512,8 @@ impl MixnetListener {
         let request_id = connect_request.request_id;
         let requested_ips = connect_request.ips;
         let reply_to = connect_request.reply_to;
-        let reply_to_hops = connect_request.reply_to_hops;
         // TODO: add to connect request
         let buffer_timeout = nym_ip_packet_requests::codec::BUFFER_TIMEOUT;
-        // TODO: ignoring reply_to_avg_mix_delays for now
 
         // Check that the IP is available in the set of connected clients
         let is_ip_taken = self.connected_clients.is_ip_connected(&requested_ips);
@@ -464,7 +545,6 @@ impl MixnetListener {
                 let (forward_from_tun_tx, close_tx, handle) =
                     connected_client_handler::ConnectedClientHandler::start(
                         reply_to,
-                        reply_to_hops,
                         buffer_timeout,
                         client_version,
                         self.mixnet_client.split_sender(),
@@ -474,7 +554,7 @@ impl MixnetListener {
                 self.connected_clients.connect(
                     requested_ips,
                     reply_to,
-                    reply_to_hops,
+                    client_version,
                     forward_from_tun_tx,
                     close_tx,
                     handle,
@@ -518,10 +598,8 @@ impl MixnetListener {
 
         let request_id = connect_request.request_id;
         let reply_to = connect_request.reply_to;
-        let reply_to_hops = connect_request.reply_to_hops;
         // TODO: add to connect request
         let buffer_timeout = nym_ip_packet_requests::codec::BUFFER_TIMEOUT;
-        // TODO: ignoring reply_to_avg_mix_delays for now
 
         // Check if it's the same client connecting again, then we just reuse the same IP
         // TODO: this is problematic. Until we sign connect requests this means you can spam people
@@ -559,7 +637,6 @@ impl MixnetListener {
         let (forward_from_tun_tx, close_tx, handle) =
             connected_client_handler::ConnectedClientHandler::start(
                 reply_to,
-                reply_to_hops,
                 buffer_timeout,
                 client_version,
                 self.mixnet_client.split_sender(),
@@ -569,7 +646,7 @@ impl MixnetListener {
         self.connected_clients.connect(
             new_ips,
             reply_to,
-            reply_to_hops,
+            client_version,
             forward_from_tun_tx,
             close_tx,
             handle,
@@ -584,11 +661,16 @@ impl MixnetListener {
 
     fn on_disconnect_request(
         &self,
-        _disconnect_request: DisconnectRequest,
-        _client_version: SupportedClientVersion,
+        disconnect_request: DisconnectRequest,
+        client_version: SupportedClientVersion,
     ) -> PacketHandleResult {
-        log::info!("Received disconnect request: not implemented, dropping");
-        Ok(None)
+        log::info!("Received disconnect request");
+
+        let request_id = disconnect_request.request_id;
+        let reply_to = disconnect_request.reply_to;
+
+        let ips = self.connected_clients.lookup_ip_from_nym_address(&reply_to);
+        self.connected_clients.disconnect(&ips);
     }
 
     async fn handle_packet(
@@ -664,6 +746,48 @@ impl MixnetListener {
         Ok(responses)
     }
 
+    fn on_ping_request(
+        &self,
+        ping_request: PingRequest,
+        client_version: SupportedClientVersion,
+    ) -> PacketHandleResult {
+        log::info!(
+            "Received ping request from {sender_address}",
+            sender_address = ping_request.reply_to
+        );
+
+        let reply_to = ping_request.reply_to;
+        let request_id = ping_request.request_id;
+
+        Ok(Some(Response::new_pong(
+            request_id,
+            reply_to,
+            client_version,
+        )))
+    }
+
+    fn on_health_request(
+        &self,
+        health_request: HealthRequest,
+        client_version: SupportedClientVersion,
+    ) -> PacketHandleResult {
+        log::info!(
+            "Received health request from {sender_address}",
+            sender_address = health_request.reply_to
+        );
+
+        let reply_to = health_request.reply_to;
+        let request_id = health_request.request_id;
+        let build_info = nym_bin_common::bin_info_owned!();
+
+        Ok(Some(Response::new_health_response(
+            request_id,
+            reply_to,
+            build_info,
+            client_version,
+        )))
+    }
+
     fn on_version_mismatch(
         &self,
         _version: u8,
@@ -718,13 +842,11 @@ impl MixnetListener {
             IpPacketRequestData::Data(data_request) => {
                 self.on_data_request(data_request, client_version).await
             }
-            IpPacketRequestData::Ping(_) => {
-                log::info!("Received ping request: not implemented, dropping");
-                Ok(vec![])
+            IpPacketRequestData::Ping(ping_request) => {
+                Ok(vec![self.on_ping_request(ping_request, client_version)])
             }
-            IpPacketRequestData::Health(_) => {
-                log::info!("Received health request: not implemented, dropping");
-                Ok(vec![])
+            IpPacketRequestData::Health(health_request) => {
+                Ok(vec![self.on_health_request(health_request, client_version)])
             }
         }
     }
@@ -733,13 +855,27 @@ impl MixnetListener {
         let stopped_clients = self.connected_clients.get_finished_client_handlers();
         let inactive_clients = self.connected_clients.get_inactive_clients().await;
 
-        // TODO: Send disconnect responses to all disconnected clients
-        //for (ip, nym_address) in stopped_clients.iter().chain(disconnected_clients.iter()) {
-        //    let response = IpPacketResponse::new_unrequested_disconnect(...)
-        //    if let Err(err) = self.handle_response(response).await {
-        //        log::error!("Failed to send disconnect response: {err}");
-        //    }
-        //}
+        // WIP(JON): confirm we should send disconnect on handle stopped
+        for (_ip, nym_address, client_version) in &stopped_clients {
+            let response = Response::new_unrequested_disconnect(
+                *nym_address,
+                v7::response::UnrequestedDisconnectReason::Other("handler stopped".to_string()),
+                *client_version,
+            );
+            if let Err(err) = self.handle_response(response).await {
+                log::error!("Failed to send disconnect response: {err}");
+            }
+        }
+        for (_ip, nym_address, client_version) in &inactive_clients {
+            let response = Response::new_unrequested_disconnect(
+                *nym_address,
+                v7::response::UnrequestedDisconnectReason::ClientMixnetTrafficTimeout,
+                *client_version,
+            );
+            if let Err(err) = self.handle_response(response).await {
+                log::error!("Failed to send disconnect response: {err}");
+            }
+        }
 
         self.connected_clients
             .disconnect_stopped_client_handlers(stopped_clients);
@@ -757,17 +893,7 @@ impl MixnetListener {
 
         let response_packet = response.to_bytes()?;
 
-        // We could avoid this lookup if we check this when we create the response.
-        let mix_hops = if let Some(c) = self
-            .connected_clients
-            .lookup_client_from_nym_address(recipient)
-        {
-            c.mix_hops
-        } else {
-            None
-        };
-
-        let input_message = create_input_message(*recipient, response_packet, mix_hops);
+        let input_message = create_input_message(*recipient, response_packet);
         self.mixnet_client
             .send(input_message)
             .await
@@ -881,14 +1007,9 @@ fn verify_signed_request(
     request: &impl SignedRequest,
     client_version: SupportedClientVersion,
 ) -> Result<()> {
-    if let Err(err) = request.verify() {
-        // If the client is V6, we don't care about missing signature
-        if client_version == SupportedClientVersion::V6 {
-            return Ok(());
-        }
-        return Err(IpPacketRouterError::FailedToVerifyRequest { source: err });
-    }
-    Ok(())
+    request
+        .verify()
+        .map_err(|source| IpPacketRouterError::FailedToVerifyRequest { source })
 }
 
 pub(crate) enum ConnectedClientEvent {

service-providers/ip-packet-router/src/util/create_message.rs

@@ -4,15 +4,8 @@ use nym_task::connections::TransmissionLane;
 pub(crate) fn create_input_message(
     nym_address: Recipient,
     response_packet: Vec<u8>,
-    mix_hops: Option<u8>,
 ) -> InputMessage {
     let lane = TransmissionLane::General;
     let packet_type = None;
-    InputMessage::new_regular_with_custom_hops(
-        nym_address,
-        response_packet,
-        lane,
-        packet_type,
-        mix_hops,
-    )
+    InputMessage::new_regular(nym_address, response_packet, lane, packet_type)
 }