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

NS Agent auth with NS API (#5127)

Browse Source 
* 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
This commit is contained in:
Dinko Zdravac
2024-11-21 14:32:15 +01:00
committed by GitHub
parent aa460076f4
commit 3f072e4e9d
27 changed files with 668 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nym-node-status-api/src/cli/mod.rs
+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_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
+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)]
nym-node-status-api/src/db/queries/testruns.rs
+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)
nym-node-status-api/src/http/api/testruns.rs
+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();
nym-node-status-api/src/http/error.rs
+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 {
nym-node-status-api/src/http/server.rs
+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?
nym-node-status-api/src/http/state.rs
+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";
nym-node-status-api/src/main.rs
+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");
nym-node-status-api/src/testruns/mod.rs
+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(())
}
nym-node-status-api/src/testruns/queue.rs
+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,