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

fix(web-gateway): add service-to-service JWT auth for gRPC calls to api-gateway

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-01 00:01:53 +01:00
parent 1aef51f99b
commit 7de84bb723
9 changed files with 131 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/web-gateway/src/grpc/mod.rs
View File

@@ -1,2 +1,3 @@
pub mod clients;
pub(crate) mod service_auth;
pub mod streams;

71
crates/web-gateway/src/grpc/service_auth.rs Normal file
View File

@@ -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<String>,
permissions: Vec<String>,
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<tonic::metadata::MetadataValue<tonic::metadata::Ascii>> {
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<T>(request: &mut tonic::Request<T>, 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<T>(inner: T, jwt_secret: &str) -> tonic::Request<T> {
let mut request = tonic::Request::new(inner);
inject_auth(&mut request, jwt_secret);
request
}

34
crates/web-gateway/src/grpc/streams.rs
View File

@@ -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<Channel>,
ws_broadcast: broadcast::Sender<String>,
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<String>) {
async fn stream_market_data(channel: Channel, tx: broadcast::Sender<String>, 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<String>) {
}
/// Bridge `SubscribeOrderUpdates` gRPC stream to WebSocket broadcast
async fn stream_order_updates(channel: Channel, tx: broadcast::Sender<String>) {
async fn stream_order_updates(channel: Channel, tx: broadcast::Sender<String>, 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<String>) {
}
/// Bridge `SubscribeRiskAlerts` gRPC stream to WebSocket broadcast
async fn stream_risk_alerts(channel: Channel, tx: broadcast::Sender<String>) {
async fn stream_risk_alerts(channel: Channel, tx: broadcast::Sender<String>, 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<String>) {
}
/// Bridge `SubscribeMetrics` gRPC stream to WebSocket broadcast
async fn stream_metrics(channel: Channel, tx: broadcast::Sender<String>) {
async fn stream_metrics(channel: Channel, tx: broadcast::Sender<String>, 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();

6
crates/web-gateway/src/main.rs
View File

@@ -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<HeaderValue> = config

7
crates/web-gateway/src/routes/config.rs
View File

@@ -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()))?,

13
crates/web-gateway/src/routes/ml.rs
View File

@@ -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()))?,

13
crates/web-gateway/src/routes/performance.rs
View File

@@ -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()))?,

9
crates/web-gateway/src/routes/risk.rs
View File

@@ -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()))?,

17
crates/web-gateway/src/routes/trading.rs
View File

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