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

All agents use the same key

Browse Source 
- 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
This commit is contained in:
dynco-nym
2024-11-21 00:18:42 +01:00
parent bacf68697c
commit d474a2840f
12 changed files with 99 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nym-node-status-agent/run.sh
+2 -20
View File
@@ -16,30 +16,13 @@ set -a
source "${monorepo_root}/envs/${ENVIRONMENT}.env"
set +a
export RUST_LOG="debug"
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"
# 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"
)
export NODE_STATUS_AGENT_AUTH_KEY="BjyC9SsHAZUzPRkQR4sPTvVrp4GgaquTh5YfSJksvvWT"
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
@@ -61,7 +44,6 @@ function swarm() {
local workers=$1
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
nym-node-status-agent/src/cli/generate_keypair.rs
+1 -4
View File
@@ -31,10 +31,7 @@ mod test {
use super::*;
use std::{
fs::{self},
path::PathBuf,
};
use std::{fs, path::PathBuf};
#[test]
fn can_generate_valid_keypair() {
nym-node-status-api/migrations/001_assigned_agent.sql
-2
View File
@@ -1,2 +0,0 @@
ALTER TABLE testruns
ADD COLUMN assigned_agent VARCHAR;
nym-node-status-api/migrations/002_last_assigned_utc.sql → nym-node-status-api/migrations/001_last_assigned_utc.sql
+3 -3
View File
@@ -1,5 +1,5 @@
ALTER TABLE testruns
ADD COLUMN last_assigned_utc INTEGER;
ALTER TABLE testruns
RENAME COLUMN timestamp_utc TO created_utc;
ALTER TABLE testruns
ADD COLUMN last_assigned_utc INTEGER;
nym-node-status-api/src/cli/mod.rs
+7
View File
@@ -73,6 +73,13 @@ pub(crate) struct Cli {
#[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_NYM_HTTP_CACHE_TTL"
)]
pub(crate) max_agent_count: i64,
}
fn parse_duration(arg: &str) -> Result<std::time::Duration, std::num::ParseIntError> {
nym-node-status-api/src/db/models.rs
-1
View File
@@ -309,7 +309,6 @@ pub struct TestRunDto {
pub created_utc: i64,
pub ip_address: String,
pub log: String,
pub assigned_agent: Option<String>,
pub last_assigned_utc: Option<i64>,
}
nym-node-status-api/src/db/queries/testruns.rs
+26 -15
View File
@@ -5,14 +5,29 @@ use crate::{
testruns::now_utc,
};
use chrono::Duration;
use nym_crypto::asymmetric::ed25519::PublicKey;
use sqlx::{pool::PoolConnection, Sqlite};
pub(crate) async fn testrun_in_progress_assigned_to_agent(
pub(crate) async fn count_testruns_in_progress(
conn: &mut PoolConnection<Sqlite>,
agent_key: &PublicKey,
) -> sqlx::Result<TestRunDto> {
let agent_key = agent_key.to_base58_string();
) -> 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,
) -> anyhow::Result<TestRunDto> {
sqlx::query_as!(
TestRunDto,
r#"SELECT
@@ -22,19 +37,20 @@ pub(crate) async fn testrun_in_progress_assigned_to_agent(
created_utc as "created_utc!",
ip_address as "ip_address!",
log as "log!",
assigned_agent,
last_assigned_utc
FROM testruns
WHERE
assigned_agent = ?
id = ?
AND
status = ?
ORDER BY created_utc"#,
agent_key,
ORDER BY created_utc
LIMIT 1"#,
testrun_id,
TestRunStatus::InProgress as i64,
)
.fetch_one(conn.as_mut())
.await
.map_err(|e| anyhow::anyhow!("Couldn't retrieve testrun {testrun_id}: {e}"))
}
pub(crate) async fn update_testruns_assigned_before(
@@ -49,8 +65,7 @@ pub(crate) async fn update_testruns_assigned_before(
r#"UPDATE
testruns
SET
status = ?,
assigned_agent = NULL
status = ?
WHERE
status = ?
AND
@@ -77,16 +92,13 @@ pub(crate) async fn update_testruns_assigned_before(
pub(crate) async fn assign_oldest_testrun(
conn: &mut PoolConnection<Sqlite>,
agent_key: PublicKey,
) -> anyhow::Result<Option<TestrunAssignment>> {
let agent_key = agent_key.to_base58_string();
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 = ?,
assigned_agent = ?,
last_assigned_utc = ?
WHERE rowid =
(
@@ -101,7 +113,6 @@ pub(crate) async fn assign_oldest_testrun(
gateway_id
"#,
TestRunStatus::InProgress as i64,
agent_key,
now,
TestRunStatus::Queued as i64,
)
nym-node-status-api/src/http/api/testruns.rs
+45 -38
View File
@@ -9,6 +9,7 @@ use reqwest::StatusCode;
use crate::db::models::TestRunStatus;
use crate::db::queries;
use crate::testruns::now_utc;
use crate::{
db,
http::{
@@ -35,15 +36,9 @@ async fn request_testrun(
) -> HttpResult<Json<TestrunAssignment>> {
// TODO dz log agent's network probe version
authenticate(&request, &state)?;
state
.update_last_request_time(
&request.payload.agent_public_key,
&request.payload.timestamp,
)
.await?;
is_fresh(&request.payload.timestamp)?;
let agent_pubkey = request.payload.agent_public_key;
tracing::debug!("Agent {} requested testrun", agent_pubkey);
tracing::debug!("Agent requested testrun");
let db = state.db_pool();
let mut conn = db
@@ -51,31 +46,29 @@ async fn request_testrun(
.await
.map_err(HttpError::internal_with_logging)?;
if let Ok(testrun) =
db::queries::testruns::testrun_in_progress_assigned_to_agent(&mut conn, &agent_pubkey).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!(
"Testrun {} already in progress for agent {:?}, rejecting",
testrun.id,
testrun.assigned_agent
"{}/{} testruns in progress, rejecting",
active_testruns,
state.agent_max_count()
);
return Err(HttpError::invalid_input(
"Testrun already in progress for this agent",
));
};
return Err(HttpError::no_testruns_available());
}
return match db::queries::testruns::assign_oldest_testrun(&mut conn, agent_pubkey).await {
return match db::queries::testruns::assign_oldest_testrun(&mut conn).await {
Ok(res) => {
if let Some(testrun) = res {
tracing::info!(
"🏃‍ Assigned testrun row_id {} gateway {} to agent {}",
"🏃‍ Assigned testrun row_id {} gateway {} to agent",
&testrun.testrun_id,
testrun.gateway_identity_key,
agent_pubkey
);
Ok(Json(testrun))
} else {
tracing::debug!("No testruns available for agent {}", agent_pubkey);
tracing::debug!("No testruns available");
Err(HttpError::no_testruns_available())
}
}
@@ -97,23 +90,17 @@ async fn submit_testrun(
.await
.map_err(HttpError::internal_with_logging)?;
let submitter_pubkey = submitted_result.payload.agent_public_key;
let assigned_testrun =
queries::testruns::testrun_in_progress_assigned_to_agent(&mut conn, &submitter_pubkey)
queries::testruns::get_in_progress_testrun_by_id(&mut conn, submitted_testrun_id)
.await
.map_err(|err| {
tracing::warn!("No testruns in progress for agent {submitter_pubkey}: {err}");
tracing::warn!(
"No testruns in progress for testrun_id {}: {}",
submitted_testrun_id,
err
);
HttpError::invalid_input("Invalid testrun submitted")
})?;
if submitted_testrun_id != assigned_testrun.id {
tracing::warn!(
"Agent {} submitted testrun {} but {} was expected",
submitter_pubkey,
submitted_testrun_id,
assigned_testrun.id
);
return 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",
@@ -132,8 +119,7 @@ async fn submit_testrun(
))
})?;
tracing::debug!(
"Agent {} submitted testrun {} for gateway {} ({} bytes)",
submitter_pubkey,
"Agent submitted testrun {} for gateway {} ({} bytes)",
submitted_testrun_id,
gw_identity,
&submitted_result.payload.probe_result.len(),
@@ -165,10 +151,20 @@ async fn submit_testrun(
.await
.map_err(HttpError::internal_with_logging)?;
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 row_id {} for gateway {} complete (last assigned at {}, created at {})",
assigned_testrun.id,
gw_identity
gw_identity,
last_assigned,
created_at
);
Ok(StatusCode::CREATED)
@@ -190,6 +186,17 @@ fn authenticate(request: &impl VerifiableRequest, state: &AppState) -> HttpResul
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();
nym-node-status-api/src/http/server.rs
+2 -1
View File
@@ -16,10 +16,11 @@ pub(crate) async fn start_http_api(
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, agent_key_list);
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?
nym-node-status-api/src/http/state.rs
+12 -50
View File
@@ -1,4 +1,4 @@
use std::{collections::HashMap, sync::Arc, time::Duration};
use std::{sync::Arc, time::Duration};
use moka::{future::Cache, Entry};
use nym_crypto::asymmetric::ed25519::PublicKey;
@@ -6,11 +6,7 @@ use tokio::sync::RwLock;
use crate::{
db::DbPool,
http::{
error::{HttpError, HttpResult},
models::{DailyStats, Gateway, Mixnode, SummaryHistory},
},
testruns::now_utc,
http::models::{DailyStats, Gateway, Mixnode, SummaryHistory},
};
#[derive(Debug, Clone)]
@@ -18,18 +14,21 @@ pub(crate) struct AppState {
db_pool: DbPool,
cache: HttpCache,
agent_key_list: Vec<PublicKey>,
/// last time agent requested a testrun
// if performance becomes a problem, consider a faster hashmap like `scc``
agent_last_request_times: Arc<RwLock<HashMap<String, i64>>>,
agent_max_count: i64,
}
impl AppState {
pub(crate) fn new(db_pool: DbPool, cache_ttl: u64, agent_key_list: Vec<PublicKey>) -> 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_last_request_times: Arc::new(RwLock::new(HashMap::new())),
agent_max_count,
}
}
@@ -45,45 +44,8 @@ impl AppState {
self.agent_key_list.contains(agent_pubkey)
}
/// Only updates if request is not a replay. Otherwise return error
pub(crate) async fn update_last_request_time(
&self,
agent_key: &PublicKey,
request_time: &i64,
) -> HttpResult<()> {
// if a request took longer than N minutes to reach NS API, something is very wrong
let cutoff_duration = chrono::Duration::minutes(1);
let cutoff_timestamp = (now_utc() - cutoff_duration).timestamp();
if *request_time < cutoff_timestamp {
tracing::warn!("Request older than {}s, rejecting", cutoff_timestamp);
return Err(HttpError::unauthorized());
}
// if a previous entry has a newer time than this request's submit time,
// it's a repeated request
let agent_key = agent_key.to_base58_string();
let request_times = self.agent_last_request_times.read().await;
if let Some(previous_request_time) = request_times.get(&agent_key) {
if request_time <= previous_request_time {
tracing::warn!(
"Request has timestamp {} but previous request was at {}, rejecting",
request_time,
previous_request_time
);
return Err(HttpError::unauthorized());
}
}
drop(request_times);
// otherwise this is a newer (or a first) request
self.agent_last_request_times
.write()
.await
.entry(agent_key)
.and_modify(|value| *value = *request_time)
.or_insert(*request_time);
Ok(())
pub(crate) fn agent_max_count(&self) -> i64 {
self.agent_max_count
}
}
nym-node-status-api/src/main.rs
+1
View File
@@ -47,6 +47,7 @@ async fn main() -> anyhow::Result<()> {
args.http_port,
args.nym_http_cache_ttl,
agent_key_list.to_owned(),
args.max_agent_count,
)
.await
.expect("Failed to start server");
nym-node-status-api/src/testruns/queue.rs
-1
View File
@@ -56,7 +56,6 @@ pub(crate) async fn try_queue_testrun(
created_utc as "created_utc!",
ip_address as "ip_address!",
log as "log!",
assigned_agent,
last_assigned_utc
FROM testruns
WHERE gateway_id = ? AND status != 2