GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Move credential verifier in peer controller (#5938)

Browse Source 
* Move credential verifier in peer controller

* Send back errors of peer controller
This commit is contained in:
Bogdan-Ștefan Neacşu
2025-08-13 13:09:44 +03:00
committed by GitHub
parent 906dfb2fb0
commit d95f66bd90
10 changed files with 303 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -8025,7 +8025,9 @@ dependencies = [
"log",
"nym-authenticator-requests",
"nym-credential-verification",
"nym-credentials-interface 0.1.0",
"nym-crypto 0.4.0",
"nym-gateway-requests",
"nym-gateway-storage",
"nym-network-defaults 0.1.0",
"nym-node-metrics",
common/wireguard/Cargo.toml
+2 -1
View File
@@ -32,9 +32,11 @@ time = { workspace = true }
tracing = { workspace = true }
nym-authenticator-requests = { path = "../authenticator-requests" }
nym-credentials-interface = { path = "../credentials-interface" }
nym-credential-verification = { path = "../credential-verification" }
nym-crypto = { path = "../crypto", features = ["asymmetric"] }
nym-gateway-storage = { path = "../gateway-storage" }
nym-gateway-requests = { path = "../gateway-requests" }
nym-network-defaults = { path = "../network-defaults" }
nym-task = { path = "../task" }
nym-wireguard-types = { path = "../wireguard-types" }
@@ -46,4 +48,3 @@ nym-gateway-storage = { path = "../gateway-storage", features = ["mock"] }
[features]
default = []
mock = ["nym-gateway-storage/mock"]
common/wireguard/src/error.rs
+2
View File
@@ -21,3 +21,5 @@ pub enum Error {
#[error("{0}")]
SystemTime(#[from] std::time::SystemTimeError),
}
pub type Result<T> = std::result::Result<T, Error>;
common/wireguard/src/lib.rs
+6 -5
View File
@@ -7,9 +7,9 @@
// #![warn(clippy::unwrap_used)]
use defguard_wireguard_rs::{host::Peer, key::Key, net::IpAddrMask, WGApi, WireguardInterfaceApi};
use nym_crypto::asymmetric::x25519::KeyPair;
#[cfg(target_os = "linux")]
use nym_gateway_storage::GatewayStorage;
use nym_credential_verification::ecash::EcashManager;
use nym_crypto::asymmetric::x25519::KeyPair;
use nym_wireguard_types::Config;
use peer_controller::PeerControlRequest;
use std::sync::Arc;
@@ -158,7 +158,7 @@ pub struct WireguardData {
/// Start wireguard device
#[cfg(target_os = "linux")]
pub async fn start_wireguard(
storage: GatewayStorage,
ecash_manager: Arc<EcashManager>,
metrics: nym_node_metrics::NymNodeMetrics,
peers: Vec<Peer>,
task_client: nym_task::TaskClient,
@@ -167,6 +167,7 @@ pub async fn start_wireguard(
use base64::{prelude::BASE64_STANDARD, Engine};
use defguard_wireguard_rs::{InterfaceConfiguration, WireguardInterfaceApi};
use ip_network::IpNetwork;
use nym_credential_verification::ecash::traits::EcashManager;
use peer_controller::PeerController;
use std::collections::HashMap;
use tokio::sync::RwLock;
@@ -178,7 +179,7 @@ pub async fn start_wireguard(
for peer in peers.iter() {
let bandwidth_manager = Arc::new(RwLock::new(
PeerController::generate_bandwidth_manager(Box::new(storage.clone()), &peer.public_key)
PeerController::generate_bandwidth_manager(ecash_manager.storage(), &peer.public_key)
.await?,
));
peer_bandwidth_managers.insert(peer.public_key.clone(), (bandwidth_manager, peer.clone()));
@@ -233,7 +234,7 @@ pub async fn start_wireguard(
let host = wg_api.read_interface_data()?;
let wg_api = std::sync::Arc::new(WgApiWrapper::new(wg_api));
let mut controller = PeerController::new(
Box::new(storage),
ecash_manager,
metrics,
wg_api.clone(),
host,
common/wireguard/src/peer_controller.rs
+92 -66
View File
@@ -9,9 +9,11 @@ use defguard_wireguard_rs::{
use futures::channel::oneshot;
use log::info;
use nym_credential_verification::{
bandwidth_storage_manager::BandwidthStorageManager, BandwidthFlushingBehaviourConfig,
ClientBandwidth,
bandwidth_storage_manager::BandwidthStorageManager, ecash::traits::EcashManager,
BandwidthFlushingBehaviourConfig, ClientBandwidth, CredentialVerifier,
};
use nym_credentials_interface::CredentialSpendingData;
use nym_gateway_requests::models::CredentialSpendingRequest;
use nym_gateway_storage::traits::BandwidthGatewayStorage;
use nym_node_metrics::NymNodeMetrics;
use nym_wireguard_types::DEFAULT_PEER_TIMEOUT_CHECK;
@@ -20,7 +22,10 @@ use std::{collections::HashMap, sync::Arc};
use tokio::sync::{mpsc, RwLock};
use tokio_stream::{wrappers::IntervalStream, StreamExt};
use crate::{error::Error, peer_handle::SharedBandwidthStorageManager};
use crate::{
error::{Error, Result},
peer_handle::SharedBandwidthStorageManager,
};
use crate::{peer_handle::PeerHandle, peer_storage_manager::CachedPeerManager};
pub enum PeerControlRequest {
@@ -40,27 +45,21 @@ pub enum PeerControlRequest {
key: Key,
response_tx: oneshot::Sender<GetClientBandwidthControlResponse>,
},
GetVerifier {
key: Key,
credential: Box<CredentialSpendingData>,
response_tx: oneshot::Sender<QueryVerifierControlResponse>,
},
}
pub struct AddPeerControlResponse {
pub success: bool,
}
pub struct RemovePeerControlResponse {
pub success: bool,
}
pub struct QueryPeerControlResponse {
pub success: bool,
pub peer: Option<Peer>,
}
pub struct GetClientBandwidthControlResponse {
pub client_bandwidth: Option<ClientBandwidth>,
}
pub type AddPeerControlResponse = Result<()>;
pub type RemovePeerControlResponse = Result<()>;
pub type QueryPeerControlResponse = Result<Option<Peer>>;
pub type GetClientBandwidthControlResponse = Result<ClientBandwidth>;
pub type QueryVerifierControlResponse = Result<CredentialVerifier>;
pub struct PeerController {
storage: Box<dyn BandwidthGatewayStorage + Send + Sync>,
ecash_verifier: Arc<dyn EcashManager + Send + Sync>,
// we have "all" metrics of a node, but they're behind a single Arc pointer,
// so the overhead is minimal
@@ -79,7 +78,7 @@ pub struct PeerController {
impl PeerController {
#[allow(clippy::too_many_arguments)]
pub fn new(
storage: Box<dyn BandwidthGatewayStorage + Send + Sync>,
ecash_verifier: Arc<dyn EcashManager + Send + Sync>,
metrics: NymNodeMetrics,
wg_api: Arc<dyn WireguardInterfaceApi + Send + Sync>,
initial_host_information: Host,
@@ -114,7 +113,7 @@ impl PeerController {
.collect();
PeerController {
storage,
ecash_verifier,
wg_api,
host_information,
bw_storage_managers,
@@ -127,8 +126,11 @@ impl PeerController {
}
// 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?;
pub async fn remove_peer(&mut self, key: &Key) -> Result<()> {
self.ecash_verifier
.storage()
.remove_wireguard_peer(&key.to_string())
.await?;
self.bw_storage_managers.remove(key);
let ret = self.wg_api.remove_peer(key);
if ret.is_err() {
@@ -140,7 +142,7 @@ impl PeerController {
pub async fn generate_bandwidth_manager(
storage: Box<dyn BandwidthGatewayStorage + Send + Sync>,
public_key: &Key,
) -> Result<BandwidthStorageManager, Error> {
) -> Result<BandwidthStorageManager> {
let client_id = storage
.get_wireguard_peer(&public_key.to_string())
.await?
@@ -161,14 +163,11 @@ impl PeerController {
))
}
async fn handle_add_request(&mut self, peer: &Peer) -> Result<(), Error> {
async fn handle_add_request(&mut self, peer: &Peer) -> Result<()> {
self.wg_api.configure_peer(peer)?;
let bandwidth_storage_manager = Arc::new(RwLock::new(
Self::generate_bandwidth_manager(
dyn_clone::clone_box(&*self.storage),
&peer.public_key,
)
.await?,
Self::generate_bandwidth_manager(self.ecash_verifier.storage(), &peer.public_key)
.await?,
));
let cached_peer_manager = CachedPeerManager::new(peer);
let mut handle = PeerHandle::new(
@@ -193,21 +192,52 @@ impl PeerController {
Ok(())
}
async fn handle_query_peer(&self, key: &Key) -> Result<Option<Peer>, Error> {
async fn handle_query_peer(&self, key: &Key) -> Result<Option<Peer>> {
Ok(self
.storage
.ecash_verifier
.storage()
.get_wireguard_peer(&key.to_string())
.await?
.map(Peer::try_from)
.transpose()?)
}
async fn handle_get_client_bandwidth(&self, key: &Key) -> Option<ClientBandwidth> {
if let Some(bandwidth_storage_manager) = self.bw_storage_managers.get(key) {
Some(bandwidth_storage_manager.read().await.client_bandwidth())
} else {
None
}
async fn handle_get_client_bandwidth(&self, key: &Key) -> Result<ClientBandwidth> {
let bandwidth_storage_manager = self
.bw_storage_managers
.get(key)
.ok_or(Error::MissingClientBandwidthEntry)?;
Ok(bandwidth_storage_manager.read().await.client_bandwidth())
}
async fn handle_query_verifier(
&self,
key: &Key,
credential: CredentialSpendingData,
) -> Result<CredentialVerifier> {
let storage = self.ecash_verifier.storage();
let client_id = storage
.get_wireguard_peer(&key.to_string())
.await?
.ok_or(Error::MissingClientBandwidthEntry)?
.client_id;
let Some(bandwidth_storage_manager) = self.bw_storage_managers.get(key) else {
return Err(Error::MissingClientBandwidthEntry);
};
let client_bandwidth = bandwidth_storage_manager.read().await.client_bandwidth();
let verifier = CredentialVerifier::new(
CredentialSpendingRequest::new(credential),
self.ecash_verifier.clone(),
BandwidthStorageManager::new(
storage,
client_bandwidth,
client_id,
BandwidthFlushingBehaviourConfig::default(),
true,
),
);
Ok(verifier)
}
async fn update_metrics(&self, new_host: &Host) {
@@ -304,28 +334,19 @@ impl PeerController {
msg = self.request_rx.recv() => {
match msg {
Some(PeerControlRequest::AddPeer { peer, response_tx }) => {
let ret = self.handle_add_request(&peer).await;
if ret.is_ok() {
response_tx.send(AddPeerControlResponse { success: true }).ok();
} else {
response_tx.send(AddPeerControlResponse { success: false }).ok();
}
response_tx.send(self.handle_add_request(&peer).await).ok();
}
Some(PeerControlRequest::RemovePeer { key, response_tx }) => {
let success = self.remove_peer(&key).await.is_ok();
response_tx.send(RemovePeerControlResponse { success }).ok();
response_tx.send(self.remove_peer(&key).await).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();
}
response_tx.send(self.handle_query_peer(&key).await).ok();
}
Some(PeerControlRequest::GetClientBandwidth { key, response_tx }) => {
let client_bandwidth = self.handle_get_client_bandwidth(&key).await;
response_tx.send(GetClientBandwidthControlResponse { client_bandwidth }).ok();
response_tx.send(self.handle_get_client_bandwidth(&key).await).ok();
}
Some(PeerControlRequest::GetVerifier { key, credential, response_tx }) => {
response_tx.send(self.handle_query_verifier(&key, *credential).await).ok();
}
None => {
log::trace!("PeerController [main loop]: stopping since channel closed");
@@ -349,21 +370,21 @@ struct MockWgApi {
impl WireguardInterfaceApi for MockWgApi {
fn create_interface(
&self,
) -> Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
) -> std::result::Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
todo!()
}
fn assign_address(
&self,
_address: &defguard_wireguard_rs::net::IpAddrMask,
) -> Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
) -> std::result::Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
todo!()
}
fn configure_peer_routing(
&self,
_peers: &[Peer],
) -> Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
) -> std::result::Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
todo!()
}
@@ -371,7 +392,7 @@ impl WireguardInterfaceApi for MockWgApi {
fn configure_interface(
&self,
_config: &defguard_wireguard_rs::InterfaceConfiguration,
) -> Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
) -> std::result::Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
todo!()
}
@@ -380,20 +401,20 @@ impl WireguardInterfaceApi for MockWgApi {
&self,
_config: &defguard_wireguard_rs::InterfaceConfiguration,
_dns: &[std::net::IpAddr],
) -> Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
) -> std::result::Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
todo!()
}
fn remove_interface(
&self,
) -> Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
) -> std::result::Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
todo!()
}
fn configure_peer(
&self,
peer: &Peer,
) -> Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
) -> std::result::Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
self.peers
.write()
.unwrap()
@@ -404,14 +425,14 @@ impl WireguardInterfaceApi for MockWgApi {
fn remove_peer(
&self,
peer_pubkey: &Key,
) -> Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
) -> std::result::Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
self.peers.write().unwrap().remove(peer_pubkey);
Ok(())
}
fn read_interface_data(
&self,
) -> Result<Host, defguard_wireguard_rs::error::WireguardInterfaceError> {
) -> std::result::Result<Host, defguard_wireguard_rs::error::WireguardInterfaceError> {
let mut host = Host::default();
host.peers = self.peers.read().unwrap().clone();
Ok(host)
@@ -420,7 +441,7 @@ impl WireguardInterfaceApi for MockWgApi {
fn configure_dns(
&self,
_dns: &[std::net::IpAddr],
) -> Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
) -> std::result::Result<(), defguard_wireguard_rs::error::WireguardInterfaceError> {
todo!()
}
}
@@ -433,13 +454,18 @@ pub fn start_controller(
Arc<RwLock<nym_gateway_storage::traits::mock::MockGatewayStorage>>,
nym_task::TaskManager,
) {
use std::sync::Arc;
let storage = Arc::new(RwLock::new(
nym_gateway_storage::traits::mock::MockGatewayStorage::default(),
));
let ecash_manager = Arc::new(nym_credential_verification::ecash::MockEcashManager::new(
Box::new(storage.clone()),
));
let wg_api = Arc::new(MockWgApi::default());
let task_manager = nym_task::TaskManager::default();
let mut peer_controller = PeerController::new(
Box::new(storage.clone()),
ecash_manager,
Default::default(),
wg_api,
Default::default(),
common/wireguard/src/peer_handle.rs
+2 -1
View File
@@ -62,7 +62,8 @@ impl PeerHandle {
let success = response_rx
.await
.map_err(|_| Error::Internal("peer controller didn't respond".to_string()))?
.success;
.inspect_err(|err| tracing::error!("Could not remove peer: {err:?}"))
.is_ok();
Ok(success)
}
gateway/src/node/mod.rs
+7 -1
View File
@@ -486,8 +486,14 @@ impl GatewayTasksBuilder {
);
};
let Some(ecash_manager) = self.ecash_manager.clone() else {
return Err(
GatewayError::InternalWireguardError("ecash manager not set".to_string()).into(),
);
};
let wg_handle = nym_wireguard::start_wireguard(
self.storage.clone(),
ecash_manager,
self.metrics.clone(),
all_peers,
self.shutdown.fork("wireguard"),
service-providers/authenticator/src/error.rs
-3
View File
@@ -80,9 +80,6 @@ pub enum AuthenticatorError {
#[error("peers can't be interacted with anymore")]
PeerInteractionStopped,
#[error("storage should have the requested bandwidth entry")]
MissingClientBandwidthEntry,
#[error("unknown version number")]
UnknownVersion,
service-providers/authenticator/src/mixnet_listener.rs
+17 -46
View File
@@ -604,12 +604,12 @@ impl MixnetListener {
request_id: u64,
reply_to: Option<Recipient>,
) -> AuthenticatorHandleResult {
let bandwidth_data = self.peer_manager.query_bandwidth(msg.pub_key()).await?;
let available_bandwidth = self.peer_manager.query_bandwidth(msg.pub_key()).await?;
let bytes = match AuthenticatorVersion::from(protocol) {
AuthenticatorVersion::V1 => {
v1::response::AuthenticatorResponse::new_remaining_bandwidth(
bandwidth_data.map(|data| v1::registration::RemainingBandwidthData {
available_bandwidth: data as u64,
Some(v1::registration::RemainingBandwidthData {
available_bandwidth: available_bandwidth as u64,
suspended: false,
}),
reply_to.ok_or(AuthenticatorError::MissingReplyToForOldClient)?,
@@ -622,10 +622,8 @@ impl MixnetListener {
}
AuthenticatorVersion::V2 => {
v2::response::AuthenticatorResponse::new_remaining_bandwidth(
bandwidth_data.map(|available_bandwidth| {
v2::registration::RemainingBandwidthData {
available_bandwidth,
}
Some(v2::registration::RemainingBandwidthData {
available_bandwidth,
}),
reply_to.ok_or(AuthenticatorError::MissingReplyToForOldClient)?,
request_id,
@@ -637,10 +635,8 @@ impl MixnetListener {
}
AuthenticatorVersion::V3 => {
v3::response::AuthenticatorResponse::new_remaining_bandwidth(
bandwidth_data.map(|available_bandwidth| {
v3::registration::RemainingBandwidthData {
available_bandwidth,
}
Some(v3::registration::RemainingBandwidthData {
available_bandwidth,
}),
reply_to.ok_or(AuthenticatorError::MissingReplyToForOldClient)?,
request_id,
@@ -652,10 +648,8 @@ impl MixnetListener {
}
AuthenticatorVersion::V4 => {
v4::response::AuthenticatorResponse::new_remaining_bandwidth(
bandwidth_data.map(|available_bandwidth| {
v4::registration::RemainingBandwidthData {
available_bandwidth,
}
Some(v4::registration::RemainingBandwidthData {
available_bandwidth,
}),
reply_to.ok_or(AuthenticatorError::MissingReplyToForOldClient)?,
request_id,
@@ -667,10 +661,8 @@ impl MixnetListener {
}
AuthenticatorVersion::V5 => {
v5::response::AuthenticatorResponse::new_remaining_bandwidth(
bandwidth_data.map(|available_bandwidth| {
v5::registration::RemainingBandwidthData {
available_bandwidth,
}
Some(v5::registration::RemainingBandwidthData {
available_bandwidth,
}),
request_id,
)
@@ -691,38 +683,17 @@ impl MixnetListener {
request_id: u64,
reply_to: Option<Recipient>,
) -> AuthenticatorHandleResult {
let client_id = self
.ecash_verifier
.storage()
.get_wireguard_peer(&msg.pub_key().to_string())
.await?
.ok_or(AuthenticatorError::MissingClientBandwidthEntry)?
.client_id;
let client_bandwidth = self
.peer_manager
.query_client_bandwidth(msg.pub_key())
.await?
.ok_or(AuthenticatorError::MissingClientBandwidthEntry)?;
let available_bandwidth = if self.received_retry(msg.as_ref()) {
// don't process the credential and just return the current bandwidth
client_bandwidth.available().await
self.peer_manager.query_bandwidth(msg.pub_key()).await?
} else {
let credential = msg.credential();
let mut verifier = CredentialVerifier::new(
CredentialSpendingRequest::new(credential.clone()),
self.ecash_verifier.clone(),
BandwidthStorageManager::new(
self.ecash_verifier.storage(),
client_bandwidth,
client_id,
BandwidthFlushingBehaviourConfig::default(),
true,
),
);
let mut verifier = self
.peer_manager
.query_verifier(msg.pub_key(), msg.credential())
.await?;
let available_bandwidth = verifier.verify().await?;
self.seen_credential_cache
.insert_credential(credential, msg.pub_key());
.insert_credential(msg.credential(), msg.pub_key());
available_bandwidth
};
service-providers/authenticator/src/peer_manager.rs
+173 -74
View File
@@ -4,14 +4,9 @@
use crate::error::*;
use defguard_wireguard_rs::{host::Peer, key::Key};
use futures::channel::oneshot;
use nym_credential_verification::ClientBandwidth;
use nym_wireguard::{
peer_controller::{
AddPeerControlResponse, GetClientBandwidthControlResponse, PeerControlRequest,
QueryPeerControlResponse, RemovePeerControlResponse,
},
WireguardGatewayData,
};
use nym_credential_verification::{ClientBandwidth, CredentialVerifier};
use nym_credentials_interface::CredentialSpendingData;
use nym_wireguard::{peer_controller::PeerControlRequest, WireguardGatewayData};
use nym_wireguard_types::PeerPublicKey;
pub struct PeerManager {
@@ -33,15 +28,14 @@ impl PeerManager {
.await
.map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
let AddPeerControlResponse { 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(),
));
}
Ok(())
response_rx
.await
.map_err(|_| AuthenticatorError::InternalError("no response for add peer".to_string()))?
.map_err(|err| {
AuthenticatorError::InternalError(format!(
"adding peer could not be performed: {err:?}"
))
})
}
pub async fn _remove_peer(&mut self, pub_key: PeerPublicKey) -> Result<()> {
@@ -54,15 +48,16 @@ impl PeerManager {
.await
.map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
let RemovePeerControlResponse { success } = response_rx.await.map_err(|_| {
AuthenticatorError::InternalError("no response for remove peer".to_string())
})?;
if !success {
return Err(AuthenticatorError::InternalError(
"removing peer could not be performed".to_string(),
));
}
Ok(())
response_rx
.await
.map_err(|_| {
AuthenticatorError::InternalError("no response for remove peer".to_string())
})?
.map_err(|err| {
AuthenticatorError::InternalError(format!(
"removing peer could not be performed: {err:?}"
))
})
}
pub async fn query_peer(&mut self, public_key: PeerPublicKey) -> Result<Option<Peer>> {
@@ -75,30 +70,24 @@ impl PeerManager {
.await
.map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
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(),
));
}
Ok(peer)
response_rx
.await
.map_err(|_| {
AuthenticatorError::InternalError("no response for query peer".to_string())
})?
.map_err(|err| {
AuthenticatorError::InternalError(format!(
"querying peer could not be performed: {err:?}"
))
})
}
pub async fn query_bandwidth(&mut self, public_key: PeerPublicKey) -> Result<Option<i64>> {
let res = if let Some(client_bandwidth) = self.query_client_bandwidth(public_key).await? {
Some(client_bandwidth.available().await)
} else {
None
};
Ok(res)
pub async fn query_bandwidth(&mut self, public_key: PeerPublicKey) -> Result<i64> {
let client_bandwidth = self.query_client_bandwidth(public_key).await?;
Ok(client_bandwidth.available().await)
}
pub async fn query_client_bandwidth(
&mut self,
key: PeerPublicKey,
) -> Result<Option<ClientBandwidth>> {
pub async fn query_client_bandwidth(&mut self, key: PeerPublicKey) -> Result<ClientBandwidth> {
let key = Key::new(key.to_bytes());
let (response_tx, response_rx) = oneshot::channel();
let msg = PeerControlRequest::GetClientBandwidth { key, response_tx };
@@ -108,13 +97,48 @@ impl PeerManager {
.await
.map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
let GetClientBandwidthControlResponse { client_bandwidth } =
response_rx.await.map_err(|_| {
response_rx
.await
.map_err(|_| {
AuthenticatorError::InternalError(
"no response for query client bandwidth".to_string(),
)
})?;
Ok(client_bandwidth)
})?
.map_err(|err| {
AuthenticatorError::InternalError(format!(
"querying client bandwidth could not be performed: {err:?}"
))
})
}
pub async fn query_verifier(
&mut self,
key: PeerPublicKey,
credential: CredentialSpendingData,
) -> Result<CredentialVerifier> {
let key = Key::new(key.to_bytes());
let (response_tx, response_rx) = oneshot::channel();
let msg = PeerControlRequest::GetVerifier {
key,
credential: Box::new(credential),
response_tx,
};
self.wireguard_gateway_data
.peer_tx()
.send(msg)
.await
.map_err(|_| AuthenticatorError::PeerInteractionStopped)?;
response_rx
.await
.map_err(|_| {
AuthenticatorError::InternalError("no response for query verifier".to_string())
})?
.map_err(|err| {
AuthenticatorError::InternalError(format!(
"querying verifier could not be performed: {err:?}"
))
})
}
}
@@ -137,6 +161,71 @@ mod tests {
use super::*;
const CREDENTIAL_BYTES: [u8; 1245] = [
0, 0, 4, 133, 96, 179, 223, 185, 136, 23, 213, 166, 59, 203, 66, 69, 209, 181, 227, 254,
16, 102, 98, 237, 59, 119, 170, 111, 31, 194, 51, 59, 120, 17, 115, 229, 79, 91, 11, 139,
154, 2, 212, 23, 68, 70, 167, 3, 240, 54, 224, 171, 221, 1, 69, 48, 60, 118, 119, 249, 123,
35, 172, 227, 131, 96, 232, 209, 187, 123, 4, 197, 102, 90, 96, 45, 125, 135, 140, 99, 1,
151, 17, 131, 143, 157, 97, 107, 139, 232, 212, 87, 14, 115, 253, 255, 166, 167, 186, 43,
90, 96, 173, 105, 120, 40, 10, 163, 250, 224, 214, 200, 178, 4, 160, 16, 130, 59, 76, 193,
39, 240, 3, 101, 141, 209, 183, 226, 186, 207, 56, 210, 187, 7, 164, 240, 164, 205, 37, 81,
184, 214, 193, 195, 90, 205, 238, 225, 195, 104, 12, 123, 203, 57, 233, 243, 215, 145, 195,
196, 57, 38, 125, 172, 18, 47, 63, 165, 110, 219, 180, 40, 58, 116, 92, 254, 160, 98, 48,
92, 254, 232, 107, 184, 80, 234, 60, 160, 235, 249, 76, 41, 38, 165, 28, 40, 136, 74, 48,
166, 50, 245, 23, 201, 140, 101, 79, 93, 235, 128, 186, 146, 126, 180, 134, 43, 13, 186,
19, 195, 48, 168, 201, 29, 216, 95, 176, 198, 132, 188, 64, 39, 212, 150, 32, 52, 53, 38,
228, 199, 122, 226, 217, 75, 40, 191, 151, 48, 164, 242, 177, 79, 14, 122, 105, 151, 85,
88, 199, 162, 17, 96, 103, 83, 178, 128, 9, 24, 30, 74, 108, 241, 85, 240, 166, 97, 241,
85, 199, 11, 198, 226, 234, 70, 107, 145, 28, 208, 114, 51, 12, 234, 108, 101, 202, 112,
48, 185, 22, 159, 67, 109, 49, 27, 149, 90, 109, 32, 226, 112, 7, 201, 208, 209, 104, 31,
97, 134, 204, 145, 27, 181, 206, 181, 106, 32, 110, 136, 115, 249, 201, 111, 5, 245, 203,
71, 121, 169, 126, 151, 178, 236, 59, 221, 195, 48, 135, 115, 6, 50, 227, 74, 97, 107, 107,
213, 90, 2, 203, 154, 138, 47, 128, 52, 134, 128, 224, 51, 65, 240, 90, 8, 55, 175, 180,
178, 204, 206, 168, 110, 51, 57, 189, 169, 48, 169, 136, 121, 99, 51, 170, 178, 214, 74, 1,
96, 151, 167, 25, 173, 180, 171, 155, 10, 55, 142, 234, 190, 113, 90, 79, 80, 244, 71, 166,
30, 235, 113, 150, 133, 1, 218, 17, 109, 111, 223, 24, 216, 177, 41, 2, 204, 65, 221, 212,
207, 236, 144, 6, 65, 224, 55, 42, 1, 1, 161, 134, 118, 127, 111, 220, 110, 127, 240, 71,
223, 129, 12, 93, 20, 220, 60, 56, 71, 146, 184, 95, 132, 69, 28, 56, 53, 192, 213, 22,
119, 230, 152, 225, 182, 188, 163, 219, 37, 175, 247, 73, 14, 247, 38, 72, 243, 1, 48, 131,
59, 8, 13, 96, 143, 185, 127, 241, 161, 217, 24, 149, 193, 40, 16, 30, 202, 151, 28, 119,
240, 153, 101, 156, 61, 193, 72, 245, 199, 181, 12, 231, 65, 166, 67, 142, 121, 207, 202,
58, 197, 113, 188, 248, 42, 124, 105, 48, 161, 241, 55, 209, 36, 194, 27, 63, 233, 144,
189, 85, 117, 234, 9, 139, 46, 31, 206, 114, 95, 131, 29, 240, 13, 81, 142, 140, 133, 33,
30, 41, 141, 37, 80, 217, 95, 221, 76, 115, 86, 201, 165, 51, 252, 9, 28, 209, 1, 48, 150,
74, 248, 212, 187, 222, 66, 210, 3, 200, 19, 217, 171, 184, 42, 148, 53, 150, 57, 50, 6,
227, 227, 62, 49, 42, 148, 148, 157, 82, 191, 58, 24, 34, 56, 98, 120, 89, 105, 176, 85,
15, 253, 241, 41, 153, 195, 136, 1, 48, 142, 126, 213, 101, 223, 79, 133, 230, 105, 38,
161, 149, 2, 21, 136, 150, 42, 72, 218, 85, 146, 63, 223, 58, 108, 186, 183, 248, 62, 20,
47, 34, 113, 160, 177, 204, 181, 16, 24, 212, 224, 35, 84, 51, 168, 56, 136, 11, 1, 48,
135, 242, 62, 149, 230, 178, 32, 224, 119, 26, 234, 163, 237, 224, 114, 95, 112, 140, 170,
150, 96, 125, 136, 221, 180, 78, 18, 11, 12, 184, 2, 198, 217, 119, 43, 69, 4, 172, 109,
55, 183, 40, 131, 172, 161, 88, 183, 101, 1, 48, 173, 216, 22, 73, 42, 255, 211, 93, 249,
87, 159, 115, 61, 91, 55, 130, 17, 216, 60, 34, 122, 55, 8, 244, 244, 153, 151, 57, 5, 144,
178, 55, 249, 64, 211, 168, 34, 148, 56, 89, 92, 203, 70, 124, 219, 152, 253, 165, 0, 32,
203, 116, 63, 7, 240, 222, 82, 86, 11, 149, 167, 72, 224, 55, 190, 66, 201, 65, 168, 184,
96, 47, 194, 241, 168, 124, 7, 74, 214, 250, 37, 76, 32, 218, 69, 122, 103, 215, 145, 169,
24, 212, 229, 168, 106, 10, 144, 31, 13, 25, 178, 242, 250, 106, 159, 40, 48, 163, 165, 61,
130, 57, 146, 4, 73, 32, 254, 233, 125, 135, 212, 29, 111, 4, 177, 114, 15, 210, 170, 82,
108, 110, 62, 166, 81, 209, 106, 176, 156, 14, 133, 242, 60, 127, 120, 242, 28, 97, 0, 1,
32, 103, 93, 109, 89, 240, 91, 1, 84, 150, 50, 206, 157, 203, 49, 220, 120, 234, 175, 234,
150, 126, 225, 94, 163, 164, 199, 138, 114, 62, 99, 106, 112, 1, 32, 171, 40, 220, 82, 241,
203, 76, 146, 111, 139, 182, 179, 237, 182, 115, 75, 128, 201, 107, 43, 214, 0, 135, 217,
160, 68, 150, 232, 144, 114, 237, 98, 32, 30, 134, 232, 59, 93, 163, 253, 244, 13, 202, 52,
147, 168, 83, 121, 123, 95, 21, 210, 209, 225, 223, 143, 49, 10, 205, 238, 1, 22, 83, 81,
70, 1, 32, 26, 76, 6, 234, 160, 50, 139, 102, 161, 232, 155, 106, 130, 171, 226, 210, 233,
178, 85, 247, 71, 123, 55, 53, 46, 67, 148, 137, 156, 207, 208, 107, 1, 32, 102, 31, 4, 98,
110, 156, 144, 61, 229, 140, 198, 84, 196, 238, 128, 35, 131, 182, 137, 125, 241, 95, 69,
131, 170, 27, 2, 144, 75, 72, 242, 102, 3, 32, 121, 80, 45, 173, 56, 65, 218, 27, 40, 251,
197, 32, 169, 104, 123, 110, 90, 78, 153, 166, 38, 9, 129, 228, 99, 8, 1, 116, 142, 233,
162, 69, 32, 216, 169, 159, 116, 95, 12, 63, 176, 195, 6, 183, 123, 135, 75, 61, 112, 106,
83, 235, 176, 41, 27, 248, 48, 71, 165, 170, 12, 92, 103, 103, 81, 32, 58, 74, 75, 145,
192, 94, 153, 69, 80, 128, 241, 3, 16, 117, 192, 86, 161, 103, 44, 174, 211, 196, 182, 124,
55, 11, 107, 142, 49, 88, 6, 41, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6,
6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6,
6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 0, 37, 139, 240, 0, 0,
0, 0, 0, 0, 0, 1,
];
#[tokio::test]
async fn add_peer() {
let (wireguard_data, request_rx) = WireguardGatewayData::new(
@@ -242,18 +331,10 @@ mod tests {
request_rx,
);
assert!(peer_manager
.query_bandwidth(public_key)
.await
.unwrap()
.is_none());
assert!(peer_manager.query_bandwidth(public_key).await.is_err());
helper_add_peer(&storage, &mut peer_manager).await;
let available_bandwidth = peer_manager
.query_bandwidth(public_key)
.await
.unwrap()
.unwrap();
let available_bandwidth = peer_manager.query_bandwidth(public_key).await.unwrap();
assert_eq!(available_bandwidth, 0);
stop_controller(task_manager).await;
@@ -276,15 +357,13 @@ mod tests {
assert!(peer_manager
.query_client_bandwidth(public_key)
.await
.unwrap()
.is_none());
.is_err());
helper_add_peer(&storage, &mut peer_manager).await;
let available_bandwidth = peer_manager
.query_client_bandwidth(public_key)
.await
.unwrap()
.unwrap()
.available()
.await;
assert_eq!(available_bandwidth, 0);
@@ -292,6 +371,35 @@ mod tests {
stop_controller(task_manager).await;
}
#[tokio::test]
async fn query_verifier() {
let (wireguard_data, request_rx) = WireguardGatewayData::new(
Authenticator::default().into(),
Arc::new(KeyPair::new(&mut OsRng)),
);
let mut peer_manager = PeerManager::new(wireguard_data);
let key = Key::default();
let public_key = PeerPublicKey::from_str(&key.to_string()).unwrap();
let (storage, task_manager) = start_controller(
peer_manager.wireguard_gateway_data.peer_tx().clone(),
request_rx,
);
let credential = CredentialSpendingData::try_from_bytes(&CREDENTIAL_BYTES).unwrap();
assert!(peer_manager
.query_verifier(public_key, credential.clone())
.await
.is_err());
helper_add_peer(&storage, &mut peer_manager).await;
peer_manager
.query_verifier(public_key, credential)
.await
.unwrap();
stop_controller(task_manager).await;
}
#[tokio::test]
async fn increase_decrease_bandwidth() {
let (wireguard_data, request_rx) = WireguardGatewayData::new(
@@ -312,7 +420,6 @@ mod tests {
let client_bandwidth = peer_manager
.query_client_bandwidth(public_key)
.await
.unwrap()
.unwrap();
let mut bw_manager = BandwidthStorageManager::new(
@@ -334,11 +441,7 @@ mod tests {
assert_eq!(client_bandwidth.available().await, top_up);
assert_eq!(
peer_manager
.query_bandwidth(public_key)
.await
.unwrap()
.unwrap(),
peer_manager.query_bandwidth(public_key).await.unwrap(),
top_up
);
@@ -346,11 +449,7 @@ mod tests {
let remaining = top_up - consume;
assert_eq!(client_bandwidth.available().await, remaining);
assert_eq!(
peer_manager
.query_bandwidth(public_key)
.await
.unwrap()
.unwrap(),
peer_manager.query_bandwidth(public_key).await.unwrap(),
remaining
);