From 7de84bb723a04b718df2ea6a2df4e7c90a5f6c0f Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sun, 1 Mar 2026 00:01:53 +0100 Subject: [PATCH] fix(web-gateway): add service-to-service JWT auth for gRPC calls to api-gateway Web-gateway routes through api-gateway for the foxhunt.tli proto, but api-gateway requires JWT Bearer auth on all gRPC requests. This adds a service_auth module that mints short-lived service JWTs (matching the api-gateway's expected claims: iss, aud, roles, permissions) and injects them into all 15 REST proxy calls and 4 gRPC stream bridges. Co-Authored-By: Claude Opus 4.6 --- crates/web-gateway/src/grpc/mod.rs | 1 + crates/web-gateway/src/grpc/service_auth.rs | 71 ++++++++++++++++++++ crates/web-gateway/src/grpc/streams.rs | 34 ++++++---- crates/web-gateway/src/main.rs | 6 +- crates/web-gateway/src/routes/config.rs | 7 +- crates/web-gateway/src/routes/ml.rs | 13 ++-- crates/web-gateway/src/routes/performance.rs | 13 ++-- crates/web-gateway/src/routes/risk.rs | 9 +-- crates/web-gateway/src/routes/trading.rs | 17 ++--- 9 files changed, 131 insertions(+), 40 deletions(-) create mode 100644 crates/web-gateway/src/grpc/service_auth.rs diff --git a/crates/web-gateway/src/grpc/mod.rs b/crates/web-gateway/src/grpc/mod.rs index e6af67a86..82698e465 100644 --- a/crates/web-gateway/src/grpc/mod.rs +++ b/crates/web-gateway/src/grpc/mod.rs @@ -1,2 +1,3 @@ pub mod clients; +pub(crate) mod service_auth; pub mod streams; diff --git a/crates/web-gateway/src/grpc/service_auth.rs b/crates/web-gateway/src/grpc/service_auth.rs new file mode 100644 index 000000000..0cded3105 --- /dev/null +++ b/crates/web-gateway/src/grpc/service_auth.rs @@ -0,0 +1,71 @@ +//! Service-to-service JWT authentication for internal gRPC calls. +//! +//! When web-gateway proxies HTTP requests to the api-gateway via gRPC, the +//! api-gateway requires a valid JWT Bearer token. This module generates +//! short-lived service tokens using the shared JWT_SECRET so internal +//! gRPC calls are authenticated without forwarding user credentials. + +use std::time::{SystemTime, UNIX_EPOCH}; + +/// JWT claims for internal service-to-service authentication. +/// Matches the api-gateway's expected `JwtClaims` structure. +#[derive(serde::Serialize)] +struct ServiceClaims { + jti: String, + sub: String, + iat: u64, + exp: u64, + nbf: u64, + iss: String, + aud: String, + roles: Vec, + permissions: Vec, + token_type: String, + session_id: String, +} + +/// Generate a Bearer metadata value containing a short-lived service JWT. +/// Returns `None` if the secret is empty or token generation fails. +fn service_bearer_value( + jwt_secret: &str, +) -> Option> { + if jwt_secret.is_empty() { + return None; + } + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .ok()? + .as_secs(); + + let claims = ServiceClaims { + jti: uuid::Uuid::new_v4().to_string(), + sub: "web-gateway-service".to_string(), + iat: now, + exp: now + 3599, // just under 1h (api-gateway enforces max 1h token age) + nbf: now, + iss: "foxhunt-api-gateway".to_string(), + aud: "foxhunt-services".to_string(), + roles: vec!["service".to_string()], + permissions: vec!["api.access".to_string()], + token_type: "access".to_string(), + session_id: uuid::Uuid::new_v4().to_string(), + }; + + let key = jsonwebtoken::EncodingKey::from_secret(jwt_secret.as_bytes()); + let token = jsonwebtoken::encode(&jsonwebtoken::Header::default(), &claims, &key).ok()?; + format!("Bearer {}", token).parse().ok() +} + +/// Inject service auth metadata into a tonic request. +pub(crate) fn inject_auth(request: &mut tonic::Request, jwt_secret: &str) { + if let Some(val) = service_bearer_value(jwt_secret) { + request.metadata_mut().insert("authorization", val); + } +} + +/// Create a tonic request with service auth already injected. +pub(crate) fn authed_request(inner: T, jwt_secret: &str) -> tonic::Request { + let mut request = tonic::Request::new(inner); + inject_auth(&mut request, jwt_secret); + request +} diff --git a/crates/web-gateway/src/grpc/streams.rs b/crates/web-gateway/src/grpc/streams.rs index 1929abbf5..e82d24ec2 100644 --- a/crates/web-gateway/src/grpc/streams.rs +++ b/crates/web-gateway/src/grpc/streams.rs @@ -5,6 +5,7 @@ use tonic::transport::Channel; use tracing::{error, info, warn}; use crate::grpc::clients::trading_client; +use crate::grpc::service_auth::inject_auth; use crate::proto::trading::{ SubscribeMarketDataRequest, SubscribeMetricsRequest, SubscribeOrderUpdatesRequest, SubscribeRiskAlertsRequest, @@ -19,33 +20,38 @@ use crate::ws::messages::ServerMessage; pub fn start_grpc_stream_bridges( trading_channel: Option, ws_broadcast: broadcast::Sender, + jwt_secret: String, ) { if let Some(channel) = trading_channel { // Market data stream bridge let tx_md = ws_broadcast.clone(); let ch_md = channel.clone(); + let secret_md = jwt_secret.clone(); tokio::spawn(async move { - stream_market_data(ch_md, tx_md).await; + stream_market_data(ch_md, tx_md, &secret_md).await; }); // Order updates stream bridge let tx_orders = ws_broadcast.clone(); let ch_orders = channel.clone(); + let secret_orders = jwt_secret.clone(); tokio::spawn(async move { - stream_order_updates(ch_orders, tx_orders).await; + stream_order_updates(ch_orders, tx_orders, &secret_orders).await; }); // Risk alerts stream bridge let tx_risk = ws_broadcast.clone(); let ch_risk = channel.clone(); + let secret_risk = jwt_secret.clone(); tokio::spawn(async move { - stream_risk_alerts(ch_risk, tx_risk).await; + stream_risk_alerts(ch_risk, tx_risk, &secret_risk).await; }); // Metrics stream bridge let tx_metrics = ws_broadcast.clone(); + let secret_metrics = jwt_secret; tokio::spawn(async move { - stream_metrics(channel, tx_metrics).await; + stream_metrics(channel, tx_metrics, &secret_metrics).await; }); } @@ -57,13 +63,14 @@ pub fn start_grpc_stream_bridges( } /// Bridge `SubscribeMarketData` gRPC stream to WebSocket broadcast -async fn stream_market_data(channel: Channel, tx: broadcast::Sender) { +async fn stream_market_data(channel: Channel, tx: broadcast::Sender, jwt_secret: &str) { reconnect_loop("market_data", || async { let mut client = trading_client(&channel); - let request = tonic::Request::new(SubscribeMarketDataRequest { + let mut request = tonic::Request::new(SubscribeMarketDataRequest { symbols: vec![], // Subscribe to all symbols data_types: vec![], }); + inject_auth(&mut request, jwt_secret); let response = client.subscribe_market_data(request).await?; let mut stream = response.into_inner(); @@ -86,10 +93,11 @@ async fn stream_market_data(channel: Channel, tx: broadcast::Sender) { } /// Bridge `SubscribeOrderUpdates` gRPC stream to WebSocket broadcast -async fn stream_order_updates(channel: Channel, tx: broadcast::Sender) { +async fn stream_order_updates(channel: Channel, tx: broadcast::Sender, jwt_secret: &str) { reconnect_loop("order_update", || async { let mut client = trading_client(&channel); - let request = tonic::Request::new(SubscribeOrderUpdatesRequest { account_id: None }); + let mut request = tonic::Request::new(SubscribeOrderUpdatesRequest { account_id: None }); + inject_auth(&mut request, jwt_secret); let response = client.subscribe_order_updates(request).await?; let mut stream = response.into_inner(); @@ -109,13 +117,14 @@ async fn stream_order_updates(channel: Channel, tx: broadcast::Sender) { } /// Bridge `SubscribeRiskAlerts` gRPC stream to WebSocket broadcast -async fn stream_risk_alerts(channel: Channel, tx: broadcast::Sender) { +async fn stream_risk_alerts(channel: Channel, tx: broadcast::Sender, jwt_secret: &str) { reconnect_loop("risk_alert", || async { let mut client = trading_client(&channel); - let request = tonic::Request::new(SubscribeRiskAlertsRequest { + let mut request = tonic::Request::new(SubscribeRiskAlertsRequest { min_severity: vec![], symbols: vec![], }); + inject_auth(&mut request, jwt_secret); let response = client.subscribe_risk_alerts(request).await?; let mut stream = response.into_inner(); @@ -138,13 +147,14 @@ async fn stream_risk_alerts(channel: Channel, tx: broadcast::Sender) { } /// Bridge `SubscribeMetrics` gRPC stream to WebSocket broadcast -async fn stream_metrics(channel: Channel, tx: broadcast::Sender) { +async fn stream_metrics(channel: Channel, tx: broadcast::Sender, jwt_secret: &str) { reconnect_loop("metrics", || async { let mut client = trading_client(&channel); - let request = tonic::Request::new(SubscribeMetricsRequest { + let mut request = tonic::Request::new(SubscribeMetricsRequest { metric_names: vec![], interval_seconds: 5, }); + inject_auth(&mut request, jwt_secret); let response = client.subscribe_metrics(request).await?; let mut stream = response.into_inner(); diff --git a/crates/web-gateway/src/main.rs b/crates/web-gateway/src/main.rs index 89a51d2d5..fdb180e21 100644 --- a/crates/web-gateway/src/main.rs +++ b/crates/web-gateway/src/main.rs @@ -44,7 +44,11 @@ async fn main() -> Result<()> { let state = AppState::new(config.clone()).await?; // Start gRPC stream bridge tasks (forward gRPC streams to WebSocket broadcast) - start_grpc_stream_bridges(state.trading_channel.clone(), state.ws_broadcast.clone()); + start_grpc_stream_bridges( + state.trading_channel.clone(), + state.ws_broadcast.clone(), + config.jwt_secret.clone(), + ); // Build CORS from configured origins with restricted methods/headers let origins: Vec = config diff --git a/crates/web-gateway/src/routes/config.rs b/crates/web-gateway/src/routes/config.rs index 9ef648974..f1e2b737c 100644 --- a/crates/web-gateway/src/routes/config.rs +++ b/crates/web-gateway/src/routes/config.rs @@ -7,6 +7,7 @@ use axum::{ use crate::auth::claims::Claims; use crate::error::AppError; use crate::grpc::clients::trading_client; +use crate::grpc::service_auth::authed_request; use crate::proto::trading::{GetConfigRequest, UpdateParametersRequest}; use crate::state::AppState; @@ -26,7 +27,7 @@ async fn get_config( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .get_config(tonic::Request::new(GetConfigRequest { keys: vec![] })) + .get_config(authed_request(GetConfigRequest { keys: vec![] }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, @@ -50,10 +51,10 @@ async fn update_config( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .update_parameters(tonic::Request::new(UpdateParametersRequest { + .update_parameters(authed_request(UpdateParametersRequest { parameters: body.parameters, persist: body.persist, - })) + }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, diff --git a/crates/web-gateway/src/routes/ml.rs b/crates/web-gateway/src/routes/ml.rs index 4959ddae7..a9320b810 100644 --- a/crates/web-gateway/src/routes/ml.rs +++ b/crates/web-gateway/src/routes/ml.rs @@ -7,6 +7,7 @@ use axum::{ use crate::auth::claims::Claims; use crate::error::AppError; use crate::grpc::clients::trading_client; +use crate::grpc::service_auth::authed_request; use crate::proto::trading::{GetMlPerformanceRequest, GetMlPredictionsRequest, SubmitMlOrderRequest}; use crate::state::AppState; @@ -35,11 +36,11 @@ async fn get_predictions( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .get_ml_predictions(tonic::Request::new(GetMlPredictionsRequest { + .get_ml_predictions(authed_request(GetMlPredictionsRequest { symbol: query.symbol, model_filter: query.model, limit: query.limit, - })) + }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, @@ -64,11 +65,11 @@ async fn submit_ml_order( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .submit_ml_order(tonic::Request::new(SubmitMlOrderRequest { + .submit_ml_order(authed_request(SubmitMlOrderRequest { symbol: body.symbol, account_id: body.account_id, model_filter: body.model_filter, - })) + }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, @@ -85,9 +86,9 @@ async fn get_performance( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .get_ml_performance(tonic::Request::new(GetMlPerformanceRequest { + .get_ml_performance(authed_request(GetMlPerformanceRequest { model_filter: None, - })) + }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, diff --git a/crates/web-gateway/src/routes/performance.rs b/crates/web-gateway/src/routes/performance.rs index 11ae54238..a01aeeec9 100644 --- a/crates/web-gateway/src/routes/performance.rs +++ b/crates/web-gateway/src/routes/performance.rs @@ -7,6 +7,7 @@ use axum::{ use crate::auth::claims::Claims; use crate::error::AppError; use crate::grpc::clients::trading_client; +use crate::grpc::service_auth::authed_request; use crate::proto::trading::{GetLatencyRequest, GetMetricsRequest, GetThroughputRequest}; use crate::state::AppState; @@ -27,11 +28,11 @@ async fn get_metrics( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .get_metrics(tonic::Request::new(GetMetricsRequest { + .get_metrics(authed_request(GetMetricsRequest { metric_names: vec![], start_time_unix_nanos: None, end_time_unix_nanos: None, - })) + }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, @@ -48,12 +49,12 @@ async fn get_latency( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .get_latency(tonic::Request::new(GetLatencyRequest { + .get_latency(authed_request(GetLatencyRequest { service_name: None, operation: None, start_time_unix_nanos: None, end_time_unix_nanos: None, - })) + }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, @@ -70,12 +71,12 @@ async fn get_throughput( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .get_throughput(tonic::Request::new(GetThroughputRequest { + .get_throughput(authed_request(GetThroughputRequest { service_name: None, operation: None, start_time_unix_nanos: None, end_time_unix_nanos: None, - })) + }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, diff --git a/crates/web-gateway/src/routes/risk.rs b/crates/web-gateway/src/routes/risk.rs index 7ce99cf72..61b9c941d 100644 --- a/crates/web-gateway/src/routes/risk.rs +++ b/crates/web-gateway/src/routes/risk.rs @@ -7,6 +7,7 @@ use axum::{ use crate::auth::claims::Claims; use crate::error::AppError; use crate::grpc::clients::trading_client; +use crate::grpc::service_auth::authed_request; use crate::proto::trading::{EmergencyStopRequest, GetRiskMetricsRequest}; use crate::state::AppState; @@ -26,11 +27,11 @@ async fn get_risk_metrics( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .get_risk_metrics(tonic::Request::new(GetRiskMetricsRequest { + .get_risk_metrics(authed_request(GetRiskMetricsRequest { portfolio_id: None, start_time_unix_nanos: None, end_time_unix_nanos: None, - })) + }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, @@ -91,12 +92,12 @@ async fn emergency_stop( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .emergency_stop(tonic::Request::new(EmergencyStopRequest { + .emergency_stop(authed_request(EmergencyStopRequest { stop_type: body.stop_type, reason: body.reason, symbols: body.symbols, confirm: body.confirm, - })) + }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()).map_err(|e| AppError::Internal(e.into()))?, diff --git a/crates/web-gateway/src/routes/trading.rs b/crates/web-gateway/src/routes/trading.rs index a6ce9ca1f..f691e9443 100644 --- a/crates/web-gateway/src/routes/trading.rs +++ b/crates/web-gateway/src/routes/trading.rs @@ -7,6 +7,7 @@ use axum::{ use crate::auth::claims::Claims; use crate::error::AppError; use crate::grpc::clients::trading_client; +use crate::grpc::service_auth::authed_request; use crate::proto::trading::{ CancelOrderRequest, GetAccountInfoRequest, GetOrderStatusRequest, GetPositionsRequest, SubmitOrderRequest, @@ -32,7 +33,7 @@ async fn get_positions( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .get_positions(tonic::Request::new(GetPositionsRequest { symbol: None })) + .get_positions(authed_request(GetPositionsRequest { symbol: None }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()) @@ -51,7 +52,7 @@ async fn get_order_status( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .get_order_status(tonic::Request::new(GetOrderStatusRequest { order_id })) + .get_order_status(authed_request(GetOrderStatusRequest { order_id }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()) @@ -161,7 +162,7 @@ async fn submit_order( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .submit_order(tonic::Request::new(SubmitOrderRequest { + .submit_order(authed_request(SubmitOrderRequest { symbol: body.symbol, side: body.side, order_type: body.order_type, @@ -170,7 +171,7 @@ async fn submit_order( stop_price: body.stop_price, time_in_force: body.time_in_force.unwrap_or_default(), client_order_id: body.client_order_id.unwrap_or_default(), - })) + }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()) @@ -191,10 +192,10 @@ async fn cancel_order( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .cancel_order(tonic::Request::new(CancelOrderRequest { + .cancel_order(authed_request(CancelOrderRequest { order_id, symbol: String::new(), - })) + }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner()) @@ -212,9 +213,9 @@ async fn get_account( .ok_or_else(|| AppError::Internal(anyhow::anyhow!("Trading service not configured")))?; let mut client = trading_client(channel); let response = client - .get_account_info(tonic::Request::new(GetAccountInfoRequest { + .get_account_info(authed_request(GetAccountInfoRequest { account_id: String::new(), - })) + }, &state.config.jwt_secret)) .await?; Ok(Json( serde_json::to_value(response.into_inner())