From 966ce374cd9ecd18c11adfc474f272e4886bc210 Mon Sep 17 00:00:00 2001 From: BossX Date: Mon, 13 Jul 2026 15:27:18 +0800 Subject: [PATCH] Add JWT expiry warning and account-stream heartbeat/staleness detection Fixes #225. Token lifetime was effectively the hard cap on maker run duration with no early warning, and the account stream could sit on a half-open TCP connection forever while reporting healthy. JWT expiry: - Decode the JWT `exp` claim (best-effort, no signature check) in Credentials. `is_expired`, `remaining_seconds`, and `expires_at_string` now honour the real venue cutoff and clamp the optimistic local metadata (notably env-var tokens tracked as ~1 year). - The live maker now monitors remaining token lifetime each loop (throttled to 60s) and emits a warning risk notification through the existing notifier at <2h, escalating to critical at <15m, once per severity band. No renewal endpoint exists in the codebase, so this ships the warning + documented gap rather than a fabricated refresh. Account-stream heartbeat/staleness: - Add a client-side ping (30s) plus a read-side idle timeout (90s) to the account stream, mirroring the market websocket heartbeat pattern. Any inbound frame resets the idle deadline; on a silent half-open connection the timeout marks AccountStreamHealth unhealthy and emits a Disconnected event, so the runtime freeze/reconnect path triggers. Tests: JWT exp parsing/threshold clamping, token-expiry severity bands, and an idle-connection staleness test driving a silent WS server. Co-Authored-By: Claude Opus 4.8 --- crates/standx-cli/src/commands/maker/mod.rs | 10 +- .../standx-cli/src/commands/maker/notify.rs | 66 +++++++++++ .../standx-cli/src/commands/maker/runtime.rs | 49 ++++++++ crates/standx-sdk/src/account_stream.rs | 98 ++++++++++++++++ crates/standx-sdk/src/auth/credentials.rs | 110 +++++++++++++++++- 5 files changed, 328 insertions(+), 5 deletions(-) diff --git a/crates/standx-cli/src/commands/maker/mod.rs b/crates/standx-cli/src/commands/maker/mod.rs index db993c3..0388ff8 100644 --- a/crates/standx-cli/src/commands/maker/mod.rs +++ b/crates/standx-cli/src/commands/maker/mod.rs @@ -34,7 +34,7 @@ use model::is_order_rejection; use model::{is_maker_order, position_for_symbol, MakerExit, PendingPlace}; #[cfg(test)] use notify::webhook_body; -use notify::{MakerNotifier, PositionChange, RiskNotice}; +use notify::{token_expiry_level, MakerNotifier, PositionChange, RiskNotice, TokenExpiryLevel}; use pipeline::{CycleRequest, CycleState}; use recovery::{ cancel_maker_orders_with_retry, order_response_reconnect_available, reconcile_ledger_snapshot, @@ -71,6 +71,14 @@ pub fn panic_webhook_body(format: AlertWebhookFormat, text: &str) -> serde_json: /// Env var gating live order placement. The live path ships code-complete but /// locked until it has been supervised-tested against production. const LIVE_MAKER_ENV: &str = "STANDX_ENABLE_LIVE_MAKER"; + +/// Warn when the JWT has under 2h of life left; escalate under 15m. Token +/// lifetime caps run duration (there is no renewal endpoint), so an operator +/// needs lead time to re-authenticate before the bot halts. +const TOKEN_EXPIRY_WARN_SECS: i64 = 2 * 60 * 60; +const TOKEN_EXPIRY_CRITICAL_SECS: i64 = 15 * 60; +/// Throttle disk/env credential reloads for the expiry check. +const TOKEN_EXPIRY_CHECK_INTERVAL: Duration = Duration::from_secs(60); pub async fn handle_maker( command: MakerCommands, output_format: OutputFormat, diff --git a/crates/standx-cli/src/commands/maker/notify.rs b/crates/standx-cli/src/commands/maker/notify.rs index da67253..0ed532b 100644 --- a/crates/standx-cli/src/commands/maker/notify.rs +++ b/crates/standx-cli/src/commands/maker/notify.rs @@ -245,6 +245,35 @@ impl MakerNotifier { } } +/// Severity band for the maker's JWT remaining-lifetime monitor. +/// +/// Token lifetime is a hard cap on run duration: once the JWT expires every +/// reconnect and REST call is rejected and the bot halts. There is no renewal +/// endpoint in the codebase, so the best we can do is warn early enough for an +/// operator to re-authenticate. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub(super) enum TokenExpiryLevel { + Ok, + Warning, + Critical, +} + +/// Classify remaining token lifetime into a severity band. Thresholds are in +/// seconds; `critical_below` should be <= `warn_below`. +pub(super) fn token_expiry_level( + remaining_secs: i64, + warn_below: i64, + critical_below: i64, +) -> TokenExpiryLevel { + if remaining_secs <= critical_below { + TokenExpiryLevel::Critical + } else if remaining_secs <= warn_below { + TokenExpiryLevel::Warning + } else { + TokenExpiryLevel::Ok + } +} + pub(super) struct RiskNotice<'a> { pub(super) kind: &'a str, pub(super) severity: &'a str, @@ -295,3 +324,40 @@ mod tests { ); } } + +#[cfg(test)] +mod token_expiry_tests { + use super::*; + + #[test] + fn classifies_remaining_lifetime_into_bands() { + let warn = 2 * 60 * 60; // 2h + let critical = 15 * 60; // 15m + assert_eq!( + token_expiry_level(6 * 60 * 60, warn, critical), + TokenExpiryLevel::Ok + ); + assert_eq!( + token_expiry_level(warn, warn, critical), + TokenExpiryLevel::Warning + ); + assert_eq!( + token_expiry_level(30 * 60, warn, critical), + TokenExpiryLevel::Warning + ); + assert_eq!( + token_expiry_level(critical, warn, critical), + TokenExpiryLevel::Critical + ); + assert_eq!( + token_expiry_level(0, warn, critical), + TokenExpiryLevel::Critical + ); + } + + #[test] + fn severity_bands_are_ordered_for_escalation_checks() { + assert!(TokenExpiryLevel::Critical > TokenExpiryLevel::Warning); + assert!(TokenExpiryLevel::Warning > TokenExpiryLevel::Ok); + } +} diff --git a/crates/standx-cli/src/commands/maker/runtime.rs b/crates/standx-cli/src/commands/maker/runtime.rs index ed7a830..52180ae 100644 --- a/crates/standx-cli/src/commands/maker/runtime.rs +++ b/crates/standx-cli/src/commands/maker/runtime.rs @@ -903,11 +903,60 @@ pub(super) async fn run_maker( let mut order_response_reconnect_attempts_used = 0_u32; let mut account_stream_reconnect_attempts_used = 0_u32; let mut account_position_mismatch: Option = None; + // JWT expiry monitor: highest severity already alerted, plus a throttle so + // credentials are only reloaded from disk/env periodically. + let mut token_expiry_alerted = TokenExpiryLevel::Ok; + let mut last_token_expiry_check: Option = None; let mut runtime_state = MakerState::starting(); runtime_state.reduce(MakerEvent::StartupReady); let exit = 'main: loop { if args.live { + // JWT expiry monitor. There is no renewal endpoint, so we can only + // warn: escalate through Warning → Critical and alert once per band. + let due = last_token_expiry_check + .map(|last| last.elapsed() >= TOKEN_EXPIRY_CHECK_INTERVAL) + .unwrap_or(true); + if due { + last_token_expiry_check = Some(std::time::Instant::now()); + if let Ok(creds) = Credentials::load() { + let remaining = creds.remaining_seconds(); + let level = token_expiry_level( + remaining, + TOKEN_EXPIRY_WARN_SECS, + TOKEN_EXPIRY_CRITICAL_SECS, + ); + if level > token_expiry_alerted { + token_expiry_alerted = level; + let (severity, event) = match level { + TokenExpiryLevel::Critical => ("critical", "token_expiry_critical"), + _ => ("warning", "token_expiry_warning"), + }; + let minutes = remaining / 60; + let message = format!( + "auth token expires in ~{minutes}m ({}); no renewal endpoint — run 'standx auth login' before it lapses or the bot will halt", + creds.expires_at_string() + ); + notifier + .risk( + RiskNotice { + kind: "token_expiry", + severity, + event, + message: &message, + symbol: &symbol, + cycle, + position_before: None, + position_after: None, + expected: None, + observed: None, + }, + false, + ) + .await; + } + } + } if let Some(health) = account_stream_health .as_ref() .filter(|health| !health.is_healthy()) diff --git a/crates/standx-sdk/src/account_stream.rs b/crates/standx-sdk/src/account_stream.rs index 7f02c9e..16c62d5 100644 --- a/crates/standx-sdk/src/account_stream.rs +++ b/crates/standx-sdk/src/account_stream.rs @@ -15,6 +15,13 @@ use tokio_tungstenite::{connect_async, tungstenite::Message}; const DEFAULT_ACCOUNT_STREAM_URL: &str = "wss://perps.standx.com/ws-stream/v1"; const ACCOUNT_STREAM_ROTATE_AFTER: Duration = Duration::from_secs(23 * 60 * 60 + 50 * 60); +/// How often we send a client-side ping to keep the connection observably +/// alive and to elicit a pong (which resets the idle deadline). +const ACCOUNT_STREAM_PING_INTERVAL: Duration = Duration::from_secs(30); +/// If no inbound frame (data, ping, or pong) arrives within this window the +/// connection is treated as stale — this catches half-open TCP sessions where +/// the socket never errors but the peer has silently gone away. +const ACCOUNT_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(90); fn string_or_number<'de, D>(deserializer: D) -> std::result::Result where @@ -160,6 +167,8 @@ pub struct AccountStream { url: String, token: String, epoch: u64, + ping_interval: Duration, + idle_timeout: Duration, } impl AccountStream { @@ -175,6 +184,8 @@ impl AccountStream { url: DEFAULT_ACCOUNT_STREAM_URL.to_string(), token: credentials.token, epoch, + ping_interval: ACCOUNT_STREAM_PING_INTERVAL, + idle_timeout: ACCOUNT_STREAM_IDLE_TIMEOUT, }) } @@ -184,9 +195,18 @@ impl AccountStream { url: url.into(), token: token.into(), epoch, + ping_interval: ACCOUNT_STREAM_PING_INTERVAL, + idle_timeout: ACCOUNT_STREAM_IDLE_TIMEOUT, } } + #[cfg(test)] + fn with_heartbeat(mut self, ping_interval: Duration, idle_timeout: Duration) -> Self { + self.ping_interval = ping_interval; + self.idle_timeout = idle_timeout; + self + } + pub async fn connect( &self, channels: &[AccountChannel], @@ -264,10 +284,22 @@ impl AccountStream { let health = AccountStreamHealth::new(self.epoch); let task_health = health.clone(); let epoch = self.epoch; + let ping_interval = self.ping_interval; + let idle_timeout = self.idle_timeout; let _ = tx.send(AccountEvent::Connected { epoch }).await; let handle = tokio::spawn(async move { let rotation = tokio::time::sleep(ACCOUNT_STREAM_ROTATE_AFTER); tokio::pin!(rotation); + // First ping fires after one interval (not immediately), so the + // handshake and any buffered startup frames are handled first. + let mut ping = tokio::time::interval_at( + tokio::time::Instant::now() + ping_interval, + ping_interval, + ); + ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + // Read-side idle deadline; reset on every inbound frame. + let idle = tokio::time::sleep(idle_timeout); + tokio::pin!(idle); loop { let message = tokio::select! { _ = &mut rotation => { @@ -276,8 +308,29 @@ impl AccountStream { let _ = tx.send(AccountEvent::Disconnected { reason }).await; return; } + _ = ping.tick() => { + if let Err(error) = write.send(Message::Ping(Vec::new().into())).await { + let reason = format!("account-stream ping failed: {error}"); + task_health.mark_unhealthy(reason.clone()); + let _ = tx.send(AccountEvent::Error { reason }).await; + return; + } + continue; + } + _ = &mut idle => { + let reason = format!( + "account stream idle for {}s (no ping/pong/data; connection likely half-open)", + idle_timeout.as_secs() + ); + task_health.mark_unhealthy(reason.clone()); + let _ = tx.send(AccountEvent::Disconnected { reason }).await; + return; + } message = read.next() => message, }; + // Any inbound frame proves the peer is alive; extend the deadline. + idle.as_mut() + .reset(tokio::time::Instant::now() + idle_timeout); let Some(message) = message else { let reason = "account stream ended without a close frame".to_string(); task_health.mark_unhealthy(reason.clone()); @@ -479,4 +532,49 @@ mod tests { .unwrap(); handle.await.unwrap(); } + + #[tokio::test] + async fn idle_connection_is_marked_unhealthy() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + // Server authenticates then stays connected but silent forever, + // simulating a half-open connection that never errors or closes. + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut websocket = accept_async(socket).await.unwrap(); + let _auth = websocket.next().await.unwrap().unwrap(); + websocket + .send(Message::Text( + r#"{"seq":1,"channel":"auth","data":{"code":200,"msg":"success"}}"#.into(), + )) + .await + .unwrap(); + // Absorb the client's pings without replying, then hold the socket + // open so the client can only detect death via the idle timeout. + while let Some(Ok(_)) = websocket.next().await {} + }); + + // Idle timeout well below the ping interval so the idle deadline, not a + // ping write failure, is what trips health. + let stream = AccountStream::with_url_and_token(url, "jwt", 7) + .with_heartbeat(Duration::from_secs(10), Duration::from_millis(200)); + let (mut events, health, handle) = stream.connect(&[AccountChannel::Order]).await.unwrap(); + assert_eq!( + events.recv().await, + Some(AccountEvent::Connected { epoch: 7 }) + ); + + let disconnect = tokio::time::timeout(Duration::from_secs(2), events.recv()) + .await + .expect("idle timeout should surface a disconnect"); + match disconnect { + Some(AccountEvent::Disconnected { reason }) => { + assert!(reason.contains("idle"), "unexpected reason: {reason}"); + } + other => panic!("expected idle Disconnected, got {other:?}"), + } + assert!(!health.is_healthy()); + handle.await.unwrap(); + server.abort(); + } } diff --git a/crates/standx-sdk/src/auth/credentials.rs b/crates/standx-sdk/src/auth/credentials.rs index 4f2ca10..fa30a35 100644 --- a/crates/standx-sdk/src/auth/credentials.rs +++ b/crates/standx-sdk/src/auth/credentials.rs @@ -1,6 +1,7 @@ //! Credential management for StandX CLI use crate::error::{Error, Result}; +use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _}; use serde::{Deserialize, Serialize}; use std::path::PathBuf; @@ -8,6 +9,25 @@ use std::path::PathBuf; pub const ENV_JWT_TOKEN: &str = "STANDX_JWT"; pub const ENV_PRIVATE_KEY: &str = "STANDX_PRIVATE_KEY"; +/// Decode the `exp` (expiration) claim from a JWT without verifying its +/// signature. Returns the expiry as a Unix timestamp (seconds) when the token +/// is a well-formed JWT whose payload carries a numeric `exp`. +/// +/// This is best-effort: opaque or malformed tokens (including the fake tokens +/// used in tests) yield `None`, in which case callers fall back to the stored +/// `created_at + validity_seconds` metadata. +pub fn decode_jwt_exp(token: &str) -> Option { + // JWTs are `header.payload.signature`; the payload is the second segment, + // base64url-encoded (no padding) JSON. + let payload_b64 = token.split('.').nth(1)?; + let payload_bytes = URL_SAFE_NO_PAD.decode(payload_b64).ok()?; + let payload: serde_json::Value = serde_json::from_slice(&payload_bytes).ok()?; + match payload.get("exp")? { + serde_json::Value::Number(exp) => exp.as_i64().or_else(|| exp.as_f64().map(|v| v as i64)), + _ => None, + } +} + /// Stored credentials #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Credentials { @@ -93,22 +113,50 @@ impl Credentials { Ok(credentials) } - /// Check if token is expired + /// The JWT `exp` claim (Unix seconds), when the stored token is a + /// well-formed JWT carrying a numeric `exp`. This is the authoritative + /// expiry the venue enforces, and can be much shorter than the stored + /// `validity_seconds` metadata (e.g. an env-var token is tracked as + /// effectively non-expiring locally while the real JWT lasts ~24h). + pub fn jwt_exp(&self) -> Option { + decode_jwt_exp(&self.token) + } + + /// Check if token is expired. + /// + /// Prefers the JWT `exp` claim when decodable (the venue's real cutoff), + /// and always also honours the stored `created_at + validity_seconds` + /// metadata so opaque tokens keep their previous behaviour. pub fn is_expired(&self) -> bool { let now = chrono::Utc::now().timestamp(); + if let Some(exp) = self.jwt_exp() { + if now >= exp { + return true; + } + } now > self.created_at + self.validity_seconds } - /// Get remaining validity in seconds + /// Get remaining validity in seconds. + /// + /// When the JWT carries an `exp` claim, the returned value is clamped to + /// whichever expiry (JWT vs. stored metadata) comes first, so callers see + /// the true remaining lifetime rather than optimistic local metadata. pub fn remaining_seconds(&self) -> i64 { let now = chrono::Utc::now().timestamp(); - let expires_at = self.created_at + self.validity_seconds; + let mut expires_at = self.created_at + self.validity_seconds; + if let Some(exp) = self.jwt_exp() { + expires_at = expires_at.min(exp); + } (expires_at - now).max(0) } /// Get expiration date as string pub fn expires_at_string(&self) -> String { - let expires = self.created_at + self.validity_seconds; + let mut expires = self.created_at + self.validity_seconds; + if let Some(exp) = self.jwt_exp() { + expires = expires.min(exp); + } let datetime = chrono::DateTime::from_timestamp(expires, 0).unwrap_or_else(chrono::Utc::now); datetime.format("%Y-%m-%d %H:%M:%S UTC").to_string() @@ -467,6 +515,60 @@ mod tests { assert!(creds.is_expired()); } + /// Build a signature-less JWT (`header.payload.`) whose payload carries the + /// given claims. The signature segment is irrelevant to `exp` decoding. + fn jwt_with_payload(payload: serde_json::Value) -> String { + let header = URL_SAFE_NO_PAD.encode(br#"{"alg":"none","typ":"JWT"}"#); + let body = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&payload).unwrap()); + format!("{header}.{body}.") + } + + #[test] + fn decode_jwt_exp_reads_numeric_exp() { + let token = jwt_with_payload(serde_json::json!({ "exp": 1_800_000_000_i64, "sub": "u" })); + assert_eq!(decode_jwt_exp(&token), Some(1_800_000_000)); + } + + #[test] + fn decode_jwt_exp_ignores_opaque_and_missing() { + // Not a JWT at all. + assert_eq!(decode_jwt_exp("opaque_token"), None); + // Well-formed JWT but no exp claim. + let token = jwt_with_payload(serde_json::json!({ "sub": "u" })); + assert_eq!(decode_jwt_exp(&token), None); + // Non-numeric exp. + let token = jwt_with_payload(serde_json::json!({ "exp": "soon" })); + assert_eq!(decode_jwt_exp(&token), None); + } + + #[test] + fn jwt_exp_overrides_optimistic_metadata() { + let now = chrono::Utc::now().timestamp(); + // Metadata claims ~1 year of validity, but the JWT expires in 10 minutes. + let creds = Credentials { + token: jwt_with_payload(serde_json::json!({ "exp": now + 600 })), + private_key: String::new(), + created_at: now, + validity_seconds: 365 * 24 * 60 * 60, + }; + assert!(!creds.is_expired()); + let remaining = creds.remaining_seconds(); + assert!(remaining > 590 && remaining <= 600, "remaining={remaining}"); + } + + #[test] + fn jwt_exp_in_the_past_marks_expired() { + let now = chrono::Utc::now().timestamp(); + let creds = Credentials { + token: jwt_with_payload(serde_json::json!({ "exp": now - 1 })), + private_key: String::new(), + created_at: now, + validity_seconds: 365 * 24 * 60 * 60, + }; + assert!(creds.is_expired()); + assert_eq!(creds.remaining_seconds(), 0); + } + #[test] fn test_jwt_token_format() { // Test that token is stored as-is