From f5ab647a7ab1d6928a15187404bc7485e2ceaa2b Mon Sep 17 00:00:00 2001 From: dynco-nym <173912580+dynco-nym@users.noreply.github.com> Date: Mon, 18 Nov 2024 12:17:42 +0100 Subject: [PATCH] /submit with better auth - also adjust agent run script to authenticate, even in parallel --- common/models/src/ns_api.rs | 3 +- nym-node-status-agent/run.sh | 44 +++++++--- .../src/cli/generate_keypair.rs | 1 - nym-node-status-agent/src/cli/run_probe.rs | 1 - nym-node-status-api/launch_node_status_api.sh | 14 ++- .../migrations/001_assigned_agent.sql | 2 + nym-node-status-api/src/cli/mod.rs | 27 +----- nym-node-status-api/src/db/models.rs | 10 +++ .../src/db/queries/testruns.rs | 45 +++++++++- nym-node-status-api/src/http/api/testruns.rs | 87 ++++++++++++------- nym-node-status-api/src/http/error.rs | 2 +- nym-node-status-api/src/main.rs | 7 +- nym-node-status-api/src/testruns/mod.rs | 2 +- nym-node-status-api/src/testruns/queue.rs | 3 +- 14 files changed, 169 insertions(+), 79 deletions(-) create mode 100644 nym-node-status-api/migrations/001_assigned_agent.sql diff --git a/common/models/src/ns_api.rs b/common/models/src/ns_api.rs index 263628eb40..306fd2ab93 100644 --- a/common/models/src/ns_api.rs +++ b/common/models/src/ns_api.rs @@ -1,4 +1,4 @@ -use nym_crypto::asymmetric::ed25519::{PublicKey, Signature}; +use nym_crypto::asymmetric::ed25519::Signature; use serde::{Deserialize, Serialize}; #[derive(Debug, Clone, Deserialize, Serialize)] @@ -11,5 +11,4 @@ pub struct TestrunAssignment { pub struct SubmitResults { pub message: String, pub signature: Signature, - pub public_key: PublicKey, } diff --git a/nym-node-status-agent/run.sh b/nym-node-status-agent/run.sh index 46b997d3c0..af1fa2353e 100755 --- a/nym-node-status-agent/run.sh +++ b/nym-node-status-agent/run.sh @@ -1,23 +1,46 @@ #!/bin/bash set -eu - environment="qa" probe_git_ref="0dd5dacdda92b1ddd51cd30a3399515e45613371" -export RUST_LOG="debug" - 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="debug" 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" + +# each key is required for a separate agent when running in parallel: their +# public counterparts need to be registered with NS API +private_keys=("BjyC9SsHAZUzPRkQR4sPTvVrp4GgaquTh5YfSJksvvWT" + "4RqSKydrEuyGF8Xtwyauvja62SAjqxFPYQzW2neZdkQD" + "CfudaSaaLTkgR8rkBijUnYocdFciWTbKqkSNYevepnbn" + "Dd3fDyPUg4edTpiCAkSweE17NdWJ7gAchbtqAeSj3MBc" + "HAtfcfnpw5XdpcVzAH6Qsxp6Zf75oe2W54HjTD8ngVbU" + "8aqWP8wZyhX5W8gfrvyh1SmS6CEgfLAR95DBhWXRUpTm" + "234U1PMkWpAsn7hD98g1D8hfRFkJJS91T2sJBQDyXmqx" + "5qUUFu83fgqpACsr3dC6iwGJxhTqN4JJDTecT2QSqwEe" + "4Pp7Cd9G3aMku9biFcxRMEja8RBMbBRGZuDAZt1yTS7H" + "4U136QykP8G831EZSDNLLvgWCGYA8naYtT8BQ9kLeL5B" +) + +workers=${1:-1} +if ((workers > ${#private_keys[@]})); then + echo "Error: ${workers} is larger than the number of private keys available (${#private_keys[@]})." + exit 1 +fi +echo "Running $workers workers in parallel" # build & copy over GW probe function copy_gw_probe() { @@ -36,11 +59,9 @@ 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 + export NODE_STATUS_AGENT_AUTH_KEY=${private_keys[i]} ../target/release/nym-node-status-agent run-probe & done @@ -50,7 +71,6 @@ function swarm() { } copy_gw_probe +build_agent -swarm 8 - -# cargo run -- run-probe +swarm $workers diff --git a/nym-node-status-agent/src/cli/generate_keypair.rs b/nym-node-status-agent/src/cli/generate_keypair.rs index ba86db838d..e89bf3e8d5 100644 --- a/nym-node-status-agent/src/cli/generate_keypair.rs +++ b/nym-node-status-agent/src/cli/generate_keypair.rs @@ -1,5 +1,4 @@ use std::{fs::File, io::Write, path::Path}; - use tracing::info; pub(crate) fn generate_key_pair(path: impl AsRef) -> anyhow::Result<()> { diff --git a/nym-node-status-agent/src/cli/run_probe.rs b/nym-node-status-agent/src/cli/run_probe.rs index 3c8ce13509..aee1116f4a 100644 --- a/nym-node-status-agent/src/cli/run_probe.rs +++ b/nym-node-status-agent/src/cli/run_probe.rs @@ -114,6 +114,5 @@ fn sign_message(key: PrivateKey, probe_outcome: String) -> SubmitResults { SubmitResults { message: probe_outcome, signature, - public_key: key.public_key(), } } diff --git a/nym-node-status-api/launch_node_status_api.sh b/nym-node-status-api/launch_node_status_api.sh index 20c1516749..83bd3dac93 100755 --- a/nym-node-status-api/launch_node_status_api.sh +++ b/nym-node-status-api/launch_node_status_api.sh @@ -6,8 +6,18 @@ 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_AGENT_KEY_LIST="H4z8kx5Kkf5JMQHhxaW1MwYndjKCDHC7HsVhHTFfBZ4J,AKYZeKaJTzJrWtidk3BZz7wsm4GapaTY4GwQYVL43cmc" +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" diff --git a/nym-node-status-api/migrations/001_assigned_agent.sql b/nym-node-status-api/migrations/001_assigned_agent.sql new file mode 100644 index 0000000000..96ec895924 --- /dev/null +++ b/nym-node-status-api/migrations/001_assigned_agent.sql @@ -0,0 +1,2 @@ +ALTER TABLE testruns +ADD COLUMN assigned_agent VARCHAR; diff --git a/nym-node-status-api/src/cli/mod.rs b/nym-node-status-api/src/cli/mod.rs index 0ca2b9bf48..fa3d3c12bf 100644 --- a/nym-node-status-api/src/cli/mod.rs +++ b/nym-node-status-api/src/cli/mod.rs @@ -1,6 +1,5 @@ use clap::Parser; use nym_bin_common::bin_info; -use nym_crypto::asymmetric::ed25519::PublicKey; use reqwest::Url; use std::{sync::OnceLock, time::Duration}; @@ -71,29 +70,9 @@ pub(crate) struct Cli { #[arg(value_parser = parse_duration)] pub(crate) testruns_refresh_interval: Duration, - #[clap(long, env = "NODE_STATUS_API_AGENT_KEY_LIST")] - #[arg(value_parser = parse_key_list)] - agent_key_list: KeyList, -} - -impl Cli { - pub(crate) fn agent_key_list(&self) -> &Vec { - &self.agent_key_list.0 - } -} - -// We need a list of keys from clap. But if we define CLI argument as Vec, -// clap interprets that as type T which can be given as a CLI argument multiple -// times (so all of them are stored in a Vec). Thus we wrap Vec in a newtype -// pattern to have a list of keys and make clap happy. -#[derive(Debug, Clone)] -struct KeyList(Vec); - -fn parse_key_list(arg: &str) -> anyhow::Result { - arg.split(',') - .map(|value| PublicKey::from_base58_string(value.trim()).map_err(anyhow::Error::from)) - .collect::>>() - .map(KeyList) + #[clap(env = "NODE_STATUS_API_AGENT_KEY_LIST")] + #[arg(value_delimiter = ',')] + pub(crate) agent_key_list: Vec, } fn parse_duration(arg: &str) -> Result { diff --git a/nym-node-status-api/src/db/models.rs b/nym-node-status-api/src/db/models.rs index a5511787f9..9e9f5aa807 100644 --- a/nym-node-status-api/src/db/models.rs +++ b/nym-node-status-api/src/db/models.rs @@ -2,6 +2,7 @@ use crate::{ http::{self, models::SummaryHistory}, monitor::NumericalCheckedCast, }; +use nym_crypto::asymmetric::ed25519::PublicKey; use nym_node_requests::api::v1::node::models::NodeDescription; use serde::{Deserialize, Serialize}; use strum_macros::{EnumString, FromRepr}; @@ -309,6 +310,15 @@ pub struct TestRunDto { pub timestamp_utc: i64, pub ip_address: String, pub log: String, + pub assigned_agent: Option, +} + +impl TestRunDto { + pub(crate) fn assigned_agent_key(&self) -> Option { + self.assigned_agent + .as_ref() + .and_then(|value| PublicKey::from_base58_string(value).ok()) + } } #[derive(Debug, Clone, strum_macros::Display, EnumString, FromRepr, PartialEq)] diff --git a/nym-node-status-api/src/db/queries/testruns.rs b/nym-node-status-api/src/db/queries/testruns.rs index 91a3a86b13..38582e2a5b 100644 --- a/nym-node-status-api/src/db/queries/testruns.rs +++ b/nym-node-status-api/src/db/queries/testruns.rs @@ -6,6 +6,7 @@ use crate::{ }; use anyhow::Context; use chrono::Duration; +use nym_crypto::asymmetric::ed25519::PublicKey; use sqlx::{pool::PoolConnection, Sqlite}; pub(crate) async fn get_in_progress_testrun_by_id( @@ -20,7 +21,8 @@ pub(crate) async fn get_in_progress_testrun_by_id( status as "status!", timestamp_utc as "timestamp_utc!", ip_address as "ip_address!", - log as "log!" + log as "log!", + assigned_agent FROM testruns WHERE id = ? @@ -35,6 +37,35 @@ pub(crate) async fn get_in_progress_testrun_by_id( .context(format!("Couldn't retrieve testrun {testrun_id}")) } +pub(crate) async fn get_testruns_assigned_to_agent( + conn: &mut PoolConnection, + agent_key: PublicKey, +) -> anyhow::Result { + let agent_key = agent_key.to_base58_string(); + sqlx::query_as!( + TestRunDto, + r#"SELECT + id as "id!", + gateway_id as "gateway_id!", + status as "status!", + timestamp_utc as "timestamp_utc!", + ip_address as "ip_address!", + log as "log!", + assigned_agent + FROM testruns + WHERE + assigned_agent = ? + AND + status = ? + ORDER BY timestamp_utc"#, + agent_key, + TestRunStatus::InProgress as i64, + ) + .fetch_one(conn.as_mut()) + .await + .context(format!("No testruns in progress for agent {agent_key}")) +} + pub(crate) async fn update_testruns_older_than(db: &DbPool, age: Duration) -> anyhow::Result { let mut conn = db.acquire().await?; let previous_run = now_utc() - age; @@ -44,7 +75,8 @@ pub(crate) async fn update_testruns_older_than(db: &DbPool, age: Duration) -> an r#"UPDATE testruns SET - status = ? + status = ?, + assigned_agent = NULL WHERE status = ? AND @@ -69,13 +101,17 @@ 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, + agent_key: PublicKey, ) -> anyhow::Result> { + let agent_key = agent_key.to_base58_string(); // find & mark as "In progress" in the same transaction to avoid race conditions let returning = sqlx::query!( r#"UPDATE testruns - SET status = ? + SET + status = ?, + assigned_agent = ? WHERE rowid = ( SELECT rowid @@ -89,6 +125,7 @@ pub(crate) async fn get_oldest_testrun_and_make_it_pending( gateway_id "#, TestRunStatus::InProgress as i64, + agent_key, TestRunStatus::Queued as i64, ) .fetch_optional(conn.as_mut()) diff --git a/nym-node-status-api/src/http/api/testruns.rs b/nym-node-status-api/src/http/api/testruns.rs index 02074f4f3b..5e19105e53 100644 --- a/nym-node-status-api/src/http/api/testruns.rs +++ b/nym-node-status-api/src/http/api/testruns.rs @@ -45,7 +45,7 @@ async fn request_testrun( .await .map_err(HttpError::internal_with_logging)?; - return match db::queries::testruns::get_oldest_testrun_and_make_it_pending(&mut conn).await { + return match db::queries::testruns::assign_oldest_testrun(&mut conn, agent_pubkey).await { Ok(res) => { if let Some(testrun) = res { tracing::info!( @@ -66,11 +66,44 @@ async fn request_testrun( #[tracing::instrument(level = "debug", skip_all)] async fn submit_testrun( - Path(testrun_id): Path, + Path(submitted_testrun_id): Path, State(state): State, Json(probe_results): Json, ) -> HttpResult { - let agent_pubkey = authenticate_agent(&probe_results.public_key.to_base58_string(), &state)?; + let db = state.db_pool(); + let mut conn = db + .acquire() + .await + .map_err(HttpError::internal_with_logging)?; + + let submitted_testrun = + queries::testruns::get_in_progress_testrun_by_id(&mut conn, submitted_testrun_id) + .await + .map_err(|e| { + tracing::warn!("testrun_id {} not found: {}", submitted_testrun_id, e); + HttpError::not_found(submitted_testrun_id) + })?; + let agent_pubkey = submitted_testrun + .assigned_agent_key() + .ok_or_else(HttpError::unauthorized)?; + + let assigned_testrun = + queries::testruns::get_testruns_assigned_to_agent(&mut conn, agent_pubkey) + .await + .map_err(|err| { + tracing::warn!("{err}"); + HttpError::invalid_input("Invalid testrun submitted") + })?; + if submitted_testrun_id != assigned_testrun.id { + tracing::warn!( + "Agent {} submitted testrun {} but {} was expected", + agent_pubkey, + submitted_testrun_id, + assigned_testrun.id + ); + return Err(HttpError::invalid_input("Invalid testrun submitted")); + } + agent_pubkey .verify(&probe_results.message, &probe_results.signature) .map_err(|_| { @@ -78,55 +111,51 @@ async fn submit_testrun( HttpError::unauthorized() })?; - 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 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)", agent_pubkey, - testrun_id, + submitted_testrun_id, gw_identity, &probe_results.message.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) - .await - .map_err(HttpError::internal_with_logging)?; + 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, - testrun.gateway_id, + assigned_testrun.gateway_id, &probe_results.message, ) .await .map_err(HttpError::internal_with_logging)?; let result = get_result_from_log(&probe_results.message); - 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) + 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)?; tracing::info!( "✅ Testrun row_id {} for gateway {} complete", - testrun.id, + assigned_testrun.id, gw_identity ); diff --git a/nym-node-status-api/src/http/error.rs b/nym-node-status-api/src/http/error.rs index 021c993827..4ce1ca1ce1 100644 --- a/nym-node-status-api/src/http/error.rs +++ b/nym-node-status-api/src/http/error.rs @@ -17,7 +17,7 @@ impl HttpError { pub(crate) fn unauthorized() -> Self { Self { - message: serde_json::json!({"message": "Make sure your public key si registered with NS API"}).to_string(), + message: serde_json::json!({"message": "Make sure your public key is registered with NS API"}).to_string(), status: axum::http::StatusCode::UNAUTHORIZED, } } diff --git a/nym-node-status-api/src/main.rs b/nym-node-status-api/src/main.rs index facb4485b9..0833c6a7b4 100644 --- a/nym-node-status-api/src/main.rs +++ b/nym-node-status-api/src/main.rs @@ -1,4 +1,5 @@ use clap::Parser; +use nym_crypto::asymmetric::ed25519::PublicKey; use nym_task::signal::wait_for_signal; mod cli; @@ -14,7 +15,11 @@ async fn main() -> anyhow::Result<()> { let args = cli::Cli::parse(); - let agent_key_list = args.agent_key_list(); + let agent_key_list = args + .agent_key_list + .iter() + .map(|value| PublicKey::from_base58_string(value.trim()).map_err(anyhow::Error::from)) + .collect::>>()?; tracing::info!("Registered {} agent keys", agent_key_list.len()); let connection_url = args.database_url.clone(); diff --git a/nym-node-status-api/src/testruns/mod.rs b/nym-node-status-api/src/testruns/mod.rs index 7ac916fc8e..522060850a 100644 --- a/nym-node-status-api/src/testruns/mod.rs +++ b/nym-node-status-api/src/testruns/mod.rs @@ -71,7 +71,7 @@ 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(()) } diff --git a/nym-node-status-api/src/testruns/queue.rs b/nym-node-status-api/src/testruns/queue.rs index 88804fff1b..ec9676ed70 100644 --- a/nym-node-status-api/src/testruns/queue.rs +++ b/nym-node-status-api/src/testruns/queue.rs @@ -55,7 +55,8 @@ pub(crate) async fn try_queue_testrun( status as "status!", timestamp_utc as "timestamp_utc!", ip_address as "ip_address!", - log as "log!" + log as "log!", + assigned_agent FROM testruns WHERE gateway_id = ? AND status != 2 ORDER BY id DESC