Compare commits

...

28 Commits

Author SHA1 Message Date
Bogdan-Ștefan Neacşu 2659293383 Extend raw ws fd for gateway client 2024-12-05 11:34:34 +00:00
benedetta davico 0041937ed3 Merge pull request #5170 from nymtech/merge/release/2024.13-magura-patched
merge patched magura into develop
2024-11-25 09:06:32 +01:00
Jon Häggblad 5cda49f996 Add derive_extended_private_key to DirectSecp256k1HdWallet (#5167) 2024-11-23 11:19:36 +01:00
Jon Häggblad 36657bcd97 Add support for DELETE to nym-http-api-client (#5166)
* Add delete support to http-api-client

* Add to trait
2024-11-23 10:42:49 +01:00
Jon Häggblad 6167243a10 Add export_to_env (#5162) 2024-11-22 17:50:43 +01:00
Jon Häggblad 920276f2ac Move two minor jobs to free tier github hosted runners (#5169) 2024-11-22 16:56:24 +01:00
Jędrzej Stuczyński 0e5bd966dd Merge branch 'release/2024.13-magura-patched' into develop 2024-11-22 15:53:59 +00:00
Jędrzej Stuczyński 6acd936368 bugfix: additional improvements to nym-api db performance (#5168)
* added statement logging to nym-api db

* adding additional pool options

* dont blow up upon failing to submit network monitor results

* store in-memory cache of node database ids
2024-11-22 15:52:26 +00:00
Dinko Zdravac 1f53da7456 Fix env var name (#5165) 2024-11-22 15:41:16 +00:00
Jon Häggblad 2fdc0dc47b Add strum::EnumIter for TicketType (#5164) 2024-11-22 16:33:23 +01:00
Fran Arbanas a720c95dd7 feat: add env vars to NNM (#5163)
* feat: add env vars to NNM

* update version to testfran

* add ip and port for locust connection as variables

* bump version

* bump version, fix env variable names

* bump version to correct one
2024-11-22 14:32:58 +00:00
benedettadavico 01c7b2819e update changelog 2024-11-22 10:50:59 +01:00
Simon Wicky 042a8a58aa start session collection for exit gateways (#5148) (#5161) 2024-11-22 09:12:01 +01:00
benedettadavico d8ab2a8f15 update version for clients and nym-node 2024-11-22 07:25:54 +01:00
mx 013941dbaf patch curl command for network_tunnel_man.sh (#5160) 2024-11-21 20:28:22 +00:00
benedetta davico 1af6e1ecdd Create network_tunnel_manager.sh (#5156)
* Create network_tunnel_manager.sh

* update link for network tunnel manager script

* updating 2 more links
2024-11-21 20:11:28 +00:00
Jędrzej Stuczyński d6d2239685 implement optional cap for number of ack retransmissions (#5158)
* implement optional cap for number of ack retransmissions

* Wasm
2024-11-21 18:57:24 +00:00
Jędrzej Stuczyński 3d704fbbf1 change: make nym-api optionally ignore nodes with illegal ip addresses, like loopback (#5159) 2024-11-21 18:54:14 +00:00
Jędrzej Stuczyński 119c36b0bb added 'deterministic_route_selection' for sphinx packet route selection (#5157)
* added 'deterministic_route_selection' for sphinx packet route selection

* clippy + wasm

* Switch to ChaCha8Rng for deterministic route generation

---------

Co-authored-by: durch <durch@users.noreply.github.com>
2024-11-21 18:21:01 +00:00
Mark Sinclair 379c1eb0d0 Update push-network-monitor.yaml 2024-11-21 16:11:59 +00:00
Mark Sinclair d1e91946e6 Update push-network-monitor.yaml 2024-11-21 16:07:12 +00:00
Mark Sinclair 402c79f2f5 Update push-network-monitor.yaml 2024-11-21 15:59:41 +00:00
Mark Sinclair 149b2f4e32 Update Cargo.toml for nym-network-monitor 2024-11-21 15:52:56 +00:00
Mark Sinclair 221e809da3 Update push-network-monitor.yaml 2024-11-21 15:52:11 +00:00
Mark Sinclair 760ee453ea Update nym-network-monitor.dockerfile 2024-11-21 15:25:38 +00:00
Dinko Zdravac 3f072e4e9d NS Agent auth with NS API (#5127)
* Agents authenticate with NSAPI

* /submit with better auth
- also adjust agent run script to authenticate, even in parallel

* /request better authentication
- moved agent API calls to Client struct

* Replay protection

* Fix testrun cleanup bug
- introduce a new column last_assigned which is different than
  created_at so that stale testruns get cleaned up based on
  last_assigned
- created_at is still useful for determining the "oldest" testrun
  to be picked up

* Uniform request authentication

* Suppress ts-rs serde warnings

* Update cargo version

* All agents use the same key
- remove assigned_agent column
- remove logic which would stop agents with
  the same key to connect
- as a safety measure, add cap to total no. of agents
2024-11-21 14:32:15 +01:00
mx aa460076f4 Max/surb doc update lifetime (#5154)
* updated surbs page with lifetimes info + attacks + diagram

* component build update

* update rewrites
2024-11-21 10:37:31 +00:00
Jędrzej Stuczyński 6b5b97199b fix: tracing logger to output to stderr 2024-11-19 17:47:13 +00:00
79 changed files with 1696 additions and 473 deletions
+1 -1
View File
@@ -11,7 +11,7 @@ on:
jobs:
check-schema:
name: Generate and check schema
runs-on: arc-ubuntu-20.04
runs-on: ubuntu-20.04
env:
CARGO_TERM_COLOR: always
steps:
+1 -1
View File
@@ -13,7 +13,7 @@ on:
jobs:
matrix_prep:
runs-on: arc-ubuntu-20.04
runs-on: ubuntu-20.04
outputs:
matrix: ${{ steps.set-matrix.outputs.matrix }}
steps:
+14 -18
View File
@@ -25,31 +25,27 @@ jobs:
git config --global user.email "lawrence@nymtech.net"
git config --global user.name "Lawrence Stalder"
- name: Get version from package.json
uses: sergeysova/jq-action@v2
- name: Get version from cargo.toml
uses: mikefarah/yq@v4.44.5
id: get_version
with:
cmd: jq -r '.version' ${{ env.WORKING_DIRECTORY }}/package.json
- name: Check if tag exists
run: |
if git rev-parse ${{ steps.get_version.outputs.value }} >/dev/null 2>&1; then
echo "Tag ${{ steps.get_version.outputs.value }} already exists"
fi
cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/nym-network-monitor/Cargo.toml
- name: Remove existing tag if exists
run: |
if git rev-parse ${{ steps.get_version.outputs.value }} >/dev/null 2>&1; then
git push --delete origin ${{ steps.get_version.outputs.value }}
git tag -d ${{ steps.get_version.outputs.value }}
echo "Checking if tag ${{ env.CONTAINER_NAME }}-${{ steps.get_version.outputs.result }} exists..."
if git rev-parse ${{ env.CONTAINER_NAME }}-${{ steps.get_version.outputs.result }} >/dev/null 2>&1; then
echo "Tag ${{ env.CONTAINER_NAME }}-${{ steps.get_version.outputs.result }} already exists"
git push --delete origin ${{ env.CONTAINER_NAME }}-${{ steps.get_version.outputs.result }}
git tag -d ${{ env.CONTAINER_NAME }}-${{ steps.get_version.outputs.result }}
fi
- name: Create tag
run: |
git tag -a ${{ steps.get_version.outputs.value }} -m "Version ${{ steps.get_version.outputs.value }}"
git push origin ${{ steps.get_version.outputs.value }}
git tag -a ${{ env.CONTAINER_NAME }}-${{ steps.get_version.outputs.result }} -m "Version ${{ steps.get_version.outputs.result }}"
git push origin ${{ env.CONTAINER_NAME }}-${{ steps.get_version.outputs.result }}
- name: BuildAndPushImageOnHarbor
run: |
docker build -f nym-network-monitor.dockerfile ${{ env.WORKING_DIRECTORY }} -t harbor.nymte.ch/nym/${{ env.CONTAINER_NAME }}:${{ steps.get_version.outputs.value }} -t harbor.nymte.ch/nym/${{ env.CONTAINER_NAME }}:latest
docker push harbor.nymte.ch/nym/${{ env.CONTAINER_NAME }} --all-tags
docker build -f nym-network-monitor.dockerfile ${{ env.WORKING_DIRECTORY }} -t harbor.nymte.ch/nym/${{ env.CONTAINER_NAME }}:${{ steps.get_version.outputs.result }} -t harbor.nymte.ch/nym/${{ env.CONTAINER_NAME }}:latest
docker push harbor.nymte.ch/nym/${{ env.CONTAINER_NAME }} --all-tags
+6
View File
@@ -4,6 +4,12 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
## [Unreleased]
## [2024.13-magura-patched] (2024-11-22)
- [experimental] allow clients to change between deterministic route selection based on packet headers and a pseudorandom distribution
- Introduced a configurable limit on retransmission frequency of packets if ACKs are not received
- Filtered out invalid IP addresses on nym-api
## [2024.13-magura] (2024-11-18)
- Limit race probability ([#5145])
Generated
+26 -13
View File
@@ -3822,9 +3822,9 @@ dependencies = [
[[package]]
name = "libc"
version = "0.2.155"
version = "0.2.162"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "97b3888a4aecf77e811145cadf6eef5901f4782c53886191b2f693f24761847c"
checksum = "18d287de67fe55fd7e1581fe933d965a5a9477b38e949cfa9f8574ef01506398"
[[package]]
name = "libm"
@@ -4470,6 +4470,7 @@ dependencies = [
"cw2",
"cw3",
"cw4",
"dashmap",
"dirs",
"futures",
"getset",
@@ -4781,7 +4782,7 @@ dependencies = [
[[package]]
name = "nym-client"
version = "1.1.43"
version = "1.1.44"
dependencies = [
"bs58",
"clap 4.5.20",
@@ -5010,6 +5011,9 @@ dependencies = [
name = "nym-common-models"
version = "0.1.0"
dependencies = [
"anyhow",
"bincode",
"nym-crypto",
"serde",
]
@@ -5865,7 +5869,7 @@ dependencies = [
[[package]]
name = "nym-network-monitor"
version = "0.1.0"
version = "1.0.2"
dependencies = [
"anyhow",
"axum 0.7.7",
@@ -5895,7 +5899,7 @@ dependencies = [
[[package]]
name = "nym-network-requester"
version = "1.1.44"
version = "1.1.45"
dependencies = [
"addr",
"anyhow",
@@ -5946,7 +5950,7 @@ dependencies = [
[[package]]
name = "nym-node"
version = "1.1.10"
version = "1.1.11"
dependencies = [
"anyhow",
"bip39",
@@ -6049,14 +6053,20 @@ dependencies = [
[[package]]
name = "nym-node-status-agent"
version = "0.1.6"
version = "1.0.0-rc.1"
dependencies = [
"anyhow",
"bincode",
"chrono",
"clap 4.5.20",
"nym-bin-common",
"nym-common-models",
"nym-crypto",
"rand",
"reqwest 0.12.4",
"serde",
"serde_json",
"tempfile",
"tokio",
"tokio-util",
"tracing",
@@ -6065,10 +6075,11 @@ dependencies = [
[[package]]
name = "nym-node-status-api"
version = "0.1.6"
version = "1.0.0-rc.2"
dependencies = [
"anyhow",
"axum 0.7.7",
"bincode",
"chrono",
"clap 4.5.20",
"cosmwasm-std",
@@ -6077,6 +6088,7 @@ dependencies = [
"moka",
"nym-bin-common",
"nym-common-models",
"nym-crypto",
"nym-explorer-client",
"nym-network-defaults",
"nym-node-requests",
@@ -6294,7 +6306,7 @@ dependencies = [
[[package]]
name = "nym-socks5-client"
version = "1.1.43"
version = "1.1.44"
dependencies = [
"bs58",
"clap 4.5.20",
@@ -8277,9 +8289,9 @@ dependencies = [
[[package]]
name = "rustix"
version = "0.38.34"
version = "0.38.40"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "70dc5ec042f7a43c4a73241207cecc9873a06d45debb38b329f8541d85c2730f"
checksum = "99e4ea3e1cdc4b559b8e5650f9c8e5998e3e5c1343b4eaf034565f32318d63c0"
dependencies = [
"bitflags 2.5.0",
"errno",
@@ -9514,12 +9526,13 @@ dependencies = [
[[package]]
name = "tempfile"
version = "3.10.1"
version = "3.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "85b77fafb263dd9d05cbeac119526425676db3784113aa9295c88498cbf8bff1"
checksum = "28cce251fcbc87fac86a866eeb0d6c2d536fc16d06f184bb61aeae11aa4cee0c"
dependencies = [
"cfg-if",
"fastrand 2.1.1",
"once_cell",
"rustix",
"windows-sys 0.52.0",
]
+1 -1
View File
@@ -329,7 +329,7 @@ syn = "1"
sysinfo = "0.30.13"
tap = "1.0.1"
tar = "0.4.42"
tempfile = "3.5.0"
tempfile = "3.14"
thiserror = "1.0.64"
time = "0.3.30"
tokio = "1.39"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-client"
version = "1.1.43"
version = "1.1.44"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
description = "Implementation of the Nym Client"
edition = "2021"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-socks5-client"
version = "1.1.43"
version = "1.1.44"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>"]
description = "A SOCKS5 localhost proxy that converts incoming messages to Sphinx and sends them to a Nym address"
edition = "2021"
+10 -1
View File
@@ -393,13 +393,20 @@ pub struct Traffic {
/// poisson distribution.
pub disable_main_poisson_packet_distribution: bool,
/// Specify whether route selection should be determined by the packet header.
pub deterministic_route_selection: bool,
/// Specify how many times particular packet can be retransmitted
/// None - no limit
pub maximum_number_of_retransmissions: Option<u32>,
/// Specifies the packet size used for sent messages.
/// Do not override it unless you understand the consequences of that change.
pub primary_packet_size: PacketSize,
/// Specifies the optional auxiliary packet size for optimizing message streams.
/// Note that its use decreases overall anonymity.
/// Do not set it it unless you understand the consequences of that change.
/// Do not set it unless you understand the consequences of that change.
pub secondary_packet_size: Option<PacketSize>,
pub packet_type: PacketType,
@@ -424,6 +431,8 @@ impl Default for Traffic {
average_packet_delay: DEFAULT_AVERAGE_PACKET_DELAY,
message_sending_average_delay: DEFAULT_MESSAGE_STREAM_AVERAGE_DELAY,
disable_main_poisson_packet_distribution: false,
deterministic_route_selection: false,
maximum_number_of_retransmissions: None,
primary_packet_size: PacketSize::RegularPacket,
secondary_packet_size: None,
packet_type: PacketType::Mix,
@@ -111,6 +111,7 @@ impl From<ConfigV5> for Config {
primary_packet_size: value.debug.traffic.primary_packet_size,
secondary_packet_size: value.debug.traffic.secondary_packet_size,
packet_type: value.debug.traffic.packet_type,
..Default::default()
},
cover_traffic: CoverTraffic {
loop_cover_traffic_average_delay: value
@@ -30,7 +30,8 @@ pub(crate) enum Action {
InsertPending(Vec<PendingAcknowledgement>),
/// Removes given `PendingAcknowledgement` from the 'shared' state. Also cancels the retransmission timer.
/// Initiated by `AcknowledgementListener`
/// Initiated by `AcknowledgementListener` upon receiving the acknowledgement. Also by `RetransmissionRequestListener`
/// upon deciding to abandon the data.
RemovePending(FragmentIdentifier),
/// Starts the retransmission timer on given `PendingAcknowledgement` with the `Duration` based on
@@ -41,7 +42,7 @@ pub(crate) enum Action {
/// Updates the expected delay of given `PendingAcknowledgement` with the new provided `SphinxDelay`.
/// Initiated by `RetransmissionRequestListener`
UpdateDelay(FragmentIdentifier, SphinxDelay),
UpdatePendingAck(FragmentIdentifier, SphinxDelay),
}
impl Action {
@@ -57,8 +58,8 @@ impl Action {
Action::StartTimer(frag_id)
}
pub(crate) fn new_update_delay(frag_id: FragmentIdentifier, delay: SphinxDelay) -> Self {
Action::UpdateDelay(frag_id, delay)
pub(crate) fn new_update_pending_ack(frag_id: FragmentIdentifier, delay: SphinxDelay) -> Self {
Action::UpdatePendingAck(frag_id, delay)
}
}
@@ -135,7 +136,7 @@ impl ActionController {
}
fn handle_start_timer(&mut self, frag_id: FragmentIdentifier) {
trace!("{} is starting its timer", frag_id);
trace!("{frag_id} is starting its timer");
if let Some((pending_ack_data, queue_key)) = self.pending_acks_data.get_mut(&frag_id) {
// the fact that this branch is now POSSIBLE is a sign of a need to refactor this whole
@@ -193,7 +194,7 @@ impl ActionController {
// initiated basically as a first step of retransmission. At first data has its delay updated
// (as new sphinx packet was created with new expected delivery time)
fn handle_update_delay(&mut self, frag_id: FragmentIdentifier, delay: SphinxDelay) {
fn handle_update_pending_ack(&mut self, frag_id: FragmentIdentifier, delay: SphinxDelay) {
trace!("{} is updating its delay", frag_id);
// TODO: is it possible to solve this without either locking or temporarily removing the value?
if let Some((pending_ack_data, queue_key)) = self.pending_acks_data.remove(&frag_id) {
@@ -202,7 +203,7 @@ impl ActionController {
// reference to this Arc. HOWEVER, before the Action was pushed onto the queue, the reference
// was dropped hence this unwrap is safe.
let mut inner_data = Arc::try_unwrap(pending_ack_data).unwrap();
inner_data.update_delay(delay);
inner_data.update_retransmitted(delay);
self.pending_acks_data
.insert(frag_id, (Arc::new(inner_data), queue_key));
@@ -225,7 +226,7 @@ impl ActionController {
// about it. Perhaps just reschedule it at later point?
let frag_id = expired_ack.into_inner();
trace!("{} has expired", frag_id);
trace!("{frag_id} has expired");
if let Some((pending_ack_data, queue_key)) = self.pending_acks_data.get_mut(&frag_id) {
if queue_key.is_none() {
@@ -258,7 +259,9 @@ impl ActionController {
Action::InsertPending(pending_acks) => self.handle_insert(pending_acks),
Action::RemovePending(frag_id) => self.handle_remove(frag_id),
Action::StartTimer(frag_id) => self.handle_start_timer(frag_id),
Action::UpdateDelay(frag_id, delay) => self.handle_update_delay(frag_id, delay),
Action::UpdatePendingAck(frag_id, delay) => {
self.handle_update_pending_ack(frag_id, delay)
}
}
}
@@ -71,6 +71,7 @@ pub(crate) struct PendingAcknowledgement {
delay: SphinxDelay,
destination: PacketDestination,
mix_hops: Option<u8>,
retransmissions: u32,
}
impl PendingAcknowledgement {
@@ -86,6 +87,7 @@ impl PendingAcknowledgement {
delay,
destination: PacketDestination::KnownRecipient(recipient.into()),
mix_hops,
retransmissions: 0,
}
}
@@ -105,6 +107,7 @@ impl PendingAcknowledgement {
// Messages sent using SURBs are using the number of mix hops set by the recipient when
// they provided the SURBs, so it doesn't make sense to include it here.
mix_hops: None,
retransmissions: 0,
}
}
@@ -116,8 +119,9 @@ impl PendingAcknowledgement {
self.message_chunk.clone()
}
fn update_delay(&mut self, new_delay: SphinxDelay) {
fn update_retransmitted(&mut self, new_delay: SphinxDelay) {
self.delay = new_delay;
self.retransmissions += 1;
}
}
@@ -163,6 +167,9 @@ impl AcknowledgementControllerConnectors {
/// Configurable parameters of the `AcknowledgementController`
pub(super) struct Config {
/// Specify how many times particular packet can be retransmitted
maximum_retransmissions: Option<u32>,
/// Given ack timeout in the form a * BASE_DELAY + b, it specifies the additive part `b`
ack_wait_addition: Duration,
@@ -174,8 +181,13 @@ pub(super) struct Config {
}
impl Config {
pub(super) fn new(ack_wait_addition: Duration, ack_wait_multiplier: f64) -> Self {
pub(super) fn new(
maximum_retransmissions: Option<u32>,
ack_wait_addition: Duration,
ack_wait_multiplier: f64,
) -> Self {
Config {
maximum_retransmissions,
ack_wait_addition,
ack_wait_multiplier,
packet_size: Default::default(),
@@ -238,6 +250,7 @@ where
// will listen for any ack timeouts and trigger retransmission
let retransmission_request_listener = RetransmissionRequestListener::new(
config.maximum_retransmissions,
connectors.ack_action_sender.clone(),
message_handler,
retransmission_rx,
@@ -20,6 +20,7 @@ use std::sync::{Arc, Weak};
// responsible for packet retransmission upon fired timer
pub(super) struct RetransmissionRequestListener<R> {
maximum_retransmissions: Option<u32>,
action_sender: AckActionSender,
message_handler: MessageHandler<R>,
request_receiver: RetransmissionRequestReceiver,
@@ -31,12 +32,14 @@ where
R: CryptoRng + Rng,
{
pub(super) fn new(
maximum_retransmissions: Option<u32>,
action_sender: AckActionSender,
message_handler: MessageHandler<R>,
request_receiver: RetransmissionRequestReceiver,
reply_controller_sender: ReplyControllerSender,
) -> Self {
RetransmissionRequestListener {
maximum_retransmissions,
action_sender,
message_handler,
request_receiver,
@@ -77,6 +80,18 @@ where
}
};
let frag_id = timed_out_ack.message_chunk.fragment_identifier();
if let Some(limit) = self.maximum_retransmissions {
if timed_out_ack.retransmissions >= limit {
warn!("reached maximum number of allowed retransmissions for the packet");
self.action_sender
.unbounded_send(Action::new_remove(frag_id))
.unwrap();
return;
}
}
let maybe_prepared_fragment = match &timed_out_ack.destination {
PacketDestination::Anonymous {
recipient_tag,
@@ -101,8 +116,6 @@ where
}
};
let frag_id = timed_out_ack.message_chunk.fragment_identifier();
let prepared_fragment = match maybe_prepared_fragment {
Ok(prepared_fragment) => prepared_fragment,
Err(err) => {
@@ -136,7 +149,7 @@ where
// with the additional poisson delay.
// And since Actions are executed in order `UpdateTimer` will HAVE TO be executed before `StartTimer`
self.action_sender
.unbounded_send(Action::new_update_delay(frag_id, new_delay))
.unbounded_send(Action::new_update_pending_ack(frag_id, new_delay))
.unwrap();
// send to `OutQueueControl` to eventually send to the mix network
@@ -91,6 +91,9 @@ pub(crate) struct Config {
/// and surb-based are going to be sent.
sender_address: Recipient,
/// Specify whether route selection should be determined by the packet header.
deterministic_route_selection: bool,
/// Average delay a data packet is going to get delay at a single mixnode.
average_packet_delay: Duration,
@@ -114,10 +117,12 @@ impl Config {
sender_address: Recipient,
average_packet_delay: Duration,
average_ack_delay: Duration,
deterministic_route_selection: bool,
) -> Self {
Config {
ack_key,
sender_address,
deterministic_route_selection,
average_packet_delay,
average_ack_delay,
num_mix_hops: DEFAULT_NUM_MIX_HOPS,
@@ -176,6 +181,7 @@ where
{
let message_preparer = MessagePreparer::new(
rng,
config.deterministic_route_selection,
config.sender_address,
config.average_packet_delay,
config.average_ack_delay,
@@ -634,7 +640,7 @@ where
pub(crate) fn update_ack_delay(&self, id: FragmentIdentifier, new_delay: Delay) {
self.action_sender
.unbounded_send(Action::UpdateDelay(id, new_delay))
.unbounded_send(Action::UpdatePendingAck(id, new_delay))
.expect("action control task has died")
}
@@ -65,6 +65,7 @@ pub struct Config {
impl<'a> From<&'a Config> for acknowledgement_control::Config {
fn from(cfg: &'a Config) -> Self {
acknowledgement_control::Config::new(
cfg.traffic.maximum_number_of_retransmissions,
cfg.acks.ack_wait_addition,
cfg.acks.ack_wait_multiplier,
)
@@ -97,6 +98,7 @@ impl<'a> From<&'a Config> for message_handler::Config {
cfg.self_recipient,
cfg.traffic.average_packet_delay,
cfg.acks.average_ack_delay,
cfg.traffic.deterministic_route_selection,
)
.with_custom_primary_packet_size(cfg.traffic.primary_packet_size)
.with_custom_secondary_packet_size(cfg.traffic.secondary_packet_size)
@@ -46,7 +46,8 @@ pub(crate) fn ws_fd(_conn: &WsConn) -> Option<RawFd> {
#[cfg(unix)]
match _conn.get_ref() {
MaybeTlsStream::Plain(stream) => Some(stream.as_raw_fd()),
&_ => None,
MaybeTlsStream::Rustls(tls_stream) => Some(tls_stream.as_raw_fd()),
_ => None,
}
#[cfg(not(unix))]
None
@@ -98,6 +98,13 @@ impl DirectSecp256k1HdWallet {
Ok((private_key, public_key))
}
pub fn derive_extended_private_key(
&self,
hd_path: &DerivationPath,
) -> Result<XPrv, DirectSecp256k1HdWalletError> {
Ok(XPrv::derive_from_path(self.seed, hd_path)?)
}
pub fn try_derive_accounts(&self) -> Result<Vec<AccountData>, DirectSecp256k1HdWalletError> {
let mut accounts = Vec::with_capacity(self.accounts.len());
for derivation_info in &self.accounts {
@@ -15,7 +15,9 @@ mixnet-contract-common = { path = "../mixnet-contract", package = "nym-mixnet-co
contracts-common = { path = "../contracts-common", package = "nym-contracts-common", version = "0.5.0" }
serde = { workspace = true, features = ["derive"] }
thiserror = { workspace = true }
ts-rs = { workspace = true, optional = true}
# without this feature, cargo clippy emits a ton of incompatibility warnings
# https://docs.rs/ts-rs/latest/ts_rs/#serde-compatability
ts-rs = { workspace = true, optional = true, features = ["no-serde-warnings"] }
[features]
schema = ["cw2"]
+9 -1
View File
@@ -221,7 +221,15 @@ impl From<PayInfo> for NymPayInfo {
}
#[derive(
Copy, Clone, Debug, PartialEq, Serialize, Deserialize, strum::Display, strum::EnumString,
Copy,
Clone,
Debug,
PartialEq,
Serialize,
Deserialize,
strum::Display,
strum::EnumString,
strum::EnumIter,
)]
#[serde(rename_all = "kebab-case")]
#[strum(serialize_all = "kebab-case")]
+126
View File
@@ -282,6 +282,48 @@ impl Client {
}
}
pub async fn create_delete_request<K, V>(
&self,
path: PathSegments<'_>,
params: Params<'_, K, V>,
) -> RequestBuilder
where
K: AsRef<str>,
V: AsRef<str>,
{
let url = sanitize_url(&self.base_url, path, params);
self.reqwest_client.delete(url)
}
pub async fn send_delete_request<K, V, E>(
&self,
path: PathSegments<'_>,
params: Params<'_, K, V>,
) -> Result<Response, HttpClientError<E>>
where
K: AsRef<str>,
V: AsRef<str>,
E: Display,
{
tracing::trace!("Sending DELETE request");
let url = sanitize_url(&self.base_url, path, params);
#[cfg(target_arch = "wasm32")]
{
Ok(wasmtimer::tokio::timeout(
self.request_timeout,
self.reqwest_client.delete(url).send(),
)
.await
.map_err(|_timeout| HttpClientError::RequestTimeout)??)
}
#[cfg(not(target_arch = "wasm32"))]
{
Ok(self.reqwest_client.delete(url).send().await?)
}
}
#[instrument(level = "debug", skip_all)]
pub async fn get_json<T, K, V, E>(
&self,
@@ -315,6 +357,21 @@ impl Client {
parse_response(res, true).await
}
pub async fn delete_json<T, K, V, E>(
&self,
path: PathSegments<'_>,
params: Params<'_, K, V>,
) -> Result<T, HttpClientError<E>>
where
for<'a> T: Deserialize<'a>,
K: AsRef<str>,
V: AsRef<str>,
E: Display + DeserializeOwned,
{
let res = self.send_delete_request(path, params).await?;
parse_response(res, false).await
}
#[instrument(level = "debug", skip_all)]
pub async fn get_json_endpoint<T, S, E>(&self, endpoint: S) -> Result<T, HttpClientError<E>>
where
@@ -380,6 +437,35 @@ impl Client {
parse_response(res, true).await
}
pub async fn delete_json_endpoint<T, S, E>(&self, endpoint: S) -> Result<T, HttpClientError<E>>
where
for<'a> T: Deserialize<'a>,
E: Display + DeserializeOwned,
S: AsRef<str>,
{
#[cfg(target_arch = "wasm32")]
let res = {
wasmtimer::tokio::timeout(
self.request_timeout,
self.reqwest_client
.delete(self.base_url.join(endpoint.as_ref())?)
.send(),
)
.await
.map_err(|_timeout| HttpClientError::RequestTimeout)??
};
#[cfg(not(target_arch = "wasm32"))]
let res = {
self.reqwest_client
.delete(self.base_url.join(endpoint.as_ref())?)
.send()
.await?
};
parse_response(res, false).await
}
}
// define those methods on the trait for nicer extensions (and not having to type the thing twice)
@@ -412,6 +498,17 @@ pub trait ApiClient {
V: AsRef<str> + Sync,
E: Display + DeserializeOwned;
async fn delete_json<T, K, V, E>(
&self,
path: PathSegments<'_>,
params: Params<'_, K, V>,
) -> Result<T, HttpClientError<E>>
where
for<'a> T: Deserialize<'a>,
K: AsRef<str> + Sync,
V: AsRef<str> + Sync,
E: Display + DeserializeOwned;
/// `get` json data from the provided absolute endpoint, i.e. for example `"/api/v1/mixnodes?since=12345"`
async fn get_json_from<T, S, E>(&self, endpoint: S) -> Result<T, HttpClientError<E>>
where
@@ -429,6 +526,12 @@ pub trait ApiClient {
for<'a> T: Deserialize<'a>,
E: Display + DeserializeOwned,
S: AsRef<str> + Sync + Send;
async fn delete_json_from<T, S, E>(&self, endpoint: S) -> Result<T, HttpClientError<E>>
where
for<'a> T: Deserialize<'a>,
E: Display + DeserializeOwned,
S: AsRef<str> + Sync + Send;
}
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
@@ -464,6 +567,20 @@ impl ApiClient for Client {
self.post_json(path, params, json_body).await
}
async fn delete_json<T, K, V, E>(
&self,
path: PathSegments<'_>,
params: Params<'_, K, V>,
) -> Result<T, HttpClientError<E>>
where
for<'a> T: Deserialize<'a>,
K: AsRef<str> + Sync,
V: AsRef<str> + Sync,
E: Display + DeserializeOwned,
{
self.delete_json(path, params).await
}
async fn get_json_from<T, S, E>(&self, endpoint: S) -> Result<T, HttpClientError<E>>
where
for<'a> T: Deserialize<'a>,
@@ -486,6 +603,15 @@ impl ApiClient for Client {
{
self.post_json_endpoint(endpoint, json_body).await
}
async fn delete_json_from<T, S, E>(&self, endpoint: S) -> Result<T, HttpClientError<E>>
where
for<'a> T: Deserialize<'a>,
E: Display + DeserializeOwned,
S: AsRef<str> + Sync + Send,
{
self.delete_json_endpoint(endpoint).await
}
}
// utility function that should solve the double slash problem in API urls forever.
+3
View File
@@ -11,4 +11,7 @@ rust-version.workspace = true
readme.workspace = true
[dependencies]
anyhow = { workspace = true }
bincode = { workspace = true }
nym-crypto = { path = "../crypto", features = ["asymmetric", "serde"] }
serde = { workspace = true, features = ["derive"] }
+94
View File
@@ -1,7 +1,101 @@
use nym_crypto::asymmetric::ed25519::{PublicKey, Signature, SignatureError};
use serde::{Deserialize, Serialize};
pub mod get_testrun {
use super::*;
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct Payload {
pub agent_public_key: PublicKey,
pub timestamp: i64,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct GetTestrunRequest {
pub payload: Payload,
pub signature: Signature,
}
impl SignedRequest for GetTestrunRequest {
type Payload = Payload;
fn public_key(&self) -> &PublicKey {
&self.payload.agent_public_key
}
fn signature(&self) -> &Signature {
&self.signature
}
fn payload(&self) -> &Self::Payload {
&self.payload
}
}
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct TestrunAssignment {
pub testrun_id: i64,
pub assigned_at_utc: i64,
pub gateway_identity_key: String,
}
pub mod submit_results {
use super::*;
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct Payload {
pub probe_result: String,
pub agent_public_key: PublicKey,
pub assigned_at_utc: i64,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct SubmitResults {
pub payload: Payload,
pub signature: Signature,
}
impl SignedRequest for SubmitResults {
type Payload = Payload;
fn public_key(&self) -> &PublicKey {
&self.payload.agent_public_key
}
fn signature(&self) -> &Signature {
&self.signature
}
fn payload(&self) -> &Self::Payload {
&self.payload
}
}
}
pub trait SignedRequest {
type Payload: serde::Serialize;
fn public_key(&self) -> &PublicKey;
fn signature(&self) -> &Signature;
fn payload(&self) -> &Self::Payload;
}
pub trait VerifiableRequest: SignedRequest {
type Error: From<bincode::Error> + From<SignatureError>;
fn verify_signature(&self) -> Result<(), Self::Error> {
bincode::serialize(self.payload())
.map_err(Self::Error::from)
.and_then(|serialized| {
self.public_key()
.verify(serialized, self.signature())
.map_err(Self::Error::from)
})
}
}
impl<T> VerifiableRequest for T
where
T: SignedRequest,
{
type Error = anyhow::Error;
}
+40
View File
@@ -157,6 +157,46 @@ impl NymNetworkDetails {
}
}
#[rustfmt::skip]
#[cfg(feature = "env")]
pub fn export_to_env(self) {
use crate::var_names;
use std::env::set_var;
fn set_optional_var(var_name: &str, value: Option<String>) {
if let Some(value) = value {
set_var(var_name, value);
}
}
set_var(var_names::NETWORK_NAME, self.network_name);
set_var(var_names::BECH32_PREFIX, self.chain_details.bech32_account_prefix);
set_var(var_names::MIX_DENOM, self.chain_details.mix_denom.base);
set_var(var_names::MIX_DENOM_DISPLAY, self.chain_details.mix_denom.display);
set_var(var_names::STAKE_DENOM, self.chain_details.stake_denom.base);
set_var(var_names::STAKE_DENOM_DISPLAY, self.chain_details.stake_denom.display);
set_var(var_names::DENOMS_EXPONENT, self.chain_details.mix_denom.display_exponent.to_string());
if let Some(e) = self.endpoints.first() {
set_var(var_names::NYXD, e.nyxd_url.clone());
set_optional_var(var_names::NYM_API, e.api_url.clone());
set_optional_var(var_names::NYXD_WEBSOCKET, e.websocket_url.clone());
}
set_optional_var(var_names::MIXNET_CONTRACT_ADDRESS, self.contracts.mixnet_contract_address);
set_optional_var(var_names::VESTING_CONTRACT_ADDRESS, self.contracts.vesting_contract_address);
set_optional_var(var_names::ECASH_CONTRACT_ADDRESS, self.contracts.ecash_contract_address);
set_optional_var(var_names::GROUP_CONTRACT_ADDRESS, self.contracts.group_contract_address);
set_optional_var(var_names::MULTISIG_CONTRACT_ADDRESS, self.contracts.multisig_contract_address);
set_optional_var(var_names::COCONUT_DKG_CONTRACT_ADDRESS, self.contracts.coconut_dkg_contract_address);
set_optional_var(var_names::EXPLORER_API, self.explorer_api);
set_optional_var(var_names::NYM_VPN_API, self.nym_vpn_api_url);
}
pub fn default_gas_price_amount(&self) -> f64 {
GAS_PRICE_AMOUNT
}
+14 -4
View File
@@ -29,6 +29,9 @@ pub struct NodeTester<R> {
packet_size: PacketSize,
/// Specify whether route selection should be determined by the packet header.
deterministic_route_selection: bool,
/// Average delay a data packet is going to get delay at a single mixnode.
average_packet_delay: Duration,
@@ -48,11 +51,13 @@ impl<R> NodeTester<R>
where
R: Rng + CryptoRng,
{
#[allow(clippy::too_many_arguments)]
pub fn new(
rng: R,
base_topology: NymTopology,
self_address: Option<Recipient>,
packet_size: PacketSize,
deterministic_route_selection: bool,
average_packet_delay: Duration,
average_ack_delay: Duration,
ack_key: Arc<AckKey>,
@@ -62,6 +67,7 @@ where
base_topology,
self_address,
packet_size,
deterministic_route_selection,
average_packet_delay,
average_ack_delay,
num_mix_hops: DEFAULT_NUM_MIX_HOPS,
@@ -289,10 +295,18 @@ where
impl<R: CryptoRng + Rng> FragmentPreparer for NodeTester<R> {
type Rng = R;
fn deterministic_route_selection(&self) -> bool {
self.deterministic_route_selection
}
fn rng(&mut self) -> &mut Self::Rng {
&mut self.rng
}
fn nonce(&self) -> i32 {
1
}
fn num_mix_hops(&self) -> u8 {
self.num_mix_hops
}
@@ -304,8 +318,4 @@ impl<R: CryptoRng + Rng> FragmentPreparer for NodeTester<R> {
fn average_ack_delay(&self) -> Duration {
self.average_ack_delay
}
fn nonce(&self) -> i32 {
1
}
}
+28 -10
View File
@@ -18,7 +18,7 @@ use nym_sphinx_params::{PacketType, ReplySurbKeyDigestAlgorithm, DEFAULT_NUM_MIX
use nym_sphinx_types::{Delay, NymPacket};
use nym_topology::{NymTopology, NymTopologyError};
use rand::{CryptoRng, Rng, SeedableRng};
use rand_chacha::ChaCha20Rng;
use rand_chacha::ChaCha8Rng;
use nym_sphinx_chunking::monitoring;
use std::time::Duration;
@@ -51,6 +51,7 @@ impl From<PreparedFragment> for MixPacket {
pub trait FragmentPreparer {
type Rng: CryptoRng + Rng;
fn deterministic_route_selection(&self) -> bool;
fn rng(&mut self) -> &mut Self::Rng;
fn nonce(&self) -> i32;
fn num_mix_hops(&self) -> u8;
@@ -201,9 +202,7 @@ pub trait FragmentPreparer {
// could perform diffie-hellman with its own keys followed by a kdf to re-derive
// the packet encryption key
let seed = fragment.seed().wrapping_mul(self.nonce());
let mut rng = ChaCha20Rng::seed_from_u64(seed as u64);
let fragment_header = fragment.header();
let destination = packet_recipient.gateway();
let hops = mix_hops.unwrap_or(self.num_mix_hops());
monitoring::fragment_sent(&fragment, self.nonce(), *destination, hops);
@@ -241,8 +240,18 @@ pub trait FragmentPreparer {
};
// generate pseudorandom route for the packet
log::trace!("Preparing chunk for sending with {} mix hops", hops);
let route = topology.random_route_to_gateway(&mut rng, hops, destination)?;
log::trace!("Preparing chunk for sending with {hops} mix hops");
let route = if self.deterministic_route_selection() {
log::trace!("using deterministic route selection");
let seed = fragment_header.seed().wrapping_mul(self.nonce());
let mut rng = ChaCha8Rng::seed_from_u64(seed as u64);
topology.random_route_to_gateway(&mut rng, hops, destination)?
} else {
log::trace!("using pseudorandom route selection");
let mut rng = self.rng();
topology.random_route_to_gateway(&mut rng, hops, destination)?
};
let destination = packet_recipient.as_sphinx_destination();
// including set of delays
@@ -313,6 +322,9 @@ pub struct MessagePreparer<R> {
/// Instance of a cryptographically secure random number generator.
rng: R,
/// Specify whether route selection should be determined by the packet header.
deterministic_route_selection: bool,
/// Address of this client which also represent an address to which all acknowledgements
/// and surb-based are going to be sent.
sender_address: Recipient,
@@ -336,6 +348,7 @@ where
{
pub fn new(
rng: R,
deterministic_route_selection: bool,
sender_address: Recipient,
average_packet_delay: Duration,
average_ack_delay: Duration,
@@ -344,6 +357,7 @@ where
let nonce = rng.gen();
MessagePreparer {
rng,
deterministic_route_selection,
sender_address,
average_packet_delay,
average_ack_delay,
@@ -457,10 +471,18 @@ where
impl<R: CryptoRng + Rng> FragmentPreparer for MessagePreparer<R> {
type Rng = R;
fn deterministic_route_selection(&self) -> bool {
self.deterministic_route_selection
}
fn rng(&mut self) -> &mut Self::Rng {
&mut self.rng
}
fn nonce(&self) -> i32 {
self.nonce
}
fn num_mix_hops(&self) -> u8 {
self.num_mix_hops
}
@@ -472,10 +494,6 @@ impl<R: CryptoRng + Rng> FragmentPreparer for MessagePreparer<R> {
fn average_ack_delay(&self) -> Duration {
self.average_ack_delay
}
fn nonce(&self) -> i32 {
self.nonce
}
}
/*
+11
View File
@@ -162,6 +162,13 @@ pub struct TrafficWasm {
/// a loop cover message is sent instead in order to preserve the rate.
pub message_sending_average_delay_ms: u32,
/// Specify how many times particular packet can be retransmitted
/// None - no limit
pub maximum_number_of_retransmissions: Option<u32>,
/// Specify whether route selection should be determined by the packet header.
pub deterministic_route_selection: bool,
/// Controls whether the main packet stream constantly produces packets according to the predefined
/// poisson distribution.
pub disable_main_poisson_packet_distribution: bool,
@@ -196,6 +203,8 @@ impl From<TrafficWasm> for ConfigTraffic {
message_sending_average_delay: Duration::from_millis(
traffic.message_sending_average_delay_ms as u64,
),
deterministic_route_selection: traffic.deterministic_route_selection,
maximum_number_of_retransmissions: traffic.maximum_number_of_retransmissions,
disable_main_poisson_packet_distribution: traffic
.disable_main_poisson_packet_distribution,
primary_packet_size: PacketSize::RegularPacket,
@@ -211,6 +220,8 @@ impl From<ConfigTraffic> for TrafficWasm {
average_packet_delay_ms: traffic.average_packet_delay.as_millis() as u32,
message_sending_average_delay_ms: traffic.message_sending_average_delay.as_millis()
as u32,
deterministic_route_selection: traffic.deterministic_route_selection,
maximum_number_of_retransmissions: traffic.maximum_number_of_retransmissions,
disable_main_poisson_packet_distribution: traffic
.disable_main_poisson_packet_distribution,
use_extended_packet_size: traffic.secondary_packet_size.is_some(),
@@ -88,6 +88,14 @@ pub struct TrafficWasmOverride {
#[tsify(optional)]
pub message_sending_average_delay_ms: Option<u32>,
/// Specify how many times particular packet can be retransmitted
#[tsify(optional)]
pub maximum_number_of_retransmissions: Option<u32>,
/// Specify whether route selection should be determined by the packet header.
#[tsify(optional)]
pub deterministic_route_selection: Option<bool>,
/// Controls whether the main packet stream constantly produces packets according to the predefined
/// poisson distribution.
#[tsify(optional)]
@@ -113,6 +121,10 @@ impl From<TrafficWasmOverride> for TrafficWasm {
message_sending_average_delay_ms: value
.message_sending_average_delay_ms
.unwrap_or(def.message_sending_average_delay_ms),
maximum_number_of_retransmissions: value.maximum_number_of_retransmissions,
deterministic_route_selection: value
.deterministic_route_selection
.unwrap_or(def.deterministic_route_selection),
disable_main_poisson_packet_distribution: value
.disable_main_poisson_packet_distribution
.unwrap_or(def.disable_main_poisson_packet_distribution),
@@ -1 +1 @@
Tuesday, November 19th 2024, 11:34:21 UTC
Wednesday, November 20th 2024, 15:53:00 UTC
@@ -1,4 +1,5 @@
```sh
2024-11-20T15:53:01.702819Z  INFO nym-api/src/main.rs:40: Starting nym api...
Usage: nym-api [OPTIONS] <COMMAND>
Commands:
@@ -9,9 +10,9 @@ Commands:
Options:
-c, --config-env-file <CONFIG_ENV_FILE>
Path pointing to an env file that configures the Nym API
Path pointing to an env file that configures the Nym API [env: NYMAPI_CONFIG_ENV_FILE_ARG=]
--no-banner
A no-op flag included for consistency with other binaries (and compatibility with nymvisor, oops)
A no-op flag included for consistency with other binaries (and compatibility with nymvisor, oops) [env: NYMAPI_NO_BANNER_ARG=]
-h, --help
Print help
-V, --version
@@ -44,6 +44,8 @@ Options:
Specify whether detailed system crypto hardware information should be exposed. default: true [env: NYMNODE_HTTP_EXPOSE_CRYPTO_HARDWARE=] [possible values: true, false]
--mixnet-bind-address <MIXNET_BIND_ADDRESS>
Address this node will bind to for listening for mixnet packets default: `0.0.0.0:1789` [env: NYMNODE_MIXNET_BIND_ADDRESS=]
--mixnet-announce-port <MIXNET_ANNOUNCE_PORT>
If applicable, custom port announced in the self-described API that other clients and nodes will use. Useful when the node is behind a proxy [env: NYMNODE_MIXNET_ANNOUNCE_PORT=]
--nym-api-urls <NYM_API_URLS>
Addresses to nym APIs from which the node gets the view of the network [env: NYMNODE_NYM_APIS=]
--nyxd-urls <NYXD_URLS>
@@ -52,14 +54,14 @@ Options:
Specifies whether the wireguard service is enabled on this node [env: NYMNODE_WG_ENABLED=] [possible values: true, false]
--wireguard-bind-address <WIREGUARD_BIND_ADDRESS>
Socket address this node will use for binding its wireguard interface. default: `0.0.0.0:51822` [env: NYMNODE_WG_BIND_ADDRESS=]
--wireguard-private-ip <WIREGUARD_PRIVATE_IP>
Private IP address of the wireguard gateway. default: `10.1.0.1` [env: NYMNODE_WG_IP=]
--wireguard-announced-port <WIREGUARD_ANNOUNCED_PORT>
Port announced to external clients wishing to connect to the wireguard interface. Useful in the instances where the node is behind a proxy [env: NYMNODE_WG_ANNOUNCED_PORT=]
--wireguard-private-network-prefix <WIREGUARD_PRIVATE_NETWORK_PREFIX>
The prefix denoting the maximum number of the clients that can be connected via Wireguard. The maximum value for IPv4 is 32 and for IPv6 is 128 [env: NYMNODE_WG_PRIVATE_NETWORK_PREFIX=]
--verloc-bind-address <VERLOC_BIND_ADDRESS>
Socket address this node will use for binding its verloc API. default: `0.0.0.0:1790` [env: NYMNODE_VERLOC_BIND_ADDRESS=]
--verloc-announce-port <VERLOC_ANNOUNCE_PORT>
If applicable, custom port announced in the self-described API that other clients and nodes will use. Useful when the node is behind a proxy [env: NYMNODE_VERLOC_ANNOUNCE_PORT=]
--entry-bind-address <ENTRY_BIND_ADDRESS>
Socket address this node will use for binding its client websocket API. default: `0.0.0.0:9000` [env: NYMNODE_ENTRY_BIND_ADDRESS=]
--announce-ws-port <ANNOUNCE_WS_PORT>
+12
View File
@@ -772,6 +772,18 @@ const config = {
permanent: true,
basePath: false,
},
{
source: "/operators/troubleshooting/nodes.html",
destination: "/docs/operators/troubleshooting/nodes",
permanent: true,
basePath: false,
},
{
source: "/operators/nodes/manual-upgrade.html",
destination: "/docs/operators/nodes/maintenance/manual-upgrade",
permanent: true,
basePath: false,
},
{
source: "/developers/faq/general-faq.html",
destination: "/docs/operators/faq/general-faq",
@@ -14,4 +14,4 @@ MultiSURBs were implemented in `v1.1.4`. Clients, when sending a message to anot
If a reply is too large still (i.e. it would use more space than the available combined payload of the SURBs sent with the original message), the receiver will use a SURB to ask the sender for more SURBs.
You can read more about SURBs in §4.5 of the [Nym Whitepaper](https://nymtech.net/nym-whitepaper.pdf) as well as the [SURB format page](../traffic/anonymous-replies).
You can read more about SURBs in §4.5 of the [Nym Whitepaper](https://nymtech.net/nym-whitepaper.pdf) as well as the [SURB traffic page](../traffic/anonymous-replies) to learn more about how SURBs are used in the Mixnet, known attacks relying on SURBs, etc.
@@ -1,13 +0,0 @@
# Anonymous Replies with SURBs
> SURBs are the Sphinx equivalent of "onion addresses" in Tor, with the caveat that a SURB can only
be used once (to prevent replay attacks) and within its epoch of validity (the mix node public keys used to
prepare the SURB are only valid for a limited period).
> ...
> A SURB effectively contains: (1) the encrypted headers of a Sphinx message that, if sent to the mixnet, will be routed back to the original sender; (2) the address of the first-layer mix node where the message should be sent; and (3) a cryptographic key to encrypt the reply payload.
>
> [Nym Whitepaper](https://nymtech.net/nym-whitepaper.pdf) §4.6
As outlined in the [concepts](../concepts/anonymous-replies) section, SURBs are layer encrypted sets of Sphinx headers detailing a reply path ending in the sending client's [Nym address](../traffic/addressing-system). Clients receiving messages with SURBs attached are able to write a payload to the provided headers without ever learning about anything other than the first hop back into the Mixnet - the Gateway they are currently connected to.
There is a balance to be struck between the amount of SURBs to compute to send along with messages (which takes computation resources) and not sending enough SURBs initially, thus having to wait for a SURB to be sent from the receiver to the sender, requesting more SURBs be sent.
@@ -0,0 +1,78 @@
import { Callout } from 'nextra/components'
# Anonymous Replies with SURBs
> SURBs are pre-computed Sphinx packet headers encoding a mixnet route that ends in the participant that created the SURB. A sender can generate one or more SURBs and include them in their Sphinx message to a recipient. The recipient can use the SURBs as Sphinx headers to send back replies or acknowledgements that anonymously reach back the original sender after going through the mixnet.
>
> SURBs are the Sphinx equivalent of "onion addresses" in Tor, with the caveat that a SURB can only be used once (to prevent replay attacks) and within its epoch of validity (the mix node public keys used to prepare the SURB are only valid for a limited period). SURB headers are encrypted by the sender, so the recipient sending it back cannot infer from it any information about the message route, the per-hop latency, or the senders address, which is encoded in the innermost (last) routing layer of the SURB. SURBs ('Single Use Reply Blocks') allow clients to reply to incoming messages anonymously.
>
> ...
>
> A SURB effectively contains: (1) the encrypted headers of a Sphinx message that, if sent to the mixnet, will be routed back to the original sender; (2) the address of the first-layer mix node where the message should be sent; and (3) a cryptographic key to encrypt the reply payload.
> [Nym Whitepaper](https://nymtech.net/nym-whitepaper.pdf) §4.5
As outlined in the [concepts](../concepts/anonymous-replies) section, SURBs are layer encrypted sets of Sphinx headers detailing a reply path ending in the sending client's [Nym address](../traffic/addressing-system). Clients receiving messages with SURBs attached are able to write a payload to the provided headers without ever learning about anything other than the first hop back into the Mixnet - the Gateway they (the sender of the reply) are currently registered with.
Put simply, client A sends client B a request of some kind. Before sending the request through the Mixnet, it creates a number of SURBs and sends those along with the request. Client B writes the response to the request to the payload of the SURBs, and then sends these through the Mixnet. Since Sphinx packets are multiply route-encrypted, the first destination of the packet is the `nym-node` running as an Entry Gateway that client B has registered with already, so no information regarding either the path through the Mixnet or the destination of the SURBs (aka the Nym address of client A) is revealed to client B.
## Anatomy of a SURB
Diagramatic representation coming soon™️.
## Sender Tags
For a session between two clients, the sending client generates a random alphanumeric string, referred to as a `sender tag` which is sent along with the SURBs to the receiver of its message(s). The `sender tag` is generated randomly, and does not refer in any way to any identifiers of the sending client.
This is done so that receiving clients have some way of differentiating incoming SURBs from multiple clients and can split them into different 'buckets' in order to facilitate concurrent anonymous replies.
## Replenishing SURBs
Since each SURB is just a pre-computed Sphinx packet header, and Sphinx packets only have a finite payload size, then the size of a possible reply for the amount of SURBs sent with a request is `# SURBs * payload size in bytes`. However it is often the case that replies may be variable in size and larger than the alloted payload size, and/or the sending client did not want to compute many Sphinx headers to send with its initial request.
As such, when a client is running out of SURBs to use for replying, it will use a SURB to send a request to the initial sending client. This request is a request for more SURBs to be sent to it.
```mermaid
---
config:
theme: neo-dark
layout: elk
---
sequenceDiagram
participant Nym Client (Sender)
participant Mixnet Nodes
participant Nym Client (Receiver)
Nym Client (Sender) ->> Nym Client (Sender): Create Sphinx packets with request payload & Sphinx packets with N SURBs
Nym Client (Sender) ->> Mixnet Nodes: Sphinx packets (with payload + with SURBs)
Mixnet Nodes ->> Nym Client (Receiver): Sphinx packets (with payload + with SURBs)
Nym Client (Receiver) ->> Nym Client (Receiver): Decrypt packets
Nym Client (Receiver) ->> Nym Client (Receiver): Perform computation
Nym Client (Receiver) ->> Nym Client (Receiver): Prepare response - problem! Response requires > N * SURB payloads
Nym Client (Receiver) ->> Mixnet Nodes: Sphinx packets (SURB request)
Mixnet Nodes ->> Nym Client (Sender): Sphinx packets (SURB request)
Nym Client (Sender) ->> Nym Client (Sender): Create more SURBs
Nym Client (Sender) ->> Mixnet Nodes: Sphinx packets (SURBs)
Mixnet Nodes ->> Nym Client (Receiver): Sphinx packets (SURBs)
Nym Client (Receiver) ->> Nym Client (Receiver): Write response to SURB payloads
Nym Client (Receiver) ->> Mixnet Nodes: SURBs with payload
Mixnet Nodes ->> Nym Client (Sender): SURBs with payload
Nym Client (Sender) ->> Nym Client (Sender): Decrypt - anonymous response received
```
There is a balance to be struck between the amount of SURBs to compute to send along with messages (aka the sending client A spending computation resources) and not sending enough SURBs initially, thus having to wait for a SURB to be sent from the receiving client B to client A requesting more SURBs be sent, which themselves have to be then sent through the Mixnet to client B to be written to and sent back through the Mixnet again.
If you are able to spend the extra resources upfront and send a lot of SURBs, fewer trips through the Mixnet are required for traffic to go back and forth. However, bear in mind that SURBs in the future will have a finite lifespan (see the section below) so precomputing a very large number to send with the initial message (assuming you are expecting a large reponse, or several messages back and forth) will not work. Furthermore, sending huge amounts of SURBs might open your app up to [possible attacks](#anatomy-of-a-surb).
## SURB Lifetimes
<Callout type="warning">
At the time of writing, SURBs themselves are valid indefinitely, but clients purge their local DB of SURBs that are older than a day on restart. SURBs are valid between topology changes over epochs as `nym-nodes` have a single static publick key, so unless a node goes offline, SURBs will still work even after a topology change.
We still have a few features to add to the Mixnet to add some extra security which will dramatically limit the amount of time SURBs are valid for, but will increase the overall security of the network, so do **not** build with the current status in mind. Instead, check out the information below, and build with this in mind.
</Callout>
Once node key rotation (part of the larger [forward secrecy](https://en.wikipedia.org/wiki/Forward_secrecy) work) and [replay protection](https://www.kaspersky.com/resource-center/definitions/replay-attack) is implemented, SURBs will only be valid for the length of the key epoch (aka for the length of time a `nym-node` retains a particular public key between rotations). The length of the key epoch is still to be decided.
Although this means that there will probably be more back-and-forth between clients sending large volumes of traffic, this will make the network more secure overall.
## Known Attacks Using SURBs
There is a known attack in which a malicious service provider / client continually requests that a sending client sends more and more SURBs to them, accruing a large number of them. The attacker then sends all SURBs back to the sending clients at once in order to try and see permutations in the traffic exiting the Mixnet to the sending client, in order to work out which Gateway they are using as their Entry Gateway.
This attack however relies upon the attacker already being able to actively scan all `nym-node`s running as Gateways and capture that traffic, as well as (once the [zk-nym scheme]() is enabled, spend money to send that traffic through the Mixnet. Furthermore, this is an _active attack_ and requires a client to be either be running a malicious service, or be in a position to request multiple bundles of SURBs from clients via some service.
@@ -1309,7 +1309,7 @@ Every `nym-node` should be upgraded to the latest version! Operators can test us
**`nym-node`**
- Make sure to fill in basic description info, into the file located at `.nym/nym-nodes/<ID>/data/description.toml` (all nodes)
- Configure wireguard routing with new [`network_tunnel_manager.sh`](https://gist.github.com/tommyv1987/ccf6ca00ffb3d7e13192edda61bb2a77) following [these steps](nodes/configuration.md#routing-configuration) (Gateways only for the time being)
- Configure wireguard routing with new [`network_tunnel_manager.sh`](https://github.com/nymtech/nym/blob/develop/scripts/network_tunnel_manager.sh) following [these steps](nodes/configuration.md#routing-configuration) (Gateways only for the time being)
- Enable Wireguard with `--wireguard-enabled true` flag included in your run command (Gateways only for the time being)
- Note: On some VPS this setup may not be enough to get the correct results as some ISPs have their own security groups setup below the individual VPS. In that case a ticket to ISP will have to be issued to open the needed settings. We are working on a template for such ticket.
- Setup [reverse proxy and WSS](nodes/proxy-configuration.md) on `nym-node` (Gateways only for the time being)
@@ -1551,9 +1551,9 @@ ufw status
**Step 2: Download and Prepare the Network Tunnel Manager Script**
1. Download the [`network_tunnel_manager.sh`](https://gist.github.com/tommyv1987/ccf6ca00ffb3d7e13192edda61bb2a77) script:
1. Download the [`network_tunnel_manager.sh`](https://github.com/nymtech/nym/blob/develop/scripts/network_tunnel_manager.sh) script:
```bash
curl -L -o network_tunnel_manager.sh https://gist.githubusercontent.com/tommyv1987/ccf6ca00ffb3d7e13192edda61bb2a77/raw/3c0a38c1416f8fdf22906c013299dd08d1497183/network_tunnel_manager.sh
curl -L -o network_tunnel_manager.sh https://github.com/nymtech/nym/blob/develop/scripts/network_tunnel_manager.sh
```
2. Make the script executable:
@@ -268,7 +268,7 @@ Make sure to keep your IPv4 address enabled while setting up IPv6, as the majori
### Routing Configuration
While we're working on Rust implementation to have these settings as a part of the binary build, to solve these connectivity requirements in the meantime we wrote a script [`network_tunnel_manager.sh`](https://gist.github.com/tommyv1987/ccf6ca00ffb3d7e13192edda61bb2a77) to support operators to configure their servers and address all the connectivity requirements.
While we're working on Rust implementation to have these settings as a part of the binary build, to solve these connectivity requirements in the meantime we wrote a script [`network_tunnel_manager.sh`](https://github.com/nymtech/nym/blob/develop/scripts/network_tunnel_manager.sh) to support operators to configure their servers and address all the connectivity requirements.
Networking configuration across different ISPs and various operation systems does not have a generic solution. If the provided configuration setup doesn't solve your problem check out [IPv6 troubleshooting](../../troubleshooting/vps-isp.mdx#ipv6-troubleshooting) page. Be aware that you may have to do more research, customised adjustments or contact your ISP to change settings for your VPS.
@@ -285,7 +285,7 @@ The script should be used in a context where `nym-node` is running to fully util
###### 1. Download `network_tunnel_manager.sh`, make executable and run:
```sh
curl -L https://gist.githubusercontent.com/tommyv1987/ccf6ca00ffb3d7e13192edda61bb2a77/raw/74241cc06492b955e582052939090f57a285a65e/network_tunnel_manager.sh -o network_tunnel_manager.sh && \
curl -L https://raw.githubusercontent.com/nymtech/nym/refs/heads/develop/scripts/network_tunnel_manager.sh -o network_tunnel_manager.sh && \
chmod +x network_tunnel_manager.sh && \
./network_tunnel_manager.sh
```
+1
View File
@@ -20,6 +20,7 @@ bloomfilter = { workspace = true }
cfg-if = { workspace = true }
clap = { workspace = true, features = ["cargo", "derive", "env"] }
console-subscriber = { workspace = true, optional = true } # validator-api needs to be built with RUSTFLAGS="--cfg tokio_unstable"
dashmap = { workspace = true }
dirs = { workspace = true }
futures = { workspace = true }
itertools = { workspace = true }
+2 -2
View File
@@ -13,7 +13,7 @@ use crate::node_status_api::NodeStatusCache;
use crate::nym_contract_cache::cache::NymContractCache;
use crate::support::caching::cache::SharedCache;
use crate::support::config;
use crate::support::http::state::AppState;
use crate::support::http::state::{AppState, ForcedRefresh};
use crate::support::storage::NymApiStorage;
use async_trait::async_trait;
use axum::Router;
@@ -1264,7 +1264,7 @@ struct TestFixture {
impl TestFixture {
fn build_app_state(storage: NymApiStorage) -> AppState {
AppState {
forced_refresh: Default::default(),
forced_refresh: ForcedRefresh::new(true),
nym_contract_cache: NymContractCache::new(),
node_status_cache: NodeStatusCache::new(),
circulating_supply_cache: CirculatingSupplyCache::new("unym".to_owned()),
+1 -5
View File
@@ -14,7 +14,6 @@ use nym_sphinx::params::PacketType;
use nym_sphinx::receiver::MessageReceiver;
use nym_task::TaskClient;
use std::collections::{HashMap, HashSet};
use std::process;
use tokio::time::{sleep, Duration, Instant};
use tracing::{debug, error, info, trace};
@@ -95,10 +94,7 @@ impl<R: MessageReceiver + Send> Monitor<R> {
)
.await
{
error!("Failed to submit monitor run information to the database - {err}",);
// TODO: slightly more graceful shutdown here
process::exit(1);
error!("Failed to submit monitor run information to the database: {err}",);
}
}
@@ -121,6 +121,7 @@ impl PacketPreparer {
test_route.topology().clone(),
self_address,
PacketSize::RegularPacket,
false,
DEFAULT_AVERAGE_PACKET_DELAY,
DEFAULT_AVERAGE_ACK_DELAY,
self.ack_key.clone(),
+37 -12
View File
@@ -57,6 +57,9 @@ pub enum NodeDescribeCacheError {
// TODO: perhaps include more details here like whether key/signature/payload was malformed
#[error("could not verify signed host information for node {node_id}")]
MissignedHostInformation { node_id: NodeId },
#[error("node {node_id} is announcing an illegal ip address")]
IllegalIpAddress { node_id: NodeId },
}
// this exists because I've been moving things around quite a lot and now the place that holds the type
@@ -199,13 +202,18 @@ impl DescribedNodes {
pub struct NodeDescriptionProvider {
contract_cache: NymContractCache,
allow_all_ips: bool,
batch_size: usize,
}
impl NodeDescriptionProvider {
pub(crate) fn new(contract_cache: NymContractCache) -> NodeDescriptionProvider {
pub(crate) fn new(
contract_cache: NymContractCache,
allow_all_ips: bool,
) -> NodeDescriptionProvider {
NodeDescriptionProvider {
contract_cache,
allow_all_ips,
batch_size: DEFAULT_NODE_DESCRIBE_BATCH_SIZE,
}
}
@@ -270,6 +278,7 @@ async fn try_get_client(
async fn try_get_description(
data: RefreshData,
allow_all_ips: bool,
) -> Result<NymNodeDescription, NodeDescribeCacheError> {
let client = try_get_client(&data.host, data.node_id, data.port).await?;
@@ -286,6 +295,12 @@ async fn try_get_description(
});
}
if !allow_all_ips && !host_info.data.check_ips() {
return Err(NodeDescribeCacheError::IllegalIpAddress {
node_id: data.node_id,
});
}
let node_info = query_for_described_data(&client, data.node_id).await?;
let description = node_info.into_node_description(host_info.data);
@@ -357,8 +372,8 @@ impl RefreshData {
self.node_id
}
pub(crate) async fn try_refresh(self) -> Option<NymNodeDescription> {
match try_get_description(self).await {
pub(crate) async fn try_refresh(self, allow_all_ips: bool) -> Option<NymNodeDescription> {
match try_get_description(self, allow_all_ips).await {
Ok(description) => Some(description),
Err(err) => {
debug!("failed to obtain node self-described data: {err}");
@@ -412,11 +427,15 @@ impl CacheItemProvider for NodeDescriptionProvider {
}
}
let nodes = stream::iter(nodes_to_query.into_iter().map(|n| n.try_refresh()))
.buffer_unordered(self.batch_size)
.filter_map(|x| async move { x.map(|d| (d.node_id, d)) })
.collect::<HashMap<_, _>>()
.await;
let nodes = stream::iter(
nodes_to_query
.into_iter()
.map(|n| n.try_refresh(self.allow_all_ips)),
)
.buffer_unordered(self.batch_size)
.filter_map(|x| async move { x.map(|d| (d.node_id, d)) })
.collect::<HashMap<_, _>>()
.await;
info!("refreshed self described data for {} nodes", nodes.len());
@@ -432,8 +451,11 @@ pub(crate) fn new_refresher(
) -> CacheRefresher<DescribedNodes, NodeDescribeCacheError> {
CacheRefresher::new(
Box::new(
NodeDescriptionProvider::new(contract_cache)
.with_batch_size(config.debug.node_describe_batch_size),
NodeDescriptionProvider::new(
contract_cache,
config.debug.node_describe_allow_illegal_ips,
)
.with_batch_size(config.debug.node_describe_batch_size),
),
config.debug.node_describe_caching_interval,
)
@@ -446,8 +468,11 @@ pub(crate) fn new_refresher_with_initial_value(
) -> CacheRefresher<DescribedNodes, NodeDescribeCacheError> {
CacheRefresher::new_with_initial_value(
Box::new(
NodeDescriptionProvider::new(contract_cache)
.with_batch_size(config.debug.node_describe_batch_size),
NodeDescriptionProvider::new(
contract_cache,
config.debug.node_describe_allow_illegal_ips,
)
.with_batch_size(config.debug.node_describe_batch_size),
),
config.debug.node_describe_caching_interval,
initial,
@@ -88,7 +88,6 @@ pub(crate) async fn submit_gateway_monitoring_results(
match state
.storage
.manager
.submit_gateway_statuses_v2(message.results())
.await
{
@@ -133,7 +132,6 @@ pub(crate) async fn submit_node_monitoring_results(
match state
.storage
.manager
.submit_mixnode_statuses_v2(message.results())
.await
{
+2 -1
View File
@@ -86,8 +86,9 @@ async fn refresh_described(
}
// to make sure you can't ddos the endpoint while a request is in progress
state.forced_refresh.set_last_refreshed(node_id).await;
let allow_all_ips = state.forced_refresh.allow_all_ip_addresses;
if let Some(updated_data) = refresh_data.try_refresh().await {
if let Some(updated_data) = refresh_data.try_refresh(allow_all_ips).await {
let Ok(mut describe_cache) = state.described_nodes_cache.write().await else {
return Err(AxumErrorResponse::service_unavailable());
};
+6 -2
View File
@@ -21,7 +21,9 @@ use crate::status::{ApiStatusState, SignerState};
use crate::support::caching::cache::SharedCache;
use crate::support::config::helpers::try_load_current_config;
use crate::support::config::Config;
use crate::support::http::state::{AppState, ShutdownHandles, TASK_MANAGER_TIMEOUT_S};
use crate::support::http::state::{
AppState, ForcedRefresh, ShutdownHandles, TASK_MANAGER_TIMEOUT_S,
};
use crate::support::http::RouterBuilder;
use crate::support::nyxd;
use crate::support::storage::runtime_migrations::m001_directory_services_v2_1::migrate_to_directory_services_v2_1;
@@ -188,7 +190,9 @@ async fn start_nym_api_tasks_axum(config: &Config) -> anyhow::Result<ShutdownHan
};
let router = router.with_state(AppState {
forced_refresh: Default::default(),
forced_refresh: ForcedRefresh::new(
config.topology_cacher.debug.node_describe_allow_illegal_ips,
),
nym_contract_cache: nym_contract_cache_state.clone(),
node_status_cache: node_status_cache_state.clone(),
circulating_supply_cache: circulating_supply_cache.clone(),
+3
View File
@@ -439,6 +439,8 @@ pub struct TopologyCacherDebug {
pub node_describe_caching_interval: Duration,
pub node_describe_batch_size: usize,
pub node_describe_allow_illegal_ips: bool,
}
impl Default for TopologyCacherDebug {
@@ -447,6 +449,7 @@ impl Default for TopologyCacherDebug {
caching_interval: DEFAULT_TOPOLOGY_CACHE_INTERVAL,
node_describe_caching_interval: DEFAULT_NODE_DESCRIBE_CACHE_INTERVAL,
node_describe_batch_size: DEFAULT_NODE_DESCRIBE_BATCH_SIZE,
node_describe_allow_illegal_ips: false,
}
}
}
+9 -1
View File
@@ -82,12 +82,20 @@ pub(crate) struct AppState {
pub(crate) node_info_cache: unstable::NodeInfoCache,
}
#[derive(Clone, Default)]
#[derive(Clone)]
pub(crate) struct ForcedRefresh {
pub(crate) allow_all_ip_addresses: bool,
pub(crate) refreshes: Arc<RwLock<HashMap<NodeId, OffsetDateTime>>>,
}
impl ForcedRefresh {
pub(crate) fn new(allow_all_ip_addresses: bool) -> ForcedRefresh {
ForcedRefresh {
allow_all_ip_addresses,
refreshes: Arc::new(Default::default()),
}
}
pub(crate) async fn last_refreshed(&self, node_id: NodeId) -> Option<OffsetDateTime> {
self.refreshes.read().await.get(&node_id).copied()
}
+96 -96
View File
@@ -7,6 +7,7 @@ use crate::support::storage::models::{
ActiveGateway, ActiveMixnode, GatewayDetails, HistoricalUptime, MixnodeDetails, NodeStatus,
RewardingReport, TestedGatewayStatus, TestedMixnodeStatus, TestingRoute,
};
use crate::support::storage::DbIdCache;
use nym_mixnet_contract_common::{EpochId, IdentityKey, NodeId};
use nym_types::monitoring::NodeResult;
use sqlx::FromRow;
@@ -51,24 +52,7 @@ impl AvgGatewayReliability {
// all SQL goes here
impl StorageManager {
pub(crate) async fn get_mixnode_mix_ids_by_identity(
&self,
identity: &str,
) -> Result<Vec<NodeId>, sqlx::Error> {
let ids = sqlx::query!(
r#"SELECT mix_id as "mix_id: NodeId" FROM mixnode_details WHERE identity_key = ?"#,
identity
)
.fetch_all(&self.connection_pool)
.await?
.into_iter()
.map(|row| row.mix_id)
.collect();
Ok(ids)
}
pub(crate) async fn get_all_avg_mix_reliability_in_last_24hr(
pub(super) async fn get_all_avg_mix_reliability_in_last_24hr(
&self,
end_ts_secs: i64,
) -> Result<Vec<AvgMixnodeReliability>, sqlx::Error> {
@@ -77,7 +61,7 @@ impl StorageManager {
.await
}
pub(crate) async fn get_all_avg_gateway_reliability_in_last_24hr(
pub(super) async fn get_all_avg_gateway_reliability_in_last_24hr(
&self,
end_ts_secs: i64,
) -> Result<Vec<AvgGatewayReliability>, sqlx::Error> {
@@ -86,7 +70,7 @@ impl StorageManager {
.await
}
pub(crate) async fn get_all_avg_mix_reliability_in_time_interval(
pub(super) async fn get_all_avg_mix_reliability_in_time_interval(
&self,
start_ts_secs: i64,
end_ts_secs: i64,
@@ -114,7 +98,7 @@ impl StorageManager {
Ok(result)
}
pub(crate) async fn get_all_avg_gateway_reliability_in_interval(
pub(super) async fn get_all_avg_gateway_reliability_in_interval(
&self,
start_ts_secs: i64,
end_ts_secs: i64,
@@ -147,7 +131,7 @@ impl StorageManager {
/// # Arguments
///
/// * `mix_id`: mix-id (as assigned by the smart contract) of the mixnode.
pub(crate) async fn get_mixnode_database_id(
pub(super) async fn get_mixnode_database_id(
&self,
mix_id: NodeId,
) -> Result<Option<i64>, sqlx::Error> {
@@ -159,7 +143,7 @@ impl StorageManager {
Ok(id)
}
pub(crate) async fn get_gateway_database_id(
pub(super) async fn get_gateway_database_id(
&self,
node_id: NodeId,
) -> Result<Option<i64>, sqlx::Error> {
@@ -172,7 +156,7 @@ impl StorageManager {
}
/// Tries to obtain row id of given gateway given its identity
pub(crate) async fn get_gateway_database_id_by_identity(
pub(super) async fn get_gateway_database_id_by_identity(
&self,
identity: &str,
) -> Result<Option<i64>, sqlx::Error> {
@@ -187,7 +171,7 @@ impl StorageManager {
Ok(id)
}
pub(crate) async fn get_gateway_node_id_from_identity_key(
pub(super) async fn get_gateway_node_id_from_identity_key(
&self,
identity: &str,
) -> Result<Option<NodeId>, sqlx::Error> {
@@ -202,7 +186,7 @@ impl StorageManager {
Ok(node_id)
}
pub(crate) async fn get_gateway_identity_key(
pub(super) async fn get_gateway_identity_key(
&self,
node_id: NodeId,
) -> Result<Option<IdentityKey>, sqlx::Error> {
@@ -222,7 +206,7 @@ impl StorageManager {
/// # Arguments
///
/// * `mix_id`: mix-id (as assigned by the smart contract) of the mixnode.
pub(crate) async fn get_mixnode_identity_key(
pub(super) async fn get_mixnode_identity_key(
&self,
mix_id: NodeId,
) -> Result<Option<IdentityKey>, sqlx::Error> {
@@ -244,7 +228,7 @@ impl StorageManager {
///
/// * `mix_id`: mix-id (as assigned by the smart contract) of the mixnode.
/// * `timestamp`: unix timestamp of the lower bound of the selection.
pub(crate) async fn get_mixnode_statuses_since(
pub(super) async fn get_mixnode_statuses_since(
&self,
mix_id: NodeId,
timestamp: i64,
@@ -272,7 +256,7 @@ impl StorageManager {
///
/// * `identity`: identity (base58-encoded public key) of the gateway.
/// * `timestamp`: unix timestamp of the lower bound of the selection.
pub(crate) async fn get_gateway_statuses_since(
pub(super) async fn get_gateway_statuses_since(
&self,
node_id: NodeId,
timestamp: i64,
@@ -298,7 +282,7 @@ impl StorageManager {
/// # Arguments
///
/// * `mix_id`: mix-id (as assigned by the smart contract) of the mixnode.
pub(crate) async fn get_mixnode_historical_uptimes(
pub(super) async fn get_mixnode_historical_uptimes(
&self,
mix_id: NodeId,
) -> Result<Vec<ApiHistoricalUptime>, sqlx::Error> {
@@ -336,7 +320,7 @@ impl StorageManager {
/// # Arguments
///
/// * `identity`: identity (base58-encoded public key) of the gateway.
pub(crate) async fn get_gateway_historical_uptimes(
pub(super) async fn get_gateway_historical_uptimes(
&self,
node_id: NodeId,
) -> Result<Vec<ApiHistoricalUptime>, sqlx::Error> {
@@ -369,7 +353,7 @@ impl StorageManager {
Ok(uptimes)
}
pub(crate) async fn get_historical_mix_uptime_on(
pub(super) async fn get_historical_mix_uptime_on(
&self,
contract_node_id: i64,
date: Date,
@@ -393,7 +377,7 @@ impl StorageManager {
.await
}
pub(crate) async fn get_historical_gateway_uptime_on(
pub(super) async fn get_historical_gateway_uptime_on(
&self,
contract_node_id: i64,
date: Date,
@@ -424,7 +408,7 @@ impl StorageManager {
///
/// * `since`: unix timestamp indicating the lower bound interval of the selection.
/// * `until`: unix timestamp indicating the upper bound interval of the selection.
pub(crate) async fn get_mixnode_statuses_by_database_id(
pub(super) async fn get_mixnode_statuses_by_database_id(
&self,
id: i64,
since: i64,
@@ -445,7 +429,7 @@ impl StorageManager {
.await
}
pub(crate) async fn get_mixnode_average_reliability_in_interval(
pub(super) async fn get_mixnode_average_reliability_in_interval(
&self,
id: i64,
start: i64,
@@ -507,7 +491,7 @@ impl StorageManager {
///
/// * `since`: unix timestamp indicating the lower bound interval of the selection.
/// * `until`: unix timestamp indicating the upper bound interval of the selection.
pub(crate) async fn get_gateway_statuses_by_database_id(
pub(super) async fn get_gateway_statuses_by_database_id(
&self,
id: i64,
since: i64,
@@ -534,27 +518,36 @@ impl StorageManager {
///
/// * `timestamp`: unix timestamp indicating when the measurements took place.
/// * `mixnode_results`: reliability results of each node that got tested.
pub(crate) async fn submit_mixnode_statuses(
pub(super) async fn submit_mixnode_statuses(
&self,
timestamp: i64,
mixnode_results: Vec<NodeResult>,
id_cache: &DbIdCache,
) -> Result<(), sqlx::Error> {
// insert it all in a transaction to make sure all nodes are updated at the same time
// (plus it's a nice guard against new nodes)
let mut tx = self.connection_pool.begin().await?;
for mixnode_result in mixnode_results {
let mixnode_id = sqlx::query!(
r#"
INSERT OR IGNORE INTO mixnode_details(mix_id, identity_key) VALUES (?, ?);
SELECT id FROM mixnode_details WHERE mix_id = ?;
"#,
mixnode_result.node_id,
mixnode_result.identity,
mixnode_result.node_id,
)
.fetch_one(&mut *tx)
.await?
.id;
let mixnode_id = match id_cache.mixnode_db_id(mixnode_result.node_id) {
Some(id) => id,
None => {
let mixnode_id = sqlx::query!(
r#"
INSERT OR IGNORE INTO mixnode_details(mix_id, identity_key) VALUES (?, ?);
SELECT id FROM mixnode_details WHERE mix_id = ?;
"#,
mixnode_result.node_id,
mixnode_result.identity,
mixnode_result.node_id,
)
.fetch_one(&mut *tx)
.await?
.id;
id_cache.set_mixnode_db_id(mixnode_result.node_id, mixnode_id);
mixnode_id
}
};
// insert the actual status
sqlx::query!(
@@ -573,7 +566,7 @@ impl StorageManager {
tx.commit().await
}
pub(crate) async fn submit_mixnode_statuses_v2(
pub(super) async fn submit_mixnode_statuses_v2(
&self,
mixnode_results: &[NodeResult],
) -> Result<(), sqlx::Error> {
@@ -620,10 +613,11 @@ impl StorageManager {
///
/// * `timestamp`: unix timestamp indicating when the measurements took place.
/// * `gateway_results`: reliability results of each node that got tested.
pub(crate) async fn submit_gateway_statuses(
pub(super) async fn submit_gateway_statuses(
&self,
timestamp: i64,
gateway_results: Vec<NodeResult>,
id_cache: &DbIdCache,
) -> Result<(), sqlx::Error> {
// insert it all in a transaction to make sure all nodes are updated at the same time
// (plus it's a nice guard against new nodes)
@@ -631,39 +625,45 @@ impl StorageManager {
for gateway_result in gateway_results {
// if gateway info doesn't exist, insert it and get its id
// same ID "problem" as described for mixnode insertion
let gateway_id = sqlx::query!(
r#"
INSERT OR IGNORE INTO gateway_details(node_id, identity) VALUES (?, ?);
SELECT id FROM gateway_details WHERE identity = ?;
"#,
gateway_result.node_id,
gateway_result.identity,
gateway_result.identity,
)
.fetch_one(&mut *tx)
.await?
.id;
let gateway_id = match id_cache.gateway_db_id(gateway_result.node_id) {
Some(id) => id,
None => {
let gateway_id = sqlx::query!(
r#"
INSERT OR IGNORE INTO gateway_details(node_id, identity) VALUES (?, ?);
SELECT id FROM gateway_details WHERE identity = ?;
"#,
gateway_result.node_id,
gateway_result.identity,
gateway_result.identity,
)
.fetch_one(&mut *tx)
.await?
.id;
id_cache.set_gateway_db_id(gateway_result.node_id, gateway_id);
gateway_id
}
};
// insert the actual status
sqlx::query!(
r#"
INSERT INTO gateway_status (gateway_details_id, reliability, timestamp) VALUES (?, ?, ?);
"#,
gateway_id,
gateway_result.reliability,
timestamp
)
.execute(&mut *tx)
.await?;
r#"
INSERT INTO gateway_status (gateway_details_id, reliability, timestamp) VALUES (?, ?, ?);
"#,
gateway_id,
gateway_result.reliability,
timestamp
)
.execute(&mut *tx)
.await?;
}
// finally commit the transaction
tx.commit().await
}
pub(crate) async fn submit_gateway_statuses_v2(
pub(super) async fn submit_gateway_statuses_v2(
&self,
gateway_results: &[NodeResult],
) -> Result<(), sqlx::Error> {
@@ -714,7 +714,7 @@ impl StorageManager {
/// # Arguments
///
/// * `testing_route`: test route used for this particular network monitor run.
pub(crate) async fn submit_testing_route_used(
pub(super) async fn submit_testing_route_used(
&self,
testing_route: TestingRoute,
) -> Result<(), sqlx::Error> {
@@ -742,7 +742,7 @@ impl StorageManager {
///
/// * `db_mixnode_id`: id (as saved in the database) of the mixnode.
/// * `since`: unix timestamp indicating the lower bound interval of the selection.
pub(crate) async fn get_mixnode_testing_route_presence_count_since(
pub(super) async fn get_mixnode_testing_route_presence_count_since(
&self,
db_mixnode_id: i64,
since: i64,
@@ -781,7 +781,7 @@ impl StorageManager {
///
/// * `gateway_id`: id (as saved in the database) of the gateway.
/// * `since`: unix timestamp indicating the lower bound interval of the selection.
pub(crate) async fn get_gateway_testing_route_presence_count_since(
pub(super) async fn get_gateway_testing_route_presence_count_since(
&self,
gateway_id: i64,
since: i64,
@@ -813,7 +813,7 @@ impl StorageManager {
}
/// Checks whether there are already any historical uptimes with this particular date.
pub(crate) async fn check_for_historical_uptime_existence(
pub(super) async fn check_for_historical_uptime_existence(
&self,
today_iso_8601: &str,
) -> Result<bool, sqlx::Error> {
@@ -833,7 +833,7 @@ impl StorageManager {
/// * `node_id`: id of the mixnode (as inserted in `mixnode_details_id` table).
/// * `date`: date associated with the uptime represented in ISO 8601, i.e. YYYY-MM-DD.
/// * `uptime`: the actual uptime of the node during the specified day.
pub(crate) async fn insert_mixnode_historical_uptime(
pub(super) async fn insert_mixnode_historical_uptime(
&self,
mix_id: i64,
date: &str,
@@ -855,7 +855,7 @@ impl StorageManager {
/// * `node_id`: id of the gateway (as inserted in `gateway_details_id` table).
/// * `date`: date associated with the uptime represented in ISO 8601, i.e. YYYY-MM-DD.
/// * `uptime`: the actual uptime of the node during the specified day.
pub(crate) async fn insert_gateway_historical_uptime(
pub(super) async fn insert_gateway_historical_uptime(
&self,
db_id: i64,
date: &str,
@@ -876,7 +876,7 @@ impl StorageManager {
/// # Arguments
///
/// * `timestamp`: unix timestamp at which the monitor test run has occurred
pub(crate) async fn insert_monitor_run(&self, timestamp: i64) -> Result<i64, sqlx::Error> {
pub(super) async fn insert_monitor_run(&self, timestamp: i64) -> Result<i64, sqlx::Error> {
let res = sqlx::query!("INSERT INTO monitor_run(timestamp) VALUES (?)", timestamp)
.execute(&self.connection_pool)
.await?;
@@ -889,7 +889,7 @@ impl StorageManager {
///
/// * `since`: unix timestamp indicating the lower bound interval of the selection.
/// * `until`: unix timestamp indicating the upper bound interval of the selection.
pub(crate) async fn get_monitor_runs_count(
pub(super) async fn get_monitor_runs_count(
&self,
since: i64,
until: i64,
@@ -911,7 +911,7 @@ impl StorageManager {
/// # Arguments
///
/// * `until`: timestamp specifying the purge cutoff.
pub(crate) async fn purge_old_mixnode_statuses(
pub(super) async fn purge_old_mixnode_statuses(
&self,
timestamp: i64,
) -> Result<(), sqlx::Error> {
@@ -927,7 +927,7 @@ impl StorageManager {
/// # Arguments
///
/// * `until`: timestamp specifying the purge cutoff.
pub(crate) async fn purge_old_gateway_statuses(
pub(super) async fn purge_old_gateway_statuses(
&self,
timestamp: i64,
) -> Result<(), sqlx::Error> {
@@ -944,7 +944,7 @@ impl StorageManager {
///
/// * `since`: indicates the lower bound timestamp for deciding whether given mixnode is active
/// * `until`: indicates the upper bound timestamp for deciding whether given mixnode is active
pub(crate) async fn get_all_active_mixnodes_in_interval(
pub(super) async fn get_all_active_mixnodes_in_interval(
&self,
since: i64,
until: i64,
@@ -978,7 +978,7 @@ impl StorageManager {
///
/// * `since`: indicates the lower bound timestamp for deciding whether given gateway is active
/// * `until`: indicates the upper bound timestamp for deciding whether given gateway is active
pub(crate) async fn get_all_active_gateways_in_interval(
pub(super) async fn get_all_active_gateways_in_interval(
&self,
since: i64,
until: i64,
@@ -1025,7 +1025,7 @@ impl StorageManager {
///
/// * `report`: report to insert into the database
#[allow(unused)]
pub(crate) async fn insert_rewarding_report(
pub(super) async fn insert_rewarding_report(
&self,
report: RewardingReport,
) -> Result<(), sqlx::Error> {
@@ -1044,7 +1044,7 @@ impl StorageManager {
}
#[allow(unused)]
pub(crate) async fn get_rewarding_report(
pub(super) async fn get_rewarding_report(
&self,
absolute_epoch_id: EpochId,
) -> Result<Option<RewardingReport>, sqlx::Error> {
@@ -1069,7 +1069,7 @@ impl StorageManager {
///
/// * `since`: unix timestamp indicating the lower bound interval of the selection.
/// * `until`: unix timestamp indicating the upper bound interval of the selection.
pub(crate) async fn get_all_active_mixnodes_statuses_in_interval(
pub(super) async fn get_all_active_mixnodes_statuses_in_interval(
&self,
since: i64,
until: i64,
@@ -1102,7 +1102,7 @@ impl StorageManager {
///
/// * `since`: unix timestamp indicating the lower bound interval of the selection.
/// * `until`: unix timestamp indicating the upper bound interval of the selection.
pub(crate) async fn get_all_active_gateways_statuses_in_interval(
pub(super) async fn get_all_active_gateways_statuses_in_interval(
&self,
since: i64,
until: i64,
@@ -1129,7 +1129,7 @@ impl StorageManager {
Ok(active_day_statuses)
}
pub(crate) async fn get_mixnode_details_by_db_id(
pub(super) async fn get_mixnode_details_by_db_id(
&self,
id: i64,
) -> Result<Option<MixnodeDetails>, sqlx::Error> {
@@ -1142,7 +1142,7 @@ impl StorageManager {
.await
}
pub(crate) async fn get_gateway_details_by_db_id(
pub(super) async fn get_gateway_details_by_db_id(
&self,
id: i64,
) -> Result<Option<GatewayDetails>, sqlx::Error> {
@@ -1154,7 +1154,7 @@ impl StorageManager {
.await
}
pub(crate) async fn get_mixnode_statuses_count(&self, db_id: i64) -> Result<i32, sqlx::Error> {
pub(super) async fn get_mixnode_statuses_count(&self, db_id: i64) -> Result<i32, sqlx::Error> {
sqlx::query!(
r#"
SELECT COUNT(*) as count
@@ -1170,7 +1170,7 @@ impl StorageManager {
.map(|record| record.count)
}
pub(crate) async fn get_mixnode_statuses(
pub(super) async fn get_mixnode_statuses(
&self,
mix_id: NodeId,
limit: u32,
@@ -1206,7 +1206,7 @@ impl StorageManager {
.await
}
pub(crate) async fn get_gateway_statuses_count(&self, db_id: i64) -> Result<i32, sqlx::Error> {
pub(super) async fn get_gateway_statuses_count(&self, db_id: i64) -> Result<i32, sqlx::Error> {
sqlx::query!(
r#"
SELECT COUNT(*) as count
@@ -1222,7 +1222,7 @@ impl StorageManager {
.map(|record| record.count)
}
pub(crate) async fn get_gateway_statuses(
pub(super) async fn get_gateway_statuses(
&self,
gateway_identity: &str,
limit: u32,
+85 -28
View File
@@ -1,6 +1,7 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use self::manager::{AvgGatewayReliability, AvgMixnodeReliability};
use crate::network_monitor::test_route::TestRoute;
use crate::node_status_api::models::{
GatewayStatusReport, GatewayUptimeHistory, HistoricalUptime as ApiHistoricalUptime,
@@ -12,37 +13,71 @@ use crate::storage::models::{NodeStatus, TestingRoute};
use crate::support::storage::models::{
GatewayDetails, HistoricalUptime, MixnodeDetails, TestedGatewayStatus, TestedMixnodeStatus,
};
use dashmap::DashMap;
use nym_mixnet_contract_common::NodeId;
use nym_types::monitoring::NodeResult;
use sqlx::ConnectOptions;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use time::{Date, OffsetDateTime};
use tracing::log::LevelFilter;
use tracing::{error, info, warn};
use self::manager::{AvgGatewayReliability, AvgMixnodeReliability};
pub(crate) mod manager;
pub(crate) mod models;
pub(crate) mod runtime_migrations;
#[derive(Default)]
pub(crate) struct DbIdCache {
pub mixnodes_v1: DashMap<NodeId, i64>,
pub gateways_v1: DashMap<NodeId, i64>,
}
impl DbIdCache {
pub(crate) fn mixnode_db_id(&self, node_id: NodeId) -> Option<i64> {
self.mixnodes_v1.get(&node_id).map(|v| *v)
}
pub(crate) fn gateway_db_id(&self, node_id: NodeId) -> Option<i64> {
self.gateways_v1.get(&node_id).map(|v| *v)
}
pub(crate) fn set_mixnode_db_id(&self, node_id: NodeId, db_id: i64) {
self.mixnodes_v1.insert(node_id, db_id);
}
pub(crate) fn set_gateway_db_id(&self, node_id: NodeId, db_id: i64) {
self.gateways_v1.insert(node_id, db_id);
}
}
// note that clone here is fine as upon cloning the same underlying pool will be used
#[derive(Clone)]
pub(crate) struct NymApiStorage {
pub manager: StorageManager,
pub db_id_cache: Arc<DbIdCache>,
}
impl NymApiStorage {
pub async fn init<P: AsRef<Path>>(database_path: P) -> Result<Self, NymApiStorageError> {
// TODO: we can inject here more stuff based on our nym-api global config
// struct. Maybe different pool size or timeout intervals?
let opts = sqlx::sqlite::SqliteConnectOptions::new()
let connect_opts = sqlx::sqlite::SqliteConnectOptions::new()
.filename(database_path)
.create_if_missing(true)
.disable_statement_logging();
.log_statements(LevelFilter::Trace)
.log_slow_statements(LevelFilter::Warn, Duration::from_millis(250));
// TODO: do we want auto_vacuum ?
let connection_pool = match sqlx::SqlitePool::connect_with(opts).await {
let pool_opts = sqlx::sqlite::SqlitePoolOptions::new()
.min_connections(5)
.max_connections(25)
.acquire_timeout(Duration::from_secs(60));
let connection_pool = match pool_opts.connect_with(connect_opts).await {
Ok(db) => db,
Err(err) => {
error!("Failed to connect to SQLx database: {err}");
@@ -59,32 +94,38 @@ impl NymApiStorage {
let storage = NymApiStorage {
manager: StorageManager { connection_pool },
db_id_cache: Arc::new(Default::default()),
};
Ok(storage)
}
#[allow(unused)]
pub(crate) async fn mix_identity_to_mix_ids(
pub(crate) async fn get_mixnode_database_id(
&self,
identity: &str,
) -> Result<Vec<NodeId>, NymApiStorageError> {
Ok(self
.manager
.get_mixnode_mix_ids_by_identity(identity)
.await?)
node_id: NodeId,
) -> Result<Option<i64>, NymApiStorageError> {
if let Some(cached) = self.db_id_cache.mixnode_db_id(node_id) {
return Ok(Some(cached));
}
if let Some(retrieved) = self.manager.get_mixnode_database_id(node_id).await? {
self.db_id_cache.set_mixnode_db_id(node_id, retrieved);
return Ok(Some(retrieved));
}
Ok(None)
}
#[allow(unused)]
pub(crate) async fn mix_identity_to_latest_mix_id(
pub(crate) async fn get_gateway_database_id(
&self,
identity: &str,
) -> Result<Option<NodeId>, NymApiStorageError> {
Ok(self
.mix_identity_to_mix_ids(identity)
.await?
.into_iter()
.max())
node_id: NodeId,
) -> Result<Option<i64>, NymApiStorageError> {
if let Some(cached) = self.db_id_cache.gateway_db_id(node_id) {
return Ok(Some(cached));
}
if let Some(retrieved) = self.manager.get_gateway_database_id(node_id).await? {
self.db_id_cache.set_gateway_db_id(node_id, retrieved);
return Ok(Some(retrieved));
}
Ok(None)
}
pub(crate) async fn get_all_avg_gateway_reliability_in_last_24hr(
@@ -576,7 +617,6 @@ impl NymApiStorage {
// we MUST have those entries in the database, otherwise the route wouldn't have been chosen
// in the first place
let layer1_mix_db_id = self
.manager
.get_mixnode_database_id(test_route.layer_one_mix().mix_id)
.await?
.ok_or_else(|| NymApiStorageError::DatabaseInconsistency {
@@ -584,7 +624,6 @@ impl NymApiStorage {
})?;
let layer2_mix_db_id = self
.manager
.get_mixnode_database_id(test_route.layer_two_mix().mix_id)
.await?
.ok_or_else(|| NymApiStorageError::DatabaseInconsistency {
@@ -592,7 +631,6 @@ impl NymApiStorage {
})?;
let layer3_mix_db_id = self
.manager
.get_mixnode_database_id(test_route.layer_three_mix().mix_id)
.await?
.ok_or_else(|| NymApiStorageError::DatabaseInconsistency {
@@ -600,7 +638,6 @@ impl NymApiStorage {
})?;
let gateway_db_id = self
.manager
.get_gateway_database_id(test_route.gateway().node_id)
.await?
.ok_or_else(|| NymApiStorageError::DatabaseInconsistency {
@@ -701,11 +738,11 @@ impl NymApiStorage {
let monitor_run_id = self.manager.insert_monitor_run(now).await?;
self.manager
.submit_mixnode_statuses(now, mixnode_results)
.submit_mixnode_statuses(now, mixnode_results, &self.db_id_cache)
.await?;
self.manager
.submit_gateway_statuses(now, gateway_results)
.submit_gateway_statuses(now, gateway_results, &self.db_id_cache)
.await?;
for test_route in test_routes {
@@ -715,6 +752,26 @@ impl NymApiStorage {
Ok(())
}
pub(crate) async fn submit_mixnode_statuses_v2(
&self,
mixnode_results: &[NodeResult],
) -> Result<(), NymApiStorageError> {
self.manager
.submit_mixnode_statuses_v2(mixnode_results)
.await?;
Ok(())
}
pub(crate) async fn submit_gateway_statuses_v2(
&self,
gateway_results: &[NodeResult],
) -> Result<(), NymApiStorageError> {
self.manager
.submit_gateway_statuses_v2(gateway_results)
.await?;
Ok(())
}
/// Obtains number of network monitor test runs that have occurred within the specified interval.
///
/// # Arguments
+1
View File
@@ -6,6 +6,7 @@ RUN cargo build --release
FROM locustio/locust
EXPOSE 8089
EXPOSE 8080
COPY --from=builder /usr/src/nym/target/release/nym-network-monitor /bin/nym-network-monitor
COPY --from=builder /usr/src/nym/nym-network-monitor/locustfile.py locustfile.py
COPY --from=builder /usr/src/nym/nym-network-monitor/entrypoint.sh entrypoint.sh
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-network-monitor"
version = "0.1.0"
version = "1.0.2"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
@@ -13,7 +13,7 @@ license.workspace = true
[dependencies]
anyhow = { workspace = true }
axum = { workspace = true, features = ["json"] }
clap = { workspace = true, features = ["derive"] }
clap = { workspace = true, features = ["derive", "env"] }
dashmap = { workspace = true }
futures = { workspace = true }
log = { workspace = true }
+1 -1
View File
@@ -15,7 +15,7 @@ nnm_pid=$!
sleep 10
python -m locust -H http://127.0.0.1:8080 --processes "${processes}" --autostart --autoquit 60 -u "${users}" -t "${timeout}"s &
python -m locust -H http://${NYM_NETWORK_MONITOR_HOST}:${NYM_NETWORK_MONITOR_PORT} --processes "${processes}" --autostart --autoquit 60 -u "${users}" -t "${timeout}"s &
locust_pid=$!
wait $locust_pid
+2 -2
View File
@@ -107,11 +107,11 @@ struct Args {
client_lifetime: u64,
/// Port to listen on
#[arg(long, default_value_t = 8080)]
#[arg(long, default_value_t = 8080, env = "NYM_NETWORK_MONITOR_PORT")]
port: u16,
/// Host to listen on
#[arg(long, default_value = "127.0.0.1")]
#[arg(long, default_value = "127.0.0.1", env = "NYM_NETWORK_MONITOR_HOST")]
host: String,
/// Path to the topology file
+1
View File
@@ -1 +1,2 @@
nym-gateway-probe
keys/
+9 -1
View File
@@ -4,7 +4,7 @@
[package]
name = "nym-node-status-agent"
version = "0.1.6"
version = "1.0.0-rc.1"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
@@ -16,12 +16,20 @@ readme.workspace = true
[dependencies]
anyhow = { workspace = true}
bincode = { workspace = true }
chrono = { workspace = true }
clap = { workspace = true, features = ["derive", "env"] }
nym-bin-common = { path = "../common/bin-common", features = ["models"]}
nym-common-models = { path = "../common/models" }
nym-crypto = { path = "../common/crypto", features = ["asymmetric", "rand"] }
rand = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt-multi-thread", "process"] }
tokio-util = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true, features = ["env-filter"] }
reqwest = { workspace = true, features = ["json"] }
serde = { workspace = true }
serde_json = { workspace = true }
[dev-dependencies]
tempfile = { workspace = true }
+4
View File
@@ -0,0 +1,4 @@
#!/bin/bash
mkdir -p keys
cargo run --package nym-node-status-agent -- generate-keypair --path keys/private
+21 -18
View File
@@ -1,25 +1,35 @@
#!/bin/bash
set -eu
export ENVIRONMENT=${ENVIRONMENT:-"sandbox"}
environment="qa"
source ../envs/${environment}.env
export RUST_LOG="debug"
probe_git_ref="nym-vpn-core-v1.0.0-rc.6"
crate_root=$(dirname $(realpath "$0"))
gateway_probe_src=$(dirname $(dirname "$crate_root"))/nym-vpn-client/nym-vpn-core
monorepo_root=$(dirname "${crate_root}")
echo "Expecting nym-vpn-client repo at a sibling level of nym monorepo dir"
gateway_probe_src=$(dirname "${monorepo_root}")/nym-vpn-client/nym-vpn-core
echo "gateway_probe_src=$gateway_probe_src"
echo "crate_root=$crate_root"
set -a
source "${monorepo_root}/envs/${ENVIRONMENT}.env"
set +a
export RUST_LOG="info"
export NODE_STATUS_AGENT_SERVER_ADDRESS="http://127.0.0.1"
export NODE_STATUS_AGENT_SERVER_PORT="8000"
export NODE_STATUS_AGENT_PROBE_PATH="$crate_root/nym-gateway-probe"
export NODE_STATUS_AGENT_AUTH_KEY="BjyC9SsHAZUzPRkQR4sPTvVrp4GgaquTh5YfSJksvvWT"
workers=${1:-1}
echo "Running $workers workers in parallel"
# build & copy over GW probe
function copy_gw_probe() {
pushd $gateway_probe_src
git switch main
git pull
git fetch -a
git checkout $probe_git_ref
cargo build --release --package nym-gateway-probe
cp target/release/nym-gateway-probe "$crate_root"
$crate_root/nym-gateway-probe --version
@@ -32,11 +42,8 @@ function build_agent() {
function swarm() {
local workers=$1
echo "Running $workers in parallel"
build_agent
for ((i = 1; i <= $workers; i++)); do
for ((i = 1; i <= workers; i++)); do
../target/release/nym-node-status-agent run-probe &
done
@@ -45,11 +52,7 @@ function swarm() {
echo "All agents completed"
}
export NODE_STATUS_AGENT_SERVER_ADDRESS="http://127.0.0.1"
export NODE_STATUS_AGENT_SERVER_PORT="8000"
copy_gw_probe
build_agent
swarm 8
# cargo run -- run-probe
swarm $workers
-118
View File
@@ -1,118 +0,0 @@
use anyhow::bail;
use clap::{Parser, Subcommand};
use nym_bin_common::bin_info;
use nym_common_models::ns_api::TestrunAssignment;
use std::sync::OnceLock;
use tracing::instrument;
use crate::probe::GwProbe;
// Helper for passing LONG_VERSION to clap
fn pretty_build_info_static() -> &'static str {
static PRETTY_BUILD_INFORMATION: OnceLock<String> = OnceLock::new();
PRETTY_BUILD_INFORMATION.get_or_init(|| bin_info!().pretty_print())
}
#[derive(Parser, Debug)]
#[clap(author = "Nymtech", version, long_version = pretty_build_info_static(), about)]
pub(crate) struct Args {
#[command(subcommand)]
pub(crate) command: Command,
#[arg(short, long, env = "NODE_STATUS_AGENT_SERVER_ADDRESS")]
pub(crate) server_address: String,
#[arg(short = 'p', long, env = "NODE_STATUS_AGENT_SERVER_PORT")]
pub(crate) server_port: u16,
// TODO dz accept keypair for identification / auth
}
#[derive(Subcommand, Debug)]
pub(crate) enum Command {
RunProbe {
/// path of binary to run
#[arg(long, env = "NODE_STATUS_AGENT_PROBE_PATH")]
probe_path: String,
},
}
impl Args {
pub(crate) async fn execute(&self) -> anyhow::Result<()> {
match &self.command {
Command::RunProbe { probe_path } => self.run_probe(probe_path).await?,
}
Ok(())
}
async fn run_probe(&self, probe_path: &str) -> anyhow::Result<()> {
let server_address = format!("{}:{}", &self.server_address, self.server_port);
let probe = GwProbe::new(probe_path.to_string());
let version = probe.version().await;
tracing::info!("Probe version:\n{}", version);
if let Some(testrun) = request_testrun(&server_address).await? {
let log = probe.run_and_get_log(&Some(testrun.gateway_identity_key));
submit_results(&server_address, testrun.testrun_id, log).await?;
} else {
tracing::info!("No testruns available, exiting")
}
Ok(())
}
}
const URL_BASE: &str = "internal/testruns";
#[instrument(level = "debug", skip_all)]
async fn request_testrun(server_addr: &str) -> anyhow::Result<Option<TestrunAssignment>> {
let target_url = format!("{}/{}", server_addr, URL_BASE);
let client = reqwest::Client::new();
let res = client.get(target_url).send().await?;
let status = res.status();
let response_text = res.text().await?;
if status.is_client_error() {
bail!("{}: {}", status, response_text);
} else if status.is_server_error() {
if matches!(status, reqwest::StatusCode::SERVICE_UNAVAILABLE)
&& response_text.contains("No testruns available")
{
return Ok(None);
} else {
bail!("{}: {}", status, response_text);
}
}
serde_json::from_str(&response_text)
.map(|testrun| {
tracing::info!("Received testrun assignment: {:?}", testrun);
testrun
})
.map_err(|err| {
tracing::error!("err");
err.into()
})
}
#[instrument(level = "debug", skip(probe_outcome))]
async fn submit_results(
server_addr: &str,
testrun_id: i64,
probe_outcome: String,
) -> anyhow::Result<()> {
let target_url = format!("{}/{}/{}", server_addr, URL_BASE, testrun_id);
let client = reqwest::Client::new();
let res = client
.post(target_url)
.body(probe_outcome)
.send()
.await
.and_then(|response| response.error_for_status())?;
tracing::debug!("Submitted results: {})", res.status());
Ok(())
}
@@ -0,0 +1,55 @@
use std::{fs::File, io::Write, path::Path};
use tracing::info;
pub(crate) fn generate_key_pair(path: impl AsRef<Path>) -> anyhow::Result<()> {
let priv_key_path = path.as_ref();
let mut rng = rand::thread_rng();
let keypair = nym_crypto::asymmetric::identity::KeyPair::new(&mut rng);
info!("Generated keypair as Base58-encoded string");
let mut private_key_file = File::create(priv_key_path)?;
private_key_file.write_all(keypair.private_key().to_base58_string().as_bytes())?;
let pub_key_path = priv_key_path.with_extension("public");
let mut public_key_file = File::create(&pub_key_path)?;
public_key_file.write_all(keypair.public_key().to_base58_string().as_bytes())?;
info!(
"Saved Base58-encoded keypair, private key to {}, public key to {}",
priv_key_path.display(),
pub_key_path.display()
);
info!("Public key should be whitelisted with NS API");
Ok(())
}
#[cfg(test)]
mod test {
use nym_crypto::asymmetric::ed25519::PrivateKey;
use tempfile::TempDir;
use super::*;
use std::{fs, path::PathBuf};
#[test]
fn can_generate_valid_keypair() {
let tmp_dir = TempDir::new().unwrap();
let pkey_file = PathBuf::from_iter(&[
tmp_dir.path().to_path_buf(),
PathBuf::from("agent-key-private"),
]);
generate_key_pair(&pkey_file).expect("Failed to generate keypair");
let pkey_raw = fs::read_to_string(&pkey_file).expect("Failed to read file");
let key = PrivateKey::from_base58_string(pkey_raw).expect("Failed to load key");
let msg = "hello, world";
let signature = key.sign(msg);
key.public_key()
.verify(msg, &signature)
.expect("Failed to verify signature");
}
}
+74
View File
@@ -0,0 +1,74 @@
use crate::probe::GwProbe;
use clap::{Parser, Subcommand};
use nym_bin_common::bin_info;
use std::sync::OnceLock;
pub(crate) mod generate_keypair;
pub(crate) mod run_probe;
// Helper for passing LONG_VERSION to clap
fn pretty_build_info_static() -> &'static str {
static PRETTY_BUILD_INFORMATION: OnceLock<String> = OnceLock::new();
PRETTY_BUILD_INFORMATION.get_or_init(|| bin_info!().pretty_print())
}
#[derive(Parser, Debug)]
#[clap(author = "Nymtech", version, long_version = pretty_build_info_static(), about)]
pub(crate) struct Args {
#[command(subcommand)]
pub(crate) command: Command,
}
#[derive(Subcommand, Debug)]
pub(crate) enum Command {
RunProbe {
#[arg(short, long, env = "NODE_STATUS_AGENT_SERVER_ADDRESS")]
server_address: String,
#[arg(short = 'p', long, env = "NODE_STATUS_AGENT_SERVER_PORT")]
server_port: u16,
/// base58-encoded private key
#[arg(long, env = "NODE_STATUS_AGENT_AUTH_KEY")]
ns_api_auth_key: String,
/// path of binary to run
#[arg(long, env = "NODE_STATUS_AGENT_PROBE_PATH")]
probe_path: String,
},
GenerateKeypair {
#[arg(long)]
path: Option<String>,
},
}
impl Args {
pub(crate) async fn execute(&self) -> anyhow::Result<()> {
match &self.command {
Command::RunProbe {
server_address,
server_port,
ns_api_auth_key,
probe_path,
} => run_probe::run_probe(
server_address,
server_port.to_owned(),
ns_api_auth_key,
probe_path,
)
.await
.inspect_err(|err| {
tracing::error!("{err}");
})?,
Command::GenerateKeypair { path } => {
let path = path
.to_owned()
.unwrap_or_else(|| String::from("private-key"));
generate_keypair::generate_key_pair(path)?
}
}
Ok(())
}
}
+140
View File
@@ -0,0 +1,140 @@
use anyhow::{bail, Context};
use nym_common_models::ns_api::{get_testrun, submit_results, TestrunAssignment};
use nym_crypto::asymmetric::ed25519::{PrivateKey, Signature};
use std::fmt::Display;
use tracing::instrument;
use crate::cli::GwProbe;
const INTERNAL_TESTRUNS: &str = "internal/testruns";
pub(crate) async fn run_probe(
server_ip: &str,
server_port: u16,
ns_api_auth_key: &str,
probe_path: &str,
) -> anyhow::Result<()> {
let auth_key = PrivateKey::from_base58_string(ns_api_auth_key)
.context("Couldn't parse auth key, exiting")?;
let ns_api_client = Client::new(server_ip, server_port, auth_key);
let probe = GwProbe::new(probe_path.to_string());
let version = probe.version().await;
tracing::info!("Probe version:\n{}", version);
if let Some(testrun) = ns_api_client.request_testrun().await? {
let log = probe.run_and_get_log(&Some(testrun.gateway_identity_key));
ns_api_client
.submit_results(testrun.testrun_id, log, testrun.assigned_at_utc)
.await?;
} else {
tracing::info!("No testruns available, exiting")
}
Ok(())
}
struct Client {
server_address: String,
client: reqwest::Client,
auth_key: PrivateKey,
}
impl Client {
pub fn new(server_ip: &str, server_port: u16, auth_key: PrivateKey) -> Self {
let server_address = format!("{}:{}", server_ip, server_port);
let client = reqwest::Client::new();
Self {
server_address,
client,
auth_key,
}
}
#[instrument(level = "debug", skip_all)]
pub(crate) async fn request_testrun(&self) -> anyhow::Result<Option<TestrunAssignment>> {
let target_url = self.api_with_subpath(None::<String>);
let payload = get_testrun::Payload {
agent_public_key: self.auth_key.public_key(),
timestamp: chrono::offset::Utc::now().timestamp(),
};
let signature = self.sign_message(&payload)?;
let request = get_testrun::GetTestrunRequest { payload, signature };
let res = self.client.get(target_url).json(&request).send().await?;
let status = res.status();
let response_text = res.text().await?;
if status.is_client_error() {
bail!("{}: {}", status, response_text);
} else if status.is_server_error() {
if matches!(status, reqwest::StatusCode::SERVICE_UNAVAILABLE)
&& response_text.contains("No testruns available")
{
return Ok(None);
} else {
bail!("{}: {}", status, response_text);
}
}
serde_json::from_str(&response_text)
.map(|testrun| {
tracing::info!("Received testrun assignment: {:?}", testrun);
testrun
})
.map_err(|err| {
tracing::error!("err");
err.into()
})
}
#[instrument(level = "debug", skip(self, probe_result))]
pub(crate) async fn submit_results(
&self,
testrun_id: i64,
probe_result: String,
assigned_at_utc: i64,
) -> anyhow::Result<()> {
let target_url = self.api_with_subpath(Some(testrun_id));
let payload = submit_results::Payload {
probe_result,
agent_public_key: self.auth_key.public_key(),
assigned_at_utc,
};
let signature = self.sign_message(&payload)?;
let submit_results = submit_results::SubmitResults { payload, signature };
let res = self
.client
.post(target_url)
.json(&submit_results)
.send()
.await
.and_then(|response| response.error_for_status())?;
tracing::debug!("Submitted results: {})", res.status());
Ok(())
}
fn sign_message<T>(&self, message: &T) -> anyhow::Result<Signature>
where
T: serde::Serialize,
{
let serialized = bincode::serialize(message)?;
let signed = self.auth_key.sign(&serialized);
Ok(signed)
}
fn api_with_subpath(&self, subpath: Option<impl Display>) -> String {
if let Some(subpath) = subpath {
format!("{}/{}/{}", self.server_address, INTERNAL_TESTRUNS, subpath)
} else {
format!("{}/{}", self.server_address, INTERNAL_TESTRUNS)
}
}
}
-15
View File
@@ -11,26 +11,11 @@ async fn main() -> anyhow::Result<()> {
setup_tracing();
let args = Args::parse();
let server_addr = format!("{}:{}", args.server_address, args.server_port);
test_ns_api_conn(&server_addr).await?;
args.execute().await?;
Ok(())
}
async fn test_ns_api_conn(server_addr: &str) -> anyhow::Result<()> {
reqwest::get(server_addr)
.await
.map(|res| {
tracing::info!(
"Testing connection to NS API at {server_addr}: {}",
res.status()
);
})
.map_err(|err| anyhow::anyhow!("Couldn't connect to server on {}: {}", server_addr, err))
}
pub(crate) fn setup_tracing() {
fn directive_checked(directive: impl Into<String>) -> Directive {
directive
+3 -1
View File
@@ -3,7 +3,7 @@
[package]
name = "nym-node-status-api"
version = "0.1.6"
version = "1.0.0-rc.2"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
@@ -15,6 +15,7 @@ rust-version.workspace = true
[dependencies]
anyhow = { workspace = true }
axum = { workspace = true, features = ["tokio", "macros"] }
bincode = { workspace = true }
chrono = { workspace = true }
clap = { workspace = true, features = ["cargo", "derive", "env", "string"] }
cosmwasm-std = { workspace = true }
@@ -23,6 +24,7 @@ futures-util = { workspace = true }
moka = { workspace = true, features = ["future"] }
nym-bin-common = { path = "../common/bin-common", features = ["models"]}
nym-common-models = { path = "../common/models" }
nym-crypto = { path = "../common/crypto", features = ["asymmetric", "serde"] }
nym-explorer-client = { path = "../explorer-api/explorer-client" }
nym-network-defaults = { path = "../common/network-defaults" }
nym-validator-client = { path = "../common/client-libs/validator-client" }
+14 -3
View File
@@ -6,14 +6,25 @@ export RUST_LOG=${RUST_LOG:-debug}
export NYM_API_CLIENT_TIMEOUT=60
export EXPLORER_CLIENT_TIMEOUT=60
export NODE_STATUS_API_TESTRUN_REFRESH_INTERVAL=60
export NODE_STATUS_API_TESTRUN_REFRESH_INTERVAL=120
# public keys corresponding to the agents NS API is expecting to be contacted from
export NODE_STATUS_API_AGENT_KEY_LIST="H4z8kx5Kkf5JMQHhxaW1MwYndjKCDHC7HsVhHTFfBZ4J,
5c2GW61135DEr73DxGrR4DR22BLEujvm1k8GYEjRB9at,
3PSFDH2iSJ61KoDNyJpAiw42xS5smV5iBXWnRGTmk2du,
2AH7pJL5PErbSFhZdu3uH8cKa1h1tyCUfSRUm6E5EBz8,
6wQ9ifPFm2EB73BrwpGSd3Ek7GFA5kiAMQDP2ox6JKZw,
G1tevJBnzaQ6zCUsFsxtGJf45BqCTDgzpEz6Sgxks8EH,
FwjL2nGrtgQQ48fPqAUzUZ8UkQZtMtgehqTqj4PQopvh,
Eujj4GmvwQBgHZaNSyqUbjMFSsnXWPSjEYUPgAsKmx1A,
5ZnfSGxW6EKcFxB8jftb9V3f897VpwpZtf7kCPYzB595,
H9kuRd8BGjEUD8Grh5U9YUPN5ZaQmSYz8U44R72AffKM"
export ENVIRONMENT="qa.env"
export ENVIRONMENT=${ENVIRONMENT:-"sandbox"}
function run_bare() {
# export necessary env vars
set -a
source ../envs/$ENVIRONMENT
source ../envs/${ENVIRONMENT}.env
set +a
export RUST_LOG=debug
@@ -0,0 +1,5 @@
ALTER TABLE testruns
RENAME COLUMN timestamp_utc TO created_utc;
ALTER TABLE testruns
ADD COLUMN last_assigned_utc INTEGER;
+13 -2
View File
@@ -56,7 +56,7 @@ pub(crate) struct Cli {
#[clap(
long,
default_value = "600",
default_value = "300",
env = "NODE_STATUS_API_MONITOR_REFRESH_INTERVAL"
)]
#[arg(value_parser = parse_duration)]
@@ -64,11 +64,22 @@ pub(crate) struct Cli {
#[clap(
long,
default_value = "600",
default_value = "300",
env = "NODE_STATUS_API_TESTRUN_REFRESH_INTERVAL"
)]
#[arg(value_parser = parse_duration)]
pub(crate) testruns_refresh_interval: Duration,
#[clap(env = "NODE_STATUS_API_AGENT_KEY_LIST")]
#[arg(value_delimiter = ',')]
pub(crate) agent_key_list: Vec<String>,
#[clap(
long,
default_value_t = 40,
env = "NYM_NODE_STATUS_API_MAX_AGENT_COUNT"
)]
pub(crate) max_agent_count: i64,
}
fn parse_duration(arg: &str) -> Result<std::time::Duration, std::num::ParseIntError> {
+2 -1
View File
@@ -306,9 +306,10 @@ pub struct TestRunDto {
pub id: i64,
pub gateway_id: i64,
pub status: i64,
pub timestamp_utc: i64,
pub created_utc: i64,
pub ip_address: String,
pub log: String,
pub last_assigned_utc: Option<i64>,
}
#[derive(Debug, Clone, strum_macros::Display, EnumString, FromRepr, PartialEq)]
+39 -13
View File
@@ -4,10 +4,26 @@ use crate::{
db::models::{TestRunDto, TestRunStatus},
testruns::now_utc,
};
use anyhow::Context;
use chrono::Duration;
use sqlx::{pool::PoolConnection, Sqlite};
pub(crate) async fn count_testruns_in_progress(
conn: &mut PoolConnection<Sqlite>,
) -> anyhow::Result<i64> {
sqlx::query_scalar!(
r#"SELECT
COUNT(id) as "count: i64"
FROM testruns
WHERE
status = ?
"#,
TestRunStatus::InProgress as i64,
)
.fetch_one(conn.as_mut())
.await
.map_err(anyhow::Error::from)
}
pub(crate) async fn get_in_progress_testrun_by_id(
conn: &mut PoolConnection<Sqlite>,
testrun_id: i64,
@@ -18,26 +34,31 @@ pub(crate) async fn get_in_progress_testrun_by_id(
id as "id!",
gateway_id as "gateway_id!",
status as "status!",
timestamp_utc as "timestamp_utc!",
created_utc as "created_utc!",
ip_address as "ip_address!",
log as "log!"
log as "log!",
last_assigned_utc
FROM testruns
WHERE
id = ?
AND
status = ?
ORDER BY timestamp_utc"#,
ORDER BY created_utc
LIMIT 1"#,
testrun_id,
TestRunStatus::InProgress as i64,
)
.fetch_one(conn.as_mut())
.await
.context(format!("Couldn't retrieve testrun {testrun_id}"))
.map_err(|e| anyhow::anyhow!("Couldn't retrieve testrun {testrun_id}: {e}"))
}
pub(crate) async fn update_testruns_older_than(db: &DbPool, age: Duration) -> anyhow::Result<u64> {
pub(crate) async fn update_testruns_assigned_before(
db: &DbPool,
max_age: Duration,
) -> anyhow::Result<u64> {
let mut conn = db.acquire().await?;
let previous_run = now_utc() - age;
let previous_run = now_utc() - max_age;
let cutoff_timestamp = previous_run.timestamp();
let res = sqlx::query!(
@@ -48,7 +69,7 @@ pub(crate) async fn update_testruns_older_than(db: &DbPool, age: Duration) -> an
WHERE
status = ?
AND
timestamp_utc < ?
last_assigned_utc < ?
"#,
TestRunStatus::Queued as i64,
TestRunStatus::InProgress as i64,
@@ -59,8 +80,8 @@ pub(crate) async fn update_testruns_older_than(db: &DbPool, age: Duration) -> an
let stale_testruns = res.rows_affected();
if stale_testruns > 0 {
tracing::debug!(
"Refreshed {} stale testruns, scheduled before {} but not yet finished",
tracing::info!(
"Refreshed {} stale testruns, assigned before {} but not yet finished",
stale_testruns,
previous_run
);
@@ -69,19 +90,22 @@ pub(crate) async fn update_testruns_older_than(db: &DbPool, age: Duration) -> an
Ok(stale_testruns)
}
pub(crate) async fn get_oldest_testrun_and_make_it_pending(
pub(crate) async fn assign_oldest_testrun(
conn: &mut PoolConnection<Sqlite>,
) -> anyhow::Result<Option<TestrunAssignment>> {
let now = now_utc().timestamp();
// find & mark as "In progress" in the same transaction to avoid race conditions
let returning = sqlx::query!(
r#"UPDATE testruns
SET status = ?
SET
status = ?,
last_assigned_utc = ?
WHERE rowid =
(
SELECT rowid
FROM testruns
WHERE status = ?
ORDER BY timestamp_utc asc
ORDER BY created_utc asc
LIMIT 1
)
RETURNING
@@ -89,6 +113,7 @@ pub(crate) async fn get_oldest_testrun_and_make_it_pending(
gateway_id
"#,
TestRunStatus::InProgress as i64,
now,
TestRunStatus::Queued as i64,
)
.fetch_optional(conn.as_mut())
@@ -111,6 +136,7 @@ pub(crate) async fn get_oldest_testrun_and_make_it_pending(
Ok(Some(TestrunAssignment {
testrun_id: testrun.id,
gateway_identity_key: gw_identity.gateway_identity_key,
assigned_at_utc: now,
}))
} else {
Ok(None)
+117 -35
View File
@@ -4,10 +4,12 @@ use axum::{
extract::{Path, State},
Router,
};
use nym_common_models::ns_api::{get_testrun, submit_results, VerifiableRequest};
use reqwest::StatusCode;
use crate::db::models::TestRunStatus;
use crate::db::queries;
use crate::testruns::now_utc;
use crate::{
db,
http::{
@@ -28,9 +30,14 @@ pub(crate) fn routes() -> Router<AppState> {
}
#[tracing::instrument(level = "debug", skip_all)]
async fn request_testrun(State(state): State<AppState>) -> HttpResult<Json<TestrunAssignment>> {
// TODO dz log agent's key
async fn request_testrun(
State(state): State<AppState>,
Json(request): Json<get_testrun::GetTestrunRequest>,
) -> HttpResult<Json<TestrunAssignment>> {
// TODO dz log agent's network probe version
authenticate(&request, &state)?;
is_fresh(&request.payload.timestamp)?;
tracing::debug!("Agent requested testrun");
let db = state.db_pool();
@@ -39,17 +46,29 @@ async fn request_testrun(State(state): State<AppState>) -> HttpResult<Json<Testr
.await
.map_err(HttpError::internal_with_logging)?;
return match db::queries::testruns::get_oldest_testrun_and_make_it_pending(&mut conn).await {
let active_testruns = db::queries::testruns::count_testruns_in_progress(&mut conn)
.await
.map_err(HttpError::internal_with_logging)?;
if active_testruns >= state.agent_max_count() {
tracing::warn!(
"{}/{} testruns in progress, rejecting",
active_testruns,
state.agent_max_count()
);
return Err(HttpError::no_testruns_available());
}
return match db::queries::testruns::assign_oldest_testrun(&mut conn).await {
Ok(res) => {
if let Some(testrun) = res {
tracing::debug!(
tracing::info!(
"🏃‍ Assigned testrun row_id {} gateway {} to agent",
&testrun.testrun_id,
testrun.gateway_identity_key
testrun.gateway_identity_key,
);
Ok(Json(testrun))
} else {
tracing::debug!("No testruns available for agent");
tracing::debug!("No testruns available");
Err(HttpError::no_testruns_available())
}
}
@@ -57,64 +76,127 @@ async fn request_testrun(State(state): State<AppState>) -> HttpResult<Json<Testr
};
}
// TODO dz accept testrun_id as query parameter
#[tracing::instrument(level = "debug", skip_all)]
async fn submit_testrun(
Path(testrun_id): Path<i64>,
Path(submitted_testrun_id): Path<i64>,
State(state): State<AppState>,
body: String,
Json(submitted_result): Json<submit_results::SubmitResults>,
) -> HttpResult<StatusCode> {
authenticate(&submitted_result, &state)?;
let db = state.db_pool();
let mut conn = db
.acquire()
.await
.map_err(HttpError::internal_with_logging)?;
let testrun = queries::testruns::get_in_progress_testrun_by_id(&mut conn, testrun_id)
.await
.map_err(|e| {
tracing::error!("{e}");
HttpError::not_found(testrun_id)
})?;
let assigned_testrun =
queries::testruns::get_in_progress_testrun_by_id(&mut conn, submitted_testrun_id)
.await
.map_err(|err| {
tracing::warn!(
"No testruns in progress for testrun_id {}: {}",
submitted_testrun_id,
err
);
HttpError::invalid_input("Invalid testrun submitted")
})?;
if Some(submitted_result.payload.assigned_at_utc) != assigned_testrun.last_assigned_utc {
tracing::warn!(
"Submitted testrun timestamp mismatch: {} != {:?}, rejecting",
submitted_result.payload.assigned_at_utc,
assigned_testrun.last_assigned_utc
);
return Err(HttpError::invalid_input("Invalid testrun submitted"));
}
let gw_identity = db::queries::select_gateway_identity(&mut conn, testrun.gateway_id)
let gw_identity = db::queries::select_gateway_identity(&mut conn, assigned_testrun.gateway_id)
.await
.map_err(|_| {
// should never happen:
HttpError::internal_with_logging("No gateway found for testrun")
HttpError::internal_with_logging(format!(
"No gateway found for testrun {submitted_testrun_id}"
))
})?;
tracing::debug!(
"Agent submitted testrun {} for gateway {} ({} bytes)",
testrun_id,
submitted_testrun_id,
gw_identity,
body.len(),
&submitted_result.payload.probe_result.len(),
);
// TODO dz this should be part of a single transaction: commit after everything is done
queries::testruns::update_testrun_status(&mut conn, testrun_id, TestRunStatus::Complete)
queries::testruns::update_testrun_status(
&mut conn,
submitted_testrun_id,
TestRunStatus::Complete,
)
.await
.map_err(HttpError::internal_with_logging)?;
queries::testruns::update_gateway_last_probe_log(
&mut conn,
assigned_testrun.gateway_id,
&submitted_result.payload.probe_result,
)
.await
.map_err(HttpError::internal_with_logging)?;
let result = get_result_from_log(&submitted_result.payload.probe_result);
queries::testruns::update_gateway_last_probe_result(
&mut conn,
assigned_testrun.gateway_id,
&result,
)
.await
.map_err(HttpError::internal_with_logging)?;
queries::testruns::update_gateway_score(&mut conn, assigned_testrun.gateway_id)
.await
.map_err(HttpError::internal_with_logging)?;
queries::testruns::update_gateway_last_probe_log(&mut conn, testrun.gateway_id, &body)
.await
.map_err(HttpError::internal_with_logging)?;
let result = get_result_from_log(&body);
queries::testruns::update_gateway_last_probe_result(&mut conn, testrun.gateway_id, &result)
.await
.map_err(HttpError::internal_with_logging)?;
queries::testruns::update_gateway_score(&mut conn, testrun.gateway_id)
.await
.map_err(HttpError::internal_with_logging)?;
// TODO dz log gw identity key
let created_at = chrono::DateTime::from_timestamp(assigned_testrun.created_utc, 0)
.map(|d| d.to_rfc3339())
.unwrap_or_default();
let last_assigned = assigned_testrun
.last_assigned_utc
.and_then(|d| chrono::DateTime::from_timestamp(d, 0))
.map(|d| d.to_rfc3339())
.unwrap_or_default();
tracing::info!(
"✅ Testrun row_id {} for gateway {} complete",
testrun.id,
gw_identity
"✅ Testrun row_id {} for gateway {} complete (last assigned at {}, created at {})",
assigned_testrun.id,
gw_identity,
last_assigned,
created_at
);
Ok(StatusCode::CREATED)
}
// TODO dz this should be middleware
#[tracing::instrument(level = "debug", skip_all)]
fn authenticate(request: &impl VerifiableRequest, state: &AppState) -> HttpResult<()> {
if !state.is_registered(request.public_key()) {
tracing::warn!("Public key not registered with NS API, rejecting");
return Err(HttpError::unauthorized());
};
request.verify_signature().map_err(|_| {
tracing::warn!("Signature verification failed, rejecting");
HttpError::unauthorized()
})?;
Ok(())
}
fn is_fresh(request_time: &i64) -> HttpResult<()> {
// if a request took longer than N minutes to reach NS API, something is very wrong
let freshness_cutoff = chrono::Duration::minutes(1);
let cutoff_timestamp = (now_utc() - freshness_cutoff).timestamp();
if *request_time < cutoff_timestamp {
tracing::warn!("Request older than {}s, rejecting", cutoff_timestamp);
return Err(HttpError::unauthorized());
}
Ok(())
}
fn get_result_from_log(log: &str) -> String {
let re = regex::Regex::new(r"\n\{\s").unwrap();
let result: Vec<_> = re.splitn(log, 2).collect();
+7 -6
View File
@@ -15,6 +15,13 @@ impl HttpError {
}
}
pub(crate) fn unauthorized() -> Self {
Self {
message: serde_json::json!({"message": "Make sure your public key is registered with NS API"}).to_string(),
status: axum::http::StatusCode::UNAUTHORIZED,
}
}
pub(crate) fn internal_with_logging(msg: impl Display) -> Self {
tracing::error!("{}", msg.to_string());
Self::internal()
@@ -33,12 +40,6 @@ impl HttpError {
status: axum::http::StatusCode::SERVICE_UNAVAILABLE,
}
}
pub(crate) fn not_found(msg: impl Display) -> Self {
Self {
message: serde_json::json!({"message": msg.to_string()}).to_string(),
status: axum::http::StatusCode::NOT_FOUND,
}
}
}
impl axum::response::IntoResponse for HttpError {
+4 -1
View File
@@ -1,5 +1,6 @@
use axum::Router;
use core::net::SocketAddr;
use nym_crypto::asymmetric::ed25519::PublicKey;
use tokio::{net::TcpListener, task::JoinHandle};
use tokio_util::sync::{CancellationToken, WaitForCancellationFutureOwned};
@@ -14,10 +15,12 @@ pub(crate) async fn start_http_api(
db_pool: DbPool,
http_port: u16,
nym_http_cache_ttl: u64,
agent_key_list: Vec<PublicKey>,
agent_max_count: i64,
) -> anyhow::Result<ShutdownHandles> {
let router_builder = RouterBuilder::with_default_routes();
let state = AppState::new(db_pool, nym_http_cache_ttl);
let state = AppState::new(db_pool, nym_http_cache_ttl, agent_key_list, agent_max_count);
let router = router_builder.with_state(state);
// TODO dz do we need this to be configurable?
+19 -1
View File
@@ -1,6 +1,7 @@
use std::{sync::Arc, time::Duration};
use moka::{future::Cache, Entry};
use nym_crypto::asymmetric::ed25519::PublicKey;
use tokio::sync::RwLock;
use crate::{
@@ -12,13 +13,22 @@ use crate::{
pub(crate) struct AppState {
db_pool: DbPool,
cache: HttpCache,
agent_key_list: Vec<PublicKey>,
agent_max_count: i64,
}
impl AppState {
pub(crate) fn new(db_pool: DbPool, cache_ttl: u64) -> Self {
pub(crate) fn new(
db_pool: DbPool,
cache_ttl: u64,
agent_key_list: Vec<PublicKey>,
agent_max_count: i64,
) -> Self {
Self {
db_pool,
cache: HttpCache::new(cache_ttl),
agent_key_list,
agent_max_count,
}
}
@@ -29,6 +39,14 @@ impl AppState {
pub(crate) fn cache(&self) -> &HttpCache {
&self.cache
}
pub(crate) fn is_registered(&self, agent_pubkey: &PublicKey) -> bool {
self.agent_key_list.contains(agent_pubkey)
}
pub(crate) fn agent_max_count(&self) -> i64 {
self.agent_max_count
}
}
static GATEWAYS_LIST_KEY: &str = "gateways";
+11
View File
@@ -1,4 +1,5 @@
use clap::Parser;
use nym_crypto::asymmetric::ed25519::PublicKey;
use nym_task::signal::wait_for_signal;
mod cli;
@@ -14,6 +15,13 @@ async fn main() -> anyhow::Result<()> {
let args = cli::Cli::parse();
let agent_key_list = args
.agent_key_list
.iter()
.map(|value| PublicKey::from_base58_string(value.trim()).map_err(anyhow::Error::from))
.collect::<anyhow::Result<Vec<_>>>()?;
tracing::info!("Registered {} agent keys", agent_key_list.len());
let connection_url = args.database_url.clone();
tracing::debug!("Using config:\n{:#?}", args);
@@ -31,12 +39,15 @@ async fn main() -> anyhow::Result<()> {
.await;
tracing::info!("Started monitor task");
});
testruns::spawn(storage.pool_owned(), args.testruns_refresh_interval).await;
let shutdown_handles = http::server::start_http_api(
storage.pool_owned(),
args.http_port,
args.nym_http_cache_ttl,
agent_key_list.to_owned(),
args.max_agent_count,
)
.await
.expect("Failed to start server");
+4 -5
View File
@@ -1,4 +1,5 @@
use crate::db::models::GatewayIdentityDto;
use crate::db::DbPool;
use futures_util::TryStreamExt;
use std::time::Duration;
@@ -24,8 +25,6 @@ pub(crate) async fn spawn(pool: DbPool, refresh_interval: Duration) {
});
}
// TODO dz make number of max agents configurable
#[instrument(level = "debug", name = "testrun_queue", skip_all)]
async fn run(pool: &DbPool) -> anyhow::Result<()> {
tracing::info!("Spawning testruns...");
@@ -72,15 +71,15 @@ async fn run(pool: &DbPool) -> anyhow::Result<()> {
testruns_created += 1;
}
}
tracing::debug!("{} testruns queued in total", testruns_created);
tracing::info!("{} testruns queued in total", testruns_created);
Ok(())
}
#[instrument(level = "debug", skip_all)]
async fn refresh_stale_testruns(pool: &DbPool, refresh_interval: Duration) -> anyhow::Result<()> {
let chrono_duration = chrono::Duration::from_std(refresh_interval)?;
crate::db::queries::testruns::update_testruns_older_than(pool, chrono_duration).await?;
let refresh_interval = chrono::Duration::from_std(refresh_interval)?;
crate::db::queries::testruns::update_testruns_assigned_before(pool, refresh_interval).await?;
Ok(())
}
+5 -4
View File
@@ -28,7 +28,7 @@ pub(crate) async fn try_queue_testrun(
LIMIT 1"#,
identity_key,
)
// TODO dz shoudl call .fetch_one
// TODO dz should call .fetch_one
// TODO dz replace this in other queries as well
.fetch(conn.as_mut())
.try_collect::<Vec<_>>()
@@ -53,9 +53,10 @@ pub(crate) async fn try_queue_testrun(
id as "id!",
gateway_id as "gateway_id!",
status as "status!",
timestamp_utc as "timestamp_utc!",
created_utc as "created_utc!",
ip_address as "ip_address!",
log as "log!"
log as "log!",
last_assigned_utc
FROM testruns
WHERE gateway_id = ? AND status != 2
ORDER BY id DESC
@@ -89,7 +90,7 @@ pub(crate) async fn try_queue_testrun(
);
let id = sqlx::query!(
"INSERT INTO testruns (gateway_id, status, ip_address, timestamp_utc, log) VALUES (?, ?, ?, ?, ?)",
"INSERT INTO testruns (gateway_id, status, ip_address, created_utc, log) VALUES (?, ?, ?, ?, ?)",
gateway_id,
status,
ip_address,
+1 -1
View File
@@ -3,7 +3,7 @@
[package]
name = "nym-node"
version = "1.1.10"
version = "1.1.11"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
@@ -59,6 +59,17 @@ pub struct HostInformation {
pub keys: HostKeys,
}
impl HostInformation {
pub fn check_ips(&self) -> bool {
for ip in &self.ip_address {
if ip.is_unspecified() || ip.is_loopback() || ip.is_multicast() {
return false;
}
}
true
}
}
#[derive(Serialize)]
pub struct LegacyHostInformationV2 {
pub ip_address: Vec<IpAddr>,
+319
View File
@@ -0,0 +1,319 @@
#!/bin/bash
network_device=$(ip route show default | awk '/default/ {print $5}')
tunnel_interface="nymtun0"
wg_tunnel_interface="nymwg"
if ! dpkg -s iptables-persistent >/dev/null 2>&1; then
sudo apt-get update
sudo apt-get install -y iptables-persistent
else
echo "iptables-persistent is already installed."
fi
fetch_ipv6_address_nym_tun() {
ipv6_global_address=$(ip -6 addr show $tunnel_interface scope global | grep inet6 | awk '{print $2}' | head -n 1)
if [[ -z "$ipv6_global_address" ]]; then
echo "no globally routable IPv6 address found on $tunnel_interface. please configure IPv6 or check your network settings."
exit 1
else
echo "using IPv6 address: $ipv6_global_address"
fi
}
fetch_and_display_ipv6() {
ipv6_address=$(ip -6 addr show ${network_device} scope global | grep inet6 | awk '{print $2}')
if [[ -z "$ipv6_address" ]]; then
echo "no global IPv6 address found on ${network_device}."
else
echo "IPv6 address on ${network_device}: $ipv6_address"
fi
}
adjust_ip_forwarding() {
ipv6_forwarding_setting="net.ipv6.conf.all.forwarding=1"
ipv4_forwarding_setting="net.ipv4.ip_forward=1"
echo "$ipv6_forwarding_setting" | sudo tee -a /etc/sysctl.conf
echo "$ipv4_forwarding_setting" | sudo tee -a /etc/sysctl.conf
sysctl -p /etc/sysctl.conf
}
apply_iptables_rules_wg() {
echo "applying IPtables rules..."
echo "network device: ${network_device}"
echo "tunnel_interface: ${wg_tunnel_interface}"
sleep 2
sudo iptables -A FORWARD -i "$wg_tunnel_interface" -o "$network_device" -j ACCEPT
sudo iptables -A FORWARD -i "$network_device" -o "$wg_tunnel_interface" -m state --state RELATED,ESTABLISHED -j ACCEPT
sudo ip6tables -A FORWARD -i "$network_device" -o "$wg_tunnel_interface" -m state --state RELATED,ESTABLISHED -j ACCEPT
sudo ip6tables -A FORWARD -i "$wg_tunnel_interface" -o "$network_device" -j ACCEPT
sudo iptables-save | sudo tee /etc/iptables/rules.v4
sudo ip6tables-save | sudo tee /etc/iptables/rules.v6
}
remove_iptables_rules_wg() {
echo "removing IPtables rules..."
echo "network device: ${network_device}"
echo "tunnel_interface: ${wg_tunnel_interface}"
sleep 2
# IPv4 rules removal wg
sudo iptables -D FORWARD -i "$wg_tunnel_interface" -o "$network_device" -j ACCEPT
sudo iptables -D FORWARD -i "$network_device" -o "$wg_tunnel_interface" -m state --state RELATED,ESTABLISHED -j ACCEPT
# IPv6 rules removal wg
sudo ip6tables -t nat -D POSTROUTING -o "$network_device" -j MASQUERADE
sudo ip6tables -D FORWARD -i "$network_device" -o "$wg_tunnel_interface" -m state --state RELATED,ESTABLISHED -j ACCEPT
sudo ip6tables -D FORWARD -i "$wg_tunnel_interface" -o "$network_device" -j ACCEPT
sudo iptables-save | sudo tee /etc/iptables/rules.v4
sudo ip6tables-save | sudo tee /etc/iptables/rules.v6
}
apply_iptables_rules() {
echo "applying IPtables rules..."
echo "network device: ${network_device}"
echo "tunnel_interface: ${tunnel_interface}"
sleep 2
sudo iptables -t nat -A POSTROUTING -o "$network_device" -j MASQUERADE
sudo iptables -A FORWARD -i "$tunnel_interface" -o "$network_device" -j ACCEPT
sudo iptables -A FORWARD -i "$network_device" -o "$tunnel_interface" -m state --state RELATED,ESTABLISHED -j ACCEPT
sudo ip6tables -t nat -A POSTROUTING -o "$network_device" -j MASQUERADE
sudo ip6tables -A FORWARD -i "$network_device" -o "$tunnel_interface" -m state --state RELATED,ESTABLISHED -j ACCEPT
sudo ip6tables -A FORWARD -i "$tunnel_interface" -o "$network_device" -j ACCEPT
adjust_ip_forwarding
sudo iptables-save | sudo tee /etc/iptables/rules.v4
sudo ip6tables-save | sudo tee /etc/iptables/rules.v6
}
remove_iptables_rules() {
echo "removing IPtables rules..."
echo "network device: ${network_device}"
echo "tunnel_interface: ${tunnel_interface}"
sleep 2
# IPv4 rules removal
sudo iptables -t nat -D POSTROUTING -o "$network_device" -j MASQUERADE
sudo iptables -D FORWARD -i "$tunnel_interface" -o "$network_device" -j ACCEPT
sudo iptables -D FORWARD -i "$network_device" -o "$tunnel_interface" -m state --state RELATED,ESTABLISHED -j ACCEPT
# IPv6 rules removal
sudo ip6tables -t nat -D POSTROUTING -o "$network_device" -j MASQUERADE
sudo ip6tables -D FORWARD -i "$network_device" -o "$tunnel_interface" -m state --state RELATED,ESTABLISHED -j ACCEPT
sudo ip6tables -D FORWARD -i "$tunnel_interface" -o "$network_device" -j ACCEPT
sudo iptables-save | sudo tee /etc/iptables/rules.v4
sudo ip6tables-save | sudo tee /etc/iptables/rules.v6
}
check_ipv6_ipv4_forwarding() {
result_ipv4=$(cat /proc/sys/net/ipv4/ip_forward)
result_ipv6=$(cat /proc/sys/net/ipv6/conf/all/forwarding)
echo "IPv4 forwarding is $([ "$result_ipv4" == "1" ] && echo "enabled" || echo "not enabled")."
echo "IPv6 forwarding is $([ "$result_ipv6" == "1" ] && echo "enabled" || echo "not enabled")."
}
check_nymtun_iptables() {
echo "network Device: $network_device"
echo "---------------------------------------"
echo
echo "inspecting IPv4 firewall rules..."
iptables -L FORWARD -v -n | awk -v dev="$network_device" '/^Chain FORWARD/ || /nymtun0/ && dev || dev && /nymtun0/ || /ufw-reject-forward/'
echo "---------------------------------------"
echo
echo "inspecting IPv6 firewall rules..."
ip6tables -L FORWARD -v -n | awk -v dev="$network_device" '/^Chain FORWARD/ || /nymtun0/ && dev || dev && /nymtun0/ || /ufw6-reject-forward/'
}
joke_through_the_mixnet() {
echo "checking Internet and mixnet connectivity (IPv4) via nymtun0..."
ipv4_address=$(ip addr show nymtun0 | grep 'inet ' | awk '{print $2}' | cut -d'/' -f1)
if [ -z "$ipv4_address" ]; then
echo
echo "no IPv4 address found on nymtun0."
echo "please ensure IPv4 is configured correctly on your device."
echo "unfortunately, there's no joke for you :( and you might not be able to route IPv4 traffic through your gateway to the internet."
else
joke=$(curl -s -H "Accept: application/json" --interface "$ipv4_address" https://icanhazdadjoke.com/ | jq -c .joke)
if [ -z "$joke" ] || [ "$joke" = "null" ]; then
echo "failed to fetch a joke. there might be an issue with the Internet connectivity or the joke service."
else
echo "joke fetched successfully:"
echo "$joke"
fi
fi
echo "checking Internet and mixnet connectivity (IPv6) via nymtun0..."
ipv6_address=$(ip addr show nymtun0 | grep 'inet6 ' | awk '{print $2}' | cut -d'/' -f1 | grep -v '^fe80:')
if [ -z "$ipv6_address" ]; then
echo
echo "no globally routable IPv6 address found on nymtun0."
echo "please ensure IPv6 is enabled on your VPS or configure your security groups/firewall settings appropriately."
echo "unfortunately there's no joke fo you :( and you can't route ipv6 traffic through your gateway to the internet"
else
joke=$(curl -s -H "Accept: application/json" --interface "$ipv6_address" https://icanhazdadjoke.com/ | jq -c .joke)
if [ -z "$joke" ] || [ "$joke" = "null" ]; then
echo "failed to fetch a joke. there might be an issue with the Internet connectivity or the joke service."
else
echo "joke fetched successfully:"
echo "$joke"
fi
fi
}
check_ip6_ipv4_routing() {
echo "---------------------------------------"
echo "examining IPv4 routing table..."
ip route
echo "---------------------------------------"
echo
echo "examining IPv6 routing table..."
ip -6 route
echo
}
perform_ipv4_ipv6_pings() {
echo "---------------------------------------"
echo "checking IPv4 connectivity (example: google.com)..."
ping -c 4 google.com
echo "---------------------------------------"
echo
echo "checking IPv6 connectivity (example: google.com)..."
ping6 -c 4 google.com
echo
}
configure_dns_and_icmp_wg() {
echo "allowing icmp (ping)..."
sudo iptables -A INPUT -p icmp --icmp-type echo-request -j ACCEPT
sudo iptables -A OUTPUT -p icmp --icmp-type echo-reply -j ACCEPT
echo "allowing dns over udp (port 53)..."
sudo iptables -A INPUT -p udp --dport 53 -j ACCEPT
echo "allowing dns over tcp (port 53)..."
sudo iptables -A INPUT -p tcp --dport 53 -j ACCEPT
echo "saving iptables rules..."
sudo iptables-save > /etc/iptables/rules.v4
echo "dns and icmp configuration completed."
}
joke_through_wg_tunnel() {
echo "checking nymwg tunnel status..."
tunnel_status=$(ip link show nymwg | grep -o "state [A-Z]*")
if [[ $tunnel_status == "state UNKNOWN" ]]; then
echo "nymwg tunnel is up."
else
echo "nymwg tunnel is down."
echo "please check your nymwg tunnel configuration."
return
fi
echo "checking internet and mixnet connectivity (ipv4) via nymwg..."
ipv4_address=$(ip addr show nymwg | grep 'inet ' | awk '{print $2}' | cut -d'/' -f1)
if [ -z "$ipv4_address" ]; then
echo
echo "no ipv4 address found on nymwg."
echo "please ensure ipv4 is configured correctly on your device."
echo "unfortunately, there's no joke for you :( and you might not be able to route ipv4 traffic through your gateway to the internet."
else
joke=$(curl -s -H "accept: application/json" --interface "$ipv4_address" https://icanhazdadjoke.com/ | jq -c .joke)
if [ -z "$joke" ] || [ "$joke" = "null" ]; then
echo "failed to fetch a joke. there might be an issue with the internet connectivity or the joke service."
else
echo "joke fetched successfully:"
echo "$joke"
fi
fi
echo "checking internet and mixnet connectivity (ipv6) via nymwg..."
ipv6_address=$(ip addr show nymwg | grep 'inet6 ' | awk '{print $2}' | cut -d'/' -f1 | grep -v '^fe80:')
if [ -z "$ipv6_address" ]; then
echo
echo "no globally routable ipv6 address found on nymwg."
echo "please ensure ipv6 is enabled on your vps or configure your security groups/firewall settings appropriately."
echo "unfortunately, there's no joke for you :( and you can't route ipv6 traffic through your gateway to the internet."
else
joke=$(curl -s -H "accept: application/json" --interface "$ipv6_address" https://icanhazdadjoke.com/ | jq -c .joke)
if [ -z "$joke" ] || [ "$joke" = "null" ]; then
echo "failed to fetch a joke. there might be an issue with the internet connectivity or the joke service."
else
echo "joke fetched successfully:"
echo "$joke"
fi
fi
}
case "$1" in
fetch_ipv6_address_nym_tun)
fetch_ipv6_address_nym_tun
;;
fetch_and_display_ipv6)
fetch_and_display_ipv6
;;
check_nymtun_iptables)
check_nymtun_iptables
;;
apply_iptables_rules)
apply_iptables_rules
;;
remove_iptables_rules)
remove_iptables_rules
;;
check_ipv6_ipv4_forwarding)
check_ipv6_ipv4_forwarding
;;
check_ip6_ipv4_routing)
check_ip6_ipv4_routing
;;
perform_ipv4_ipv6_pings)
perform_ipv4_ipv6_pings
;;
joke_through_the_mixnet)
joke_through_the_mixnet
;;
apply_iptables_rules_wg)
apply_iptables_rules_wg
;;
joke_through_wg_tunnel)
joke_through_wg_tunnel
;;
configure_dns_and_icmp_wg)
configure_dns_and_icmp_wg
;;
*)
echo "usage: $0 [command]"
echo "commands:"
echo " fetch_ipv6_address_nym_tun - Fetches the IPv6 address assigned to the '$tunnel_interface'."
echo " fetch_and_display_ipv6 - Displays the IPv6 address on the default network device."
echo " apply_iptables_rules - Applies necessary IPv4 and IPv6 iptables rules."
echo " apply_iptables_rules_wg - Applies iptable rules for IPv4 and IPv6 for Wireguard."
echo " remove_iptables_rules - Removes applied IPv4 and IPv6 iptables rules."
echo " remove_iptables_rules_wg - Removes applied IPv4 and IPv6 iptables rules for Wireguard."
echo " check_ipv6_ipv4_forwarding - Checks if IPv4 and IPv6 forwarding are enabled."
echo " check_nymtun_iptables - Check nymtun0 device."
echo " perform_ipv4_ipv6_pings - Perform IPv4 and IPv6 pings to google."
echo " check_ip6_ipv4_routing - Check IPv6 and IPv4 routing."
echo " joke_through_the_mixnet - Run a joke through the mixnet via IPv4 and IPv6."
echo " joke_through_wg_tunnel - Run a wg test, and get a joke through the wg tunnel."
echo " configure_dns_and_icmp_wg - Allows icmp ping tests for probes alongside configuring dns"
echo "please provide one of the above commands."
exit 1
;;
esac
echo "operation $1 completed successfully."
@@ -4,7 +4,7 @@
[package]
name = "nym-network-requester"
license = "GPL-3.0"
version = "1.1.44"
version = "1.1.45"
authors.workspace = true
edition.workspace = true
rust-version = "1.70"
+1
View File
@@ -231,6 +231,7 @@ impl NymNodeTesterBuilder {
self.base_topology,
Some(address(&managed_keys, gateway_identity)),
PacketSize::default(),
false,
Duration::from_millis(5),
Duration::from_millis(5),
managed_keys.ack_key(),