gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

safety(web-gateway,web-dashboard): WS topic validation, audit logging, API timeouts

Browse Source 
- 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 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-02-22 09:45:01 +01:00
parent d966f30c7f
commit 576ad25d25
7 changed files with 153 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
web-dashboard/src/lib/api.ts
View File

@@ -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<T>(
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

9
web-gateway/src/routes/backtesting.rs
View File

@@ -31,11 +31,18 @@ struct StartBacktestBody {
async fn start_backtest(
State(state): State<AppState>,
Extension(_claims): Extension<Claims>,
Extension(claims): Extension<Claims>,
Json(body): Json<StartBacktestBody>,
) -> Result<Json<serde_json::Value>, 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()

10
web-gateway/src/routes/risk.rs
View File

@@ -73,10 +73,18 @@ fn validate_emergency_stop(body: &EmergencyStopBody) -> Result<(), AppError> {
async fn emergency_stop(
State(state): State<AppState>,
Extension(_claims): Extension<Claims>,
Extension(claims): Extension<Claims>,
Json(body): Json<EmergencyStopBody>,
) -> Result<Json<serde_json::Value>, 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()

15
web-gateway/src/routes/trading.rs
View File

@@ -140,12 +140,21 @@ fn validate_order(body: &SubmitOrderBody) -> Result<(), AppError> {
async fn submit_order(
State(state): State<AppState>,
Extension(_claims): Extension<Claims>,
Extension(claims): Extension<Claims>,
Json(body): Json<SubmitOrderBody>,
) -> Result<Json<serde_json::Value>, 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<AppState>,
Extension(_claims): Extension<Claims>,
Extension(claims): Extension<Claims>,
Path(order_id): Path<String>,
) -> Result<Json<serde_json::Value>, AppError> {
tracing::info!(user = %claims.sub, order_id = %order_id, "Cancelling order");
let channel = state
.trading_channel
.as_ref()

13
web-gateway/src/routes/training.rs
View File

@@ -57,7 +57,7 @@ const VALID_MODEL_TYPES: &[&str] = &["dqn", "ppo", "tft", "mamba2"];
async fn start_job(
State(state): State<AppState>,
Extension(_claims): Extension<Claims>,
Extension(claims): Extension<Claims>,
Json(body): Json<StartJobBody>,
) -> Result<Json<serde_json::Value>, 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<AppState>,
Extension(_claims): Extension<Claims>,
Extension(claims): Extension<Claims>,
Path(id): Path<String>,
) -> Result<Json<serde_json::Value>, AppError> {
tracing::info!(user = %claims.sub, job_id = %id, "Cancelling training job");
let channel = state
.ml_training_channel
.as_ref()

14
web-gateway/src/routes/tune.rs
View File

@@ -60,10 +60,18 @@ fn validate_tune(body: &StartTuneBody) -> Result<(), AppError> {
async fn start_tune(
State(state): State<AppState>,
Extension(_claims): Extension<Claims>,
Extension(claims): Extension<Claims>,
Json(body): Json<StartTuneBody>,
) -> Result<Json<serde_json::Value>, 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<AppState>,
Extension(_claims): Extension<Claims>,
Extension(claims): Extension<Claims>,
Path(id): Path<String>,
) -> Result<Json<serde_json::Value>, AppError> {
tracing::info!(user = %claims.sub, job_id = %id, "Stopping tuning job");
let channel = state
.ml_training_channel
.as_ref()

81
web-gateway/src/ws/handler.rs
View File

@@ -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<ServerMessage> = 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);
}
}