From 576ad25d252c23a0f5648b10863ab77cb44d0d48 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sun, 22 Feb 2026 09:45:01 +0100 Subject: [PATCH] safety(web-gateway,web-dashboard): WS topic validation, audit logging, API timeouts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add VALID_TOPICS whitelist to WS handler, reject unknown topic subscriptions - Add per-connection rate limit (50 topic changes/min) and max 20 subscriptions - Remove wildcard "*" subscription (security risk) - Add tracing::info! audit logging to all 8 mutation handlers - Add 30s AbortController timeout to frontend API calls - Fix clippy: rename _claims → claims in handlers that now use it Co-Authored-By: Claude Opus 4.6 --- web-dashboard/src/lib/api.ts | 25 +++++++-- web-gateway/src/routes/backtesting.rs | 9 ++- web-gateway/src/routes/risk.rs | 10 +++- web-gateway/src/routes/trading.rs | 15 ++++- web-gateway/src/routes/training.rs | 13 ++++- web-gateway/src/routes/tune.rs | 14 ++++- web-gateway/src/ws/handler.rs | 81 ++++++++++++++++++++++++++- 7 files changed, 153 insertions(+), 14 deletions(-) diff --git a/web-dashboard/src/lib/api.ts b/web-dashboard/src/lib/api.ts index c089d7dd0..b1427d65e 100644 --- a/web-dashboard/src/lib/api.ts +++ b/web-dashboard/src/lib/api.ts @@ -5,6 +5,9 @@ const API_BASE = import.meta.env.VITE_API_URL ?? '/api'; /** Margin in seconds before actual expiry to treat the token as expired */ const EXPIRY_MARGIN_SECONDS = 60; +/** Maximum time in ms to wait for an API response before aborting */ +const REQUEST_TIMEOUT_MS = 30_000; + /** * Redirect to login with a session-expired flag so the LoginPage can * display a user-friendly message instead of a silent redirect. @@ -37,10 +40,24 @@ async function apiFetch( headers['Authorization'] = `Bearer ${token}`; } - const res = await fetch(`${API_BASE}${path}`, { - ...options, - headers, - }); + const controller = new AbortController(); + const timeoutId = setTimeout(() => controller.abort(), REQUEST_TIMEOUT_MS); + + let res: Response; + try { + res = await fetch(`${API_BASE}${path}`, { + ...options, + headers, + signal: controller.signal, + }); + } catch (err) { + if (err instanceof DOMException && err.name === 'AbortError') { + throw new Error('Request timed out. Please try again.'); + } + throw err; + } finally { + clearTimeout(timeoutId); + } if (!res.ok) { // Server rejected the token — redirect gracefully diff --git a/web-gateway/src/routes/backtesting.rs b/web-gateway/src/routes/backtesting.rs index 72bae93bb..72fe3db7a 100644 --- a/web-gateway/src/routes/backtesting.rs +++ b/web-gateway/src/routes/backtesting.rs @@ -31,11 +31,18 @@ struct StartBacktestBody { async fn start_backtest( State(state): State, - Extension(_claims): Extension, + Extension(claims): Extension, Json(body): Json, ) -> Result, AppError> { validate_backtest(&body)?; + tracing::info!( + user = %claims.sub, + strategy = %body.strategy_name, + symbols = ?body.symbols, + "Starting backtest" + ); + let channel = state .backtesting_channel .as_ref() diff --git a/web-gateway/src/routes/risk.rs b/web-gateway/src/routes/risk.rs index aad0bf45a..7ce99cf72 100644 --- a/web-gateway/src/routes/risk.rs +++ b/web-gateway/src/routes/risk.rs @@ -73,10 +73,18 @@ fn validate_emergency_stop(body: &EmergencyStopBody) -> Result<(), AppError> { async fn emergency_stop( State(state): State, - Extension(_claims): Extension, + Extension(claims): Extension, Json(body): Json, ) -> Result, AppError> { validate_emergency_stop(&body)?; + + tracing::info!( + user = %claims.sub, + stop_type = body.stop_type, + symbols = ?body.symbols, + "Emergency stop triggered" + ); + let channel = state .trading_channel .as_ref() diff --git a/web-gateway/src/routes/trading.rs b/web-gateway/src/routes/trading.rs index c07494a4e..a6ce9ca1f 100644 --- a/web-gateway/src/routes/trading.rs +++ b/web-gateway/src/routes/trading.rs @@ -140,12 +140,21 @@ fn validate_order(body: &SubmitOrderBody) -> Result<(), AppError> { async fn submit_order( State(state): State, - Extension(_claims): Extension, + Extension(claims): Extension, Json(body): Json, ) -> Result, AppError> { // Server-side input validation (defense in depth — don't trust frontend) validate_order(&body)?; + tracing::info!( + user = %claims.sub, + symbol = %body.symbol, + side = body.side, + order_type = body.order_type, + quantity = body.quantity, + "Submitting order" + ); + let channel = state .trading_channel .as_ref() @@ -171,9 +180,11 @@ async fn submit_order( async fn cancel_order( State(state): State, - Extension(_claims): Extension, + Extension(claims): Extension, Path(order_id): Path, ) -> Result, AppError> { + tracing::info!(user = %claims.sub, order_id = %order_id, "Cancelling order"); + let channel = state .trading_channel .as_ref() diff --git a/web-gateway/src/routes/training.rs b/web-gateway/src/routes/training.rs index 8c93c94dc..8194eb509 100644 --- a/web-gateway/src/routes/training.rs +++ b/web-gateway/src/routes/training.rs @@ -57,7 +57,7 @@ const VALID_MODEL_TYPES: &[&str] = &["dqn", "ppo", "tft", "mamba2"]; async fn start_job( State(state): State, - Extension(_claims): Extension, + Extension(claims): Extension, Json(body): Json, ) -> Result, AppError> { if body.model_type.is_empty() { @@ -71,6 +71,13 @@ async fn start_job( ))); } + tracing::info!( + user = %claims.sub, + model_type = %body.model_type, + use_gpu = body.use_gpu, + "Starting training job" + ); + let channel = state .ml_training_channel .as_ref() @@ -93,9 +100,11 @@ async fn start_job( async fn cancel_job( State(state): State, - Extension(_claims): Extension, + Extension(claims): Extension, Path(id): Path, ) -> Result, AppError> { + tracing::info!(user = %claims.sub, job_id = %id, "Cancelling training job"); + let channel = state .ml_training_channel .as_ref() diff --git a/web-gateway/src/routes/tune.rs b/web-gateway/src/routes/tune.rs index a3111d64f..cbdd286f7 100644 --- a/web-gateway/src/routes/tune.rs +++ b/web-gateway/src/routes/tune.rs @@ -60,10 +60,18 @@ fn validate_tune(body: &StartTuneBody) -> Result<(), AppError> { async fn start_tune( State(state): State, - Extension(_claims): Extension, + Extension(claims): Extension, Json(body): Json, ) -> Result, AppError> { validate_tune(&body)?; + + tracing::info!( + user = %claims.sub, + model_type = %body.model_type, + num_trials = body.num_trials, + "Starting tuning job" + ); + let channel = state .ml_training_channel .as_ref() @@ -107,9 +115,11 @@ async fn get_status( async fn stop_tune( State(state): State, - Extension(_claims): Extension, + Extension(claims): Extension, Path(id): Path, ) -> Result, AppError> { + tracing::info!(user = %claims.sub, job_id = %id, "Stopping tuning job"); + let channel = state .ml_training_channel .as_ref() diff --git a/web-gateway/src/ws/handler.rs b/web-gateway/src/ws/handler.rs index 8d51e9389..d027b7827 100644 --- a/web-gateway/src/ws/handler.rs +++ b/web-gateway/src/ws/handler.rs @@ -18,6 +18,24 @@ use crate::error::AppError; use crate::state::AppState; use crate::ws::messages::{ClientMessage, ServerMessage}; +/// Valid WebSocket subscription topics (must match `ServerMessage` variants) +const VALID_TOPICS: &[&str] = &[ + "market_data", + "order_update", + "risk_alert", + "position_update", + "ml_prediction", + "training_progress", + "metrics", + "config_update", +]; + +/// Maximum number of topic changes (subscribe + unsubscribe) per connection per minute +const MAX_TOPIC_CHANGES_PER_MINUTE: u32 = 50; + +/// Maximum number of topics a single client can be subscribed to simultaneously +const MAX_SUBSCRIPTIONS: usize = 20; + /// Query parameters for WebSocket upgrade (JWT passed via query string) #[derive(Debug, Deserialize)] pub struct WsParams { @@ -76,7 +94,6 @@ async fn handle_ws_connection(socket: WebSocket, state: AppState) { { let topics = topics_for_broadcast.lock().await; topics.contains(server_msg.topic()) - || topics.contains("*") // wildcard subscription } else { false }; @@ -93,6 +110,9 @@ async fn handle_ws_connection(socket: WebSocket, state: AppState) { // Task: receive client messages (subscribe/unsubscribe) and respond to pings let topics_for_recv = Arc::clone(&subscribed_topics); let mut recv_task = tokio::spawn(async move { + let mut topic_change_count: u32 = 0; + let mut rate_limit_window = tokio::time::Instant::now(); + while let Some(Ok(msg)) = ws_receiver.next().await { match msg { Message::Text(text) => { @@ -100,13 +120,40 @@ async fn handle_ws_connection(socket: WebSocket, state: AppState) { let mut topics = topics_for_recv.lock().await; match client_msg { ClientMessage::Subscribe { topics: new_topics } => { + // Rate limit check + if rate_limit_window.elapsed() >= std::time::Duration::from_secs(60) { + topic_change_count = 0; + rate_limit_window = tokio::time::Instant::now(); + } + topic_change_count += new_topics.len() as u32; + if topic_change_count > MAX_TOPIC_CHANGES_PER_MINUTE { + tracing::warn!("WebSocket client exceeded topic change rate limit"); + continue; + } + for topic in new_topics { - topics.insert(topic); + if !VALID_TOPICS.contains(&topic.as_str()) { + tracing::debug!(topic = %topic, "Rejected invalid WS topic subscription"); + continue; + } + if topics.len() < MAX_SUBSCRIPTIONS { + topics.insert(topic); + } } } ClientMessage::Unsubscribe { topics: remove_topics, } => { + if rate_limit_window.elapsed() >= std::time::Duration::from_secs(60) { + topic_change_count = 0; + rate_limit_window = tokio::time::Instant::now(); + } + topic_change_count += remove_topics.len() as u32; + if topic_change_count > MAX_TOPIC_CHANGES_PER_MINUTE { + tracing::warn!("WebSocket client exceeded topic change rate limit"); + continue; + } + for topic in &remove_topics { topics.remove(topic); } @@ -209,4 +256,34 @@ mod tests { let result = validate_ws_token(&token, &config); assert!(result.is_err()); } + + #[test] + fn test_valid_topics_matches_server_message_variants() { + use crate::ws::messages::ServerMessage; + use serde_json::json; + + // Every ServerMessage topic must be in VALID_TOPICS + let all_variants: Vec = vec![ + ServerMessage::MarketData { symbol: String::new(), data: json!(null) }, + ServerMessage::OrderUpdate { data: json!(null) }, + ServerMessage::RiskAlert { severity: String::new(), data: json!(null) }, + ServerMessage::PositionUpdate { data: json!(null) }, + ServerMessage::MlPrediction { data: json!(null) }, + ServerMessage::TrainingProgress { data: json!(null) }, + ServerMessage::Metrics { data: json!(null) }, + ServerMessage::ConfigUpdate { data: json!(null) }, + ]; + for variant in &all_variants { + assert!( + VALID_TOPICS.contains(&variant.topic()), + "ServerMessage topic '{}' not in VALID_TOPICS", + variant.topic() + ); + } + } + + #[test] + fn test_valid_topics_count() { + assert_eq!(VALID_TOPICS.len(), 8); + } }