From 95ec91daa1297ea9bb1acd176969d801d87b04ac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bogdan-=C8=98tefan=20Neac=C5=9Fu?= Date: Mon, 23 Sep 2024 14:49:18 +0200 Subject: [PATCH] Entry wireguard tickets (#4888) * Create credential verifier in authenticator * Add new version of peer storage with client id * Fix v1 to what it was before * Compact storage into ecash verifier * Fix non-linux build * Less overlapping conditions * Remove moved code * Use handler thread for each peer * Re-spawn stored handles at startup * Keep new function without async & Result * Put query peer in function too * Query bandwidth * Fix clippy * Replace tap with inspect_err * Fix copyright year * Handle version 2 on the reqeust deser * Add protocol type in req/resp messages --- Cargo.lock | 27 +- Cargo.toml | 1 + common/authenticator-requests/Cargo.toml | 15 + common/authenticator-requests/src/error.rs | 22 + common/authenticator-requests/src/lib.rs | 7 +- common/authenticator-requests/src/v1/mod.rs | 6 + .../src/v1}/registration.rs | 14 +- .../authenticator-requests/src/v1/request.rs | 24 +- .../authenticator-requests/src/v1/response.rs | 2 +- .../src/v2/conversion.rs | 69 +++ common/authenticator-requests/src/v2/mod.rs | 9 + .../src/v2/registration.rs | 235 ++++++++++ .../authenticator-requests/src/v2/request.rs | 119 +++++ .../authenticator-requests/src/v2/response.rs | 129 ++++++ .../src/bandwidth_storage_manager.rs | 6 +- .../src/client_bandwidth.rs | 13 + .../credential-verification/src/ecash/mod.rs | 4 + .../20240920120000_wireguard_client_id.sql | 10 + common/gateway-storage/src/lib.rs | 53 ++- common/gateway-storage/src/models.rs | 4 +- common/gateway-storage/src/wireguard_peers.rs | 12 +- .../Cargo.toml | 14 + .../src/lib.rs | 17 + common/wireguard-types/Cargo.toml | 18 - common/wireguard-types/src/error.rs | 15 - common/wireguard-types/src/lib.rs | 8 - common/wireguard/Cargo.toml | 10 +- common/wireguard/src/error.rs | 9 + common/wireguard/src/lib.rs | 55 +-- common/wireguard/src/peer_controller.rs | 431 ++++++++++-------- common/wireguard/src/peer_handle.rs | 131 ++++++ gateway/src/node/mod.rs | 32 +- nym-node/nym-node-http-api/Cargo.toml | 5 +- .../src/router/api/v1/openapi.rs | 3 - nym-node/nym-node-requests/Cargo.toml | 13 +- .../client_interfaces/wireguard/models.rs | 4 +- nym-node/src/node/mod.rs | 4 +- service-providers/authenticator/Cargo.toml | 4 + .../authenticator/src/authenticator.rs | 25 +- .../authenticator/src/cli/peer_handler.rs | 33 +- .../authenticator/src/cli/run.rs | 13 +- service-providers/authenticator/src/error.rs | 6 + .../authenticator/src/mixnet_listener.rs | 160 +++++-- .../authenticator/src/peer_manager.rs | 112 ++--- 44 files changed, 1433 insertions(+), 470 deletions(-) create mode 100644 common/authenticator-requests/src/error.rs rename common/{wireguard-types/src => authenticator-requests/src/v1}/registration.rs (92%) create mode 100644 common/authenticator-requests/src/v2/conversion.rs create mode 100644 common/authenticator-requests/src/v2/mod.rs create mode 100644 common/authenticator-requests/src/v2/registration.rs create mode 100644 common/authenticator-requests/src/v2/request.rs create mode 100644 common/authenticator-requests/src/v2/response.rs create mode 100644 common/gateway-storage/migrations/20240920120000_wireguard_client_id.sql create mode 100644 common/service-provider-requests-common/Cargo.toml create mode 100644 common/service-provider-requests-common/src/lib.rs create mode 100644 common/wireguard/src/peer_handle.rs diff --git a/Cargo.lock b/Cargo.lock index aff191a0e8..77feb33354 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4228,7 +4228,11 @@ 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", @@ -4252,11 +4256,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 +5495,6 @@ dependencies = [ "nym-node-requests", "nym-task", "nym-wireguard", - "nym-wireguard-types", "rand", "serde_json", "thiserror", @@ -5678,6 +5689,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 +6210,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 +6231,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", ] diff --git a/Cargo.toml b/Cargo.toml index b99802bee6..93f8bcdfa7 100644 --- a/Cargo.toml +++ b/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", diff --git a/common/authenticator-requests/Cargo.toml b/common/authenticator-requests/Cargo.toml index dfe51ae0d2..4dd8570281 100644 --- a/common/authenticator-requests/Cargo.toml +++ b/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"] diff --git a/common/authenticator-requests/src/error.rs b/common/authenticator-requests/src/error.rs new file mode 100644 index 0000000000..d07ad7c1e7 --- /dev/null +++ b/common/authenticator-requests/src/error.rs @@ -0,0 +1,22 @@ +// Copyright 2024 - Nym Technologies SA +// 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, + }, +} diff --git a/common/authenticator-requests/src/lib.rs b/common/authenticator-requests/src/lib.rs index 8360b09ee3..25a03fb858 100644 --- a/common/authenticator-requests/src/lib.rs +++ b/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; diff --git a/common/authenticator-requests/src/v1/mod.rs b/common/authenticator-requests/src/v1/mod.rs index 610fbedfdf..ecd43e2e53 100644 --- a/common/authenticator-requests/src/v1/mod.rs +++ b/common/authenticator-requests/src/v1/mod.rs @@ -1,7 +1,13 @@ // Copyright 2024 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 +pub mod registration; pub mod request; pub mod response; +pub use registration::{ClientMac, ClientMessage, GatewayClient, InitMessage, Nonce}; + +#[cfg(feature = "verify")] +pub use registration::HmacSha256; + const VERSION: u8 = 1; diff --git a/common/wireguard-types/src/registration.rs b/common/authenticator-requests/src/v1/registration.rs similarity index 92% rename from common/wireguard-types/src/registration.rs rename to common/authenticator-requests/src/v1/registration.rs index ef90642b95..034f0d3a60 100644 --- a/common/wireguard-types/src/registration.rs +++ b/common/authenticator-requests/src/v1/registration.rs @@ -1,9 +1,9 @@ -// Copyright 2023 - Nym Technologies SA +// Copyright 2023-2024 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 use crate::error::Error; -use crate::PeerPublicKey; use base64::{engine::general_purpose, Engine}; +use nym_wireguard_types::PeerPublicKey; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::net::IpAddr; @@ -30,7 +30,6 @@ 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), @@ -38,18 +37,12 @@ pub enum ClientMessage { } #[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 } } @@ -78,17 +71,14 @@ pub struct RemainingBandwidthData { /// 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, } diff --git a/common/authenticator-requests/src/v1/request.rs b/common/authenticator-requests/src/v1/request.rs index f849031218..b317e1c73f 100644 --- a/common/authenticator-requests/src/v1/request.rs +++ b/common/authenticator-requests/src/v1/request.rs @@ -1,8 +1,9 @@ // Copyright 2024 - Nym Technologies SA // 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); + } +} diff --git a/common/authenticator-requests/src/v1/response.rs b/common/authenticator-requests/src/v1/response.rs index 83c5a71d50..4e7ba61eb4 100644 --- a/common/authenticator-requests/src/v1/response.rs +++ b/common/authenticator-requests/src/v1/response.rs @@ -1,8 +1,8 @@ // Copyright 2024 - Nym Technologies SA // 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; diff --git a/common/authenticator-requests/src/v2/conversion.rs b/common/authenticator-requests/src/v2/conversion.rs new file mode 100644 index 0000000000..1c40a4e357 --- /dev/null +++ b/common/authenticator-requests/src/v2/conversion.rs @@ -0,0 +1,69 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use nym_service_provider_requests_common::{Protocol, ServiceProviderType}; + +use crate::{v1, v2}; + +impl From 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 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 for v2::registration::InitMessage { + fn from(init_msg: v1::registration::InitMessage) -> Self { + Self { + pub_key: init_msg.pub_key, + } + } +} + +impl From for Box { + fn from(gw_client: v1::registration::GatewayClient) -> Self { + Box::new(v2::registration::FinalMessage { + gateway_client: gw_client.into(), + credential: None, + }) + } +} + +impl From 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 for v2::registration::ClientMac { + fn from(mac: v1::registration::ClientMac) -> Self { + Self::new(mac.to_vec()) + } +} diff --git a/common/authenticator-requests/src/v2/mod.rs b/common/authenticator-requests/src/v2/mod.rs new file mode 100644 index 0000000000..29be89580d --- /dev/null +++ b/common/authenticator-requests/src/v2/mod.rs @@ -0,0 +1,9 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +pub mod conversion; +pub mod registration; +pub mod request; +pub mod response; + +const VERSION: u8 = 2; diff --git a/common/authenticator-requests/src/v2/registration.rs b/common/authenticator-requests/src/v2/registration.rs new file mode 100644 index 0000000000..1d08e02b68 --- /dev/null +++ b/common/authenticator-requests/src/v2/registration.rs @@ -0,0 +1,235 @@ +// Copyright 2023-2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use crate::error::Error; +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; +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; +pub type PrivateIPs = HashMap; + +#[cfg(feature = "verify")] +pub type HmacSha256 = Hmac; + +pub type Nonce = u64; +pub type Taken = Option; + +pub const BANDWIDTH_CAP_PER_DAY: u64 = 1024 * 1024 * 1024; // 1 GB + +#[derive(Serialize, Deserialize, Debug, Clone)] +#[serde(tag = "type", rename_all = "camelCase")] +pub enum ClientMessage { + Initial(InitMessage), + Final(GatewayClient), + Query(PeerPublicKey), +} + +#[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 FinalMessage { + /// Gateway client data + pub gateway_client: GatewayClient, + + /// Ecash credential + pub credential: Option, +} + +#[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, +} + +/// 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); + +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) -> Self { + ClientMac(mac) + } +} + +impl Deref for ClientMac { + type Target = Vec; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +impl FromStr for ClientMac { + type Err = Error; + + fn from_str(s: &str) -> Result { + let mac_bytes: Vec = + 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(&self, serializer: S) -> Result { + let encoded_key = general_purpose::STANDARD.encode(self.0.clone()); + serializer.serialize_str(&encoded_key) + } +} + +impl<'de> Deserialize<'de> for ClientMac { + fn deserialize>(deserializer: D) -> Result { + 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()) + } +} diff --git a/common/authenticator-requests/src/v2/request.rs b/common/authenticator-requests/src/v2/request.rs new file mode 100644 index 0000000000..1c2b05af11 --- /dev/null +++ b/common/authenticator-requests/src/v2/request.rs @@ -0,0 +1,119 @@ +// Copyright 2024 - Nym Technologies SA +// 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 { + 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, bincode::Error> { + use bincode::Options; + make_bincode_serializer().serialize(self) + } +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum AuthenticatorRequestData { + Initial(InitMessage), + Final(Box), + QueryBandwidth(PeerPublicKey), +} + +#[cfg(test)] +mod tests { + use super::*; + use std::str::FromStr; + + #[test] + fn check_first_byte_version() { + 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(); + assert_eq!(*bytes.first().unwrap(), version); + } +} diff --git a/common/authenticator-requests/src/v2/response.rs b/common/authenticator-requests/src/v2/response.rs new file mode 100644 index 0000000000..ab05dfcd35 --- /dev/null +++ b/common/authenticator-requests/src/v2/response.rs @@ -0,0 +1,129 @@ +// Copyright 2024 - Nym Technologies SA +// 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, + 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, bincode::Error> { + use bincode::Options; + make_bincode_serializer().serialize(self) + } + + pub fn from_reconstructed_message( + message: &nym_sphinx::receiver::ReconstructedMessage, + ) -> Result { + use bincode::Options; + make_bincode_serializer().deserialize(&message.message) + } + + pub fn id(&self) -> Option { + 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, +} diff --git a/common/credential-verification/src/bandwidth_storage_manager.rs b/common/credential-verification/src/bandwidth_storage_manager.rs index 3a67d43743..8d538f3aa7 100644 --- a/common/credential-verification/src/bandwidth_storage_manager.rs +++ b/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 BandwidthStorageManager { } } + 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) diff --git a/common/credential-verification/src/client_bandwidth.rs b/common/credential-verification/src/client_bandwidth.rs index 610a198e88..43a494798a 100644 --- a/common/credential-verification/src/client_bandwidth.rs +++ b/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, diff --git a/common/credential-verification/src/ecash/mod.rs b/common/credential-verification/src/ecash/mod.rs index c024f37733..1fc847bc36 100644 --- a/common/credential-verification/src/ecash/mod.rs +++ b/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, diff --git a/common/gateway-storage/migrations/20240920120000_wireguard_client_id.sql b/common/gateway-storage/migrations/20240920120000_wireguard_client_id.sql new file mode 100644 index 0000000000..9653750149 --- /dev/null +++ b/common/gateway-storage/migrations/20240920120000_wireguard_client_id.sql @@ -0,0 +1,10 @@ +/* + * Copyright 2024 - Nym Technologies SA + * 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; diff --git a/common/gateway-storage/src/lib.rs b/common/gateway-storage/src/lib.rs index 3c3e5472d9..86da88c69b 100644 --- a/common/gateway-storage/src/lib.rs +++ b/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, 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 { - 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, 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( diff --git a/common/gateway-storage/src/models.rs b/common/gateway-storage/src/models.rs index 70b6c407c1..74a2e5162d 100644 --- a/common/gateway-storage/src/models.rs +++ b/common/gateway-storage/src/models.rs @@ -116,7 +116,7 @@ pub struct WireguardPeer { pub rx_bytes: i64, pub persistent_keepalive_interval: Option, pub allowed_ips: Vec, - pub suspended: bool, + pub client_id: Option, } impl From for WireguardPeer { @@ -146,7 +146,7 @@ impl From for WireguardPeer { &value.allowed_ips, ) .unwrap_or_default(), - suspended: false, + client_id: None, } } } diff --git a/common/gateway-storage/src/wireguard_peers.rs b/common/gateway-storage/src/wireguard_peers.rs index c90eb9647d..00c41483be 100644 --- a/common/gateway-storage/src/wireguard_peers.rs +++ b/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 /// diff --git a/common/service-provider-requests-common/Cargo.toml b/common/service-provider-requests-common/Cargo.toml new file mode 100644 index 0000000000..f88fdcc0e1 --- /dev/null +++ b/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"] } diff --git a/common/service-provider-requests-common/src/lib.rs b/common/service-provider-requests-common/src/lib.rs new file mode 100644 index 0000000000..ef56062be3 --- /dev/null +++ b/common/service-provider-requests-common/src/lib.rs @@ -0,0 +1,17 @@ +// Copyright 2024 - Nym Technologies SA +// SPDX-License-Identifier: Apache-2.0 + +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub enum ServiceProviderType { + Authenticator, + IpPacketRouter, + NetworkRequester, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct Protocol { + pub version: u8, + pub service_provider_type: ServiceProviderType, +} diff --git a/common/wireguard-types/Cargo.toml b/common/wireguard-types/Cargo.toml index d0c387fe89..e1ffda761c 100644 --- a/common/wireguard-types/Cargo.toml +++ b/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"] diff --git a/common/wireguard-types/src/error.rs b/common/wireguard-types/src/error.rs index 0bd4a8d3f6..bd8c54b6f7 100644 --- a/common/wireguard-types/src/error.rs +++ b/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, - }, } diff --git a/common/wireguard-types/src/lib.rs b/common/wireguard-types/src/lib.rs index bdb8b575b7..8f73b40419 100644 --- a/common/wireguard-types/src/lib.rs +++ b/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; diff --git a/common/wireguard/Cargo.toml b/common/wireguard/Cargo.toml index f90708a68d..f20651acbd 100644 --- a/common/wireguard/Cargo.toml +++ b/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 } diff --git a/common/wireguard/src/error.rs b/common/wireguard/src/error.rs index bd5fa06399..c261ac123a 100644 --- a/common/wireguard/src/error.rs +++ b/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), } diff --git a/common/wireguard/src/lib.rs b/common/wireguard/src/lib.rs index cf6ce7e857..599cbaf082 100644 --- a/common/wireguard/src/lib.rs +++ b/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, - peer_tx: UnboundedSender, + peer_tx: Sender, } impl WireguardGatewayData { - pub fn new( - config: Config, - keypair: Arc, - ) -> (Self, UnboundedReceiver) { - let (peer_tx, peer_rx) = mpsc::unbounded_channel(); + pub fn new(config: Config, keypair: Arc) -> (Self, Receiver) { + let (peer_tx, peer_rx) = mpsc::channel(1); ( WireguardGatewayData { config, @@ -70,44 +68,45 @@ impl WireguardGatewayData { &self.keypair } - pub fn peer_tx(&self) -> &UnboundedSender { + pub fn peer_tx(&self) -> &Sender { &self.peer_tx } } pub struct WireguardData { pub inner: WireguardGatewayData, - pub peer_rx: UnboundedReceiver, + pub peer_rx: Receiver, } /// Start wireguard device #[cfg(target_os = "linux")] -pub async fn start_wireguard( +pub async fn start_wireguard( storage: St, all_peers: Vec, task_client: nym_task::TaskClient, wireguard_data: WireguardData, - control_tx: UnboundedSender, ) -> Result, Box> { 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::, _>>()?; + 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( )]); 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) } diff --git a/common/wireguard/src/peer_controller.rs b/common/wireguard/src/peer_controller.rs index 77375c23d7..aa8d5f975e 100644 --- a/common/wireguard/src/peer_controller.rs +++ b/common/wireguard/src/peer_controller.rs @@ -1,259 +1,292 @@ // Copyright 2024 - Nym Technologies SA // 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, }, RemovePeer { - success: bool, + key: Key, + response_tx: oneshot::Sender, }, QueryPeer { - success: bool, - peer: Option, + key: Key, + response_tx: oneshot::Sender, }, QueryBandwidth { - bandwidth_data: Option, + key: Key, + response_tx: oneshot::Sender, }, } -pub struct PeerController { - storage: St, - request_rx: mpsc::UnboundedReceiver, - response_tx: mpsc::UnboundedSender, - wg_api: Arc, - timeout_check_interval: IntervalStream, - active_peers: HashMap, - suspended_peers: HashMap, - last_seen_bandwidth: HashMap, - timeout_count: u8, +pub struct AddPeerControlResponse { + pub success: bool, + pub client_id: Option, } -impl PeerController { +pub struct RemovePeerControlResponse { + pub success: bool, +} + +pub struct QueryPeerControlResponse { + pub success: bool, + pub peer: Option, +} + +pub struct QueryBandwidthControlResponse { + pub success: bool, + pub bandwidth_data: Option, +} + +pub struct PeerController { + storage: St, + // used to receive commands from individual handles too + request_tx: mpsc::Sender, + request_rx: mpsc::Receiver, + wg_api: Arc, + host_information: Arc>, + bw_storage_managers: HashMap>>, + timeout_check_interval: IntervalStream, + task_client: nym_task::TaskClient, +} + +impl PeerController { pub fn new( storage: St, wg_api: Arc, - peers: Vec, - suspended_peers: Vec, - request_rx: mpsc::UnboundedReceiver, - response_tx: mpsc::UnboundedSender, + initial_host_information: Host, + bw_storage_managers: HashMap>>, + request_tx: mpsc::Sender, + request_rx: mpsc::Receiver, + 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 = peers - .into_iter() - .map(|peer| (peer.public_key.clone(), peer)) - .collect(); - let suspended_peers: HashMap = 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 { - 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, 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::, _>>()?; - 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>, 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, 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, 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, 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"); diff --git a/common/wireguard/src/peer_handle.rs b/common/wireguard/src/peer_handle.rs new file mode 100644 index 0000000000..abf367b3c9 --- /dev/null +++ b/common/wireguard/src/peer_handle.rs @@ -0,0 +1,131 @@ +// Copyright 2024 - Nym Technologies SA +// 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 = Arc>>; + +pub struct PeerHandle { + storage: St, + public_key: Key, + host_information: Arc>, + bandwidth_storage_manager: Option>, + request_tx: mpsc::Sender, + timeout_check_interval: IntervalStream, + task_client: TaskClient, +} + +impl PeerHandle { + pub fn new( + storage: St, + public_key: Key, + host_information: Arc>, + bandwidth_storage_manager: Option>, + request_tx: mpsc::Sender, + 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 { + 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 { + 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(()) + } +} diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index e6eece9139..bf5c3f0ab0 100644 --- a/gateway/src/node/mod.rs +++ b/gateway/src/node/mod.rs @@ -246,6 +246,7 @@ impl Gateway { &mut self, forwarding_channel: MixForwardingSender, shutdown: TaskClient, + ecash_verifier: Arc>, ) -> Result> where St: Storage + Clone + 'static, @@ -256,7 +257,6 @@ impl Gateway { .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 Gateway { 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 Gateway { all_peers, shutdown, wireguard_data, - peer_response_tx, ) .await?; @@ -342,6 +341,7 @@ impl Gateway { &self, _forwarding_channel: MixForwardingSender, _shutdown: TaskClient, + _ecash_verifier: Arc>, ) -> Result> { todo!("Authenticator is currently only supported on Linux"); } @@ -616,14 +616,16 @@ impl Gateway { .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 Gateway { 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 Gateway { 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); diff --git a/nym-node/nym-node-http-api/Cargo.toml b/nym-node/nym-node-http-api/Cargo.toml index 352a2fca6e..bf96afc08c 100644 --- a/nym-node/nym-node-http-api/Cargo.toml +++ b/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 } diff --git a/nym-node/nym-node-http-api/src/router/api/v1/openapi.rs b/nym-node/nym-node-http-api/src/router/api/v1/openapi.rs index f63f48065b..a0ef945f38 100644 --- a/nym-node/nym-node-http-api/src/router/api/v1/openapi.rs +++ b/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, diff --git a/nym-node/nym-node-requests/Cargo.toml b/nym-node/nym-node-requests/Cargo.toml index da61331809..fc1d0853c3 100644 --- a/nym-node/nym-node-requests/Cargo.toml +++ b/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"] diff --git a/nym-node/nym-node-requests/src/api/v1/gateway/client_interfaces/wireguard/models.rs b/nym-node/nym-node-requests/src/api/v1/gateway/client_interfaces/wireguard/models.rs index 9b67da4c62..37ff421f37 100644 --- a/nym-node/nym-node-requests/src/api/v1/gateway/client_interfaces/wireguard/models.rs +++ b/nym-node/nym-node-requests/src/api/v1/gateway/client_interfaces/wireguard/models.rs @@ -1,6 +1,4 @@ // Copyright 2023 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -pub use nym_wireguard_types::{ - ClientMac, ClientMessage, GatewayClient, InitMessage, Nonce, PeerPublicKey, -}; +pub use nym_wireguard_types::PeerPublicKey; diff --git a/nym-node/src/node/mod.rs b/nym-node/src/node/mod.rs index 1e1b42ff54..ce832617aa 100644 --- a/nym-node/src/node/mod.rs +++ b/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, + peer_rx: mpsc::Receiver, } impl WireguardData { diff --git a/service-providers/authenticator/Cargo.toml b/service-providers/authenticator/Cargo.toml index c25a5b2d47..61da8e1d96 100644 --- a/service-providers/authenticator/Cargo.toml +++ b/service-providers/authenticator/Cargo.toml @@ -35,7 +35,11 @@ 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" } diff --git a/service-providers/authenticator/src/authenticator.rs b/service-providers/authenticator/src/authenticator.rs index 92de903913..d5a69e7d44 100644 --- a/service-providers/authenticator/src/authenticator.rs +++ b/service-providers/authenticator/src/authenticator.rs @@ -1,15 +1,16 @@ // Copyright 2024 - Nym Technologies SA // 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 { #[allow(unused)] config: Config, wait_for_gateway: bool, custom_topology_provider: Option>, custom_gateway_transceiver: Option>, wireguard_gateway_data: WireguardGatewayData, + ecash_verifier: Option>>, used_private_network_ips: Vec, - response_rx: UnboundedReceiver, shutdown: Option, on_start: Option>, } -impl Authenticator { +impl Authenticator { pub fn new( config: Config, wireguard_gateway_data: WireguardGatewayData, used_private_network_ips: Vec, - response_rx: UnboundedReceiver, ) -> 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>) -> 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}"); diff --git a/service-providers/authenticator/src/cli/peer_handler.rs b/service-providers/authenticator/src/cli/peer_handler.rs index 8c644f7ef7..561e995b65 100644 --- a/service-providers/authenticator/src/cli/peer_handler.rs +++ b/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, - response_tx: mpsc::UnboundedSender, + peer_rx: mpsc::Receiver, task_client: TaskClient, } impl DummyHandler { - pub fn new( - peer_rx: mpsc::UnboundedReceiver, - response_tx: mpsc::UnboundedSender, - task_client: TaskClient, - ) -> Self { + pub fn new(peer_rx: mpsc::Receiver, 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 { diff --git a/service-providers/authenticator/src/cli/run.rs b/service-providers/authenticator/src/cli/run.rs index 4844cd1284..42f9e9ab0e 100644 --- a/service-providers/authenticator/src/cli/run.rs +++ b/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::::new( + config, + wireguard_gateway_data, + vec![], + ); if let Some(custom_mixnet) = &args.common_args.custom_mixnet { server = server.with_stored_topology(custom_mixnet)? } diff --git a/service-providers/authenticator/src/error.rs b/service-providers/authenticator/src/error.rs index add82fd778..2a3a781906 100644 --- a/service-providers/authenticator/src/error.rs +++ b/service-providers/authenticator/src/error.rs @@ -14,6 +14,9 @@ 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, @@ -41,6 +44,9 @@ 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), diff --git a/service-providers/authenticator/src/mixnet_listener.rs b/service-providers/authenticator/src/mixnet_listener.rs index 2bb32181db..1bb62cfbfc 100644 --- a/service-providers/authenticator/src/mixnet_listener.rs +++ b/service-providers/authenticator/src/mixnet_listener.rs @@ -8,22 +8,36 @@ 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_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 +59,7 @@ impl RegistredAndFree { } } -pub(crate) struct MixnetListener { +pub(crate) struct MixnetListener { // The configuration for the mixnet listener pub(crate) config: Config, @@ -60,17 +74,19 @@ pub(crate) struct MixnetListener { pub(crate) peer_manager: PeerManager, + pub(crate) ecash_verifier: Option>>, + pub(crate) timeout_check_interval: IntervalStream, } -impl MixnetListener { +impl MixnetListener { pub fn new( config: Config, free_private_network_ips: PrivateIPs, wireguard_gateway_data: WireguardGatewayData, - response_rx: UnboundedReceiver, mixnet_client: nym_sdk::mixnet::MixnetClient, task_handle: TaskHandle, + ecash_verifier: Option>>, ) -> Self { let timeout_check_interval = IntervalStream::new(tokio::time::interval(DEFAULT_REGISTRATION_TIMEOUT_CHECK)); @@ -79,7 +95,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 +149,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 +182,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 +216,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>, + credential: CredentialSpendingData, + client_id: i64, + ) -> Result { + 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 +349,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) => { @@ -356,7 +438,11 @@ fn deserialize_request(reconstructed: &ReconstructedMessage) -> Result v1::request::AuthenticatorRequest::from_reconstructed_message(reconstructed) - .map_err(|err| AuthenticatorError::FailedToDeserializeTaggedPacket { source: err }), + .map_err(|err| AuthenticatorError::FailedToDeserializeTaggedPacket { source: err }) + .map(Into::into), + 2 => v2::request::AuthenticatorRequest::from_reconstructed_message(reconstructed) + .map_err(|err| AuthenticatorError::FailedToDeserializeTaggedPacket { source: err }) + .map(Into::into), _ => { log::info!("Received packet with invalid version: v{request_version}"); Err(AuthenticatorError::InvalidPacketVersion(request_version)) diff --git a/service-providers/authenticator/src/peer_manager.rs b/service-providers/authenticator/src/peer_manager.rs index cdc847a2ca..fbbdf6a06f 100644 --- a/service-providers/authenticator/src/peer_manager.rs +++ b/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, } impl PeerManager { - pub fn new( - wireguard_gateway_data: WireguardGatewayData, - response_rx: UnboundedReceiver, - ) -> 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> { 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> { 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> { 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) } }