From 29f4b02d3bc81069e8ce7103945f857ca95d7b86 Mon Sep 17 00:00:00 2001 From: oxoxDev <164490987+oxoxDev@users.noreply.github.com> Date: Wed, 13 May 2026 08:27:18 +0530 Subject: [PATCH] fix(socket): follow HTTP 3xx during WebSocket handshake (OPENHUMAN-TAURI-9X) (#1547) Co-authored-by: Claude Opus 4.7 (1M context) Co-authored-by: Steven Enamakel --- src/openhuman/socket/event_handlers.rs | 1 + src/openhuman/socket/manager.rs | 23 ++- src/openhuman/socket/ws_loop.rs | 188 ++++++++++++++++++- src/openhuman/socket/ws_loop_tests.rs | 246 +++++++++++++++++++++++++ 4 files changed, 449 insertions(+), 9 deletions(-) diff --git a/src/openhuman/socket/event_handlers.rs b/src/openhuman/socket/event_handlers.rs index 9768df26e..e2833b8f4 100644 --- a/src/openhuman/socket/event_handlers.rs +++ b/src/openhuman/socket/event_handlers.rs @@ -244,6 +244,7 @@ mod tests { webhook_router: RwLock::new(None), status: RwLock::new(ConnectionStatus::Disconnected), socket_id: RwLock::new(None), + error: RwLock::new(None), }) } diff --git a/src/openhuman/socket/manager.rs b/src/openhuman/socket/manager.rs index 0959bfe41..a530f7861 100644 --- a/src/openhuman/socket/manager.rs +++ b/src/openhuman/socket/manager.rs @@ -51,6 +51,10 @@ pub(super) struct SharedState { pub(super) status: RwLock, /// Socket ID assigned by the server. pub(super) socket_id: RwLock>, + /// Last user-visible connection warning surfaced through `SocketState.error` + /// (e.g. "backend redirected ws→wss; update BACKEND_URL"). Cleared on every + /// successful handshake and on disconnect. + pub(super) error: RwLock>, } // --------------------------------------------------------------------------- @@ -82,6 +86,7 @@ impl SocketManager { webhook_router: RwLock::new(None), status: RwLock::new(ConnectionStatus::Disconnected), socket_id: RwLock::new(None), + error: RwLock::new(None), }), emit_tx: tokio::sync::Mutex::new(None), shutdown_tx: tokio::sync::Mutex::new(None), @@ -105,7 +110,7 @@ impl SocketManager { SocketState { status: *self.shared.status.read(), socket_id: self.shared.socket_id.read().clone(), - error: None, + error: self.shared.error.read().clone(), } } @@ -133,6 +138,7 @@ impl SocketManager { log::info!("[socket] Connecting to {}", url); *self.shared.status.write() = ConnectionStatus::Connecting; + *self.shared.error.write() = None; emit_state_change(&self.shared); let (emit_tx, emit_rx) = mpsc::unbounded_channel::(); @@ -165,6 +171,7 @@ impl SocketManager { } *self.shared.status.write() = ConnectionStatus::Disconnected; *self.shared.socket_id.write() = None; + *self.shared.error.write() = None; emit_state_change(&self.shared); log::debug!("[socket] Disconnected"); Ok(()) @@ -247,6 +254,18 @@ mod tests { assert_eq!(state.socket_id.as_deref(), Some("sid-abc")); } + #[test] + fn get_state_surfaces_stored_error_to_callers() { + let mgr = SocketManager::new(); + *mgr.shared.error.write() = + Some("backend redirected ws→wss; update BACKEND_URL".to_string()); + let state = mgr.get_state(); + assert_eq!( + state.error.as_deref(), + Some("backend redirected ws→wss; update BACKEND_URL") + ); + } + #[tokio::test] async fn emit_without_connection_errors_without_panic() { let mgr = SocketManager::new(); @@ -269,6 +288,7 @@ mod tests { webhook_router: RwLock::new(None), status: RwLock::new(ConnectionStatus::Connecting), socket_id: RwLock::new(None), + error: RwLock::new(None), }; // Must not panic even with all default state. emit_state_change(&shared); @@ -280,6 +300,7 @@ mod tests { webhook_router: RwLock::new(None), status: RwLock::new(ConnectionStatus::Connected), socket_id: RwLock::new(Some("x".into())), + error: RwLock::new(None), }; // Pure logging — must not touch state or panic. emit_server_event(&shared, "any.event", json!({})); diff --git a/src/openhuman/socket/ws_loop.rs b/src/openhuman/socket/ws_loop.rs index cc6c33bdf..9c6a6eef3 100644 --- a/src/openhuman/socket/ws_loop.rs +++ b/src/openhuman/socket/ws_loop.rs @@ -6,7 +6,10 @@ use futures_util::{SinkExt, StreamExt}; use serde_json::json; use tokio::sync::{mpsc, watch}; use tokio::time::{Duration, Instant}; -use tokio_tungstenite::{connect_async, tungstenite::Message as WsMessage}; +use tokio_tungstenite::{ + connect_async, + tungstenite::{http::StatusCode, Error as WsError, Message as WsMessage}, +}; use crate::api::models::socket::ConnectionStatus; @@ -14,6 +17,14 @@ use super::event_handlers::{handle_sio_event, parse_sio_event}; use super::manager::{emit_state_change, SharedState}; use super::types::{ConnectionOutcome, WsStream}; +/// Maximum HTTP redirect hops to follow during a single WebSocket connect attempt. +/// +/// Cloudflare and similar edges return a single 301 (e.g. when the configured +/// `BACKEND_URL` is `http://...` and the server only serves the upgrade over TLS) +/// before the upgrade succeeds. Three hops is enough headroom for chained +/// redirects while still bounding pathological loops. +const MAX_REDIRECT_HOPS: u8 = 3; + // --------------------------------------------------------------------------- // Background loop // --------------------------------------------------------------------------- @@ -30,6 +41,13 @@ pub(super) async fn ws_loop( let mut backoff = Duration::from_millis(1000); let max_backoff = Duration::from_secs(30); + // `ws_url` is the *resolved* socket URL we're currently connecting to. + // If the backend responds with an HTTP 3xx during the upgrade (typical when + // BACKEND_URL is configured as `http://` and the edge forces TLS), we + // follow the Location header and pin the resolved URL here so subsequent + // reconnects skip the redirect round-trip entirely. + let mut ws_url = crate::api::socket::websocket_url(&url); + loop { if *shutdown_rx.borrow() { break; @@ -40,7 +58,7 @@ pub(super) async fn ws_loop( emit_state_change(&shared); let outcome = run_connection( - &url, + &mut ws_url, &token, &shared, &mut emit_rx, @@ -93,21 +111,27 @@ pub(super) async fn ws_loop( // --------------------------------------------------------------------------- /// Run a single WebSocket connection through handshake and event loop. +/// +/// `ws_url` is taken by mutable reference so that any HTTP redirect we follow +/// during the upgrade (see `connect_with_redirects`) is pinned for the next +/// reconnect attempt — we don't want to re-hit the redirect every time the +/// loop backs off and retries. async fn run_connection( - url: &str, + ws_url: &mut String, token: &str, shared: &Arc, emit_rx: &mut mpsc::UnboundedReceiver, shutdown_rx: &mut watch::Receiver, internal_tx: &mpsc::UnboundedSender, ) -> ConnectionOutcome { - // 1. Build WebSocket URL (appends /socket.io/?EIO=4&transport=websocket) - let ws_url = crate::api::socket::websocket_url(url); log::info!("[socket] WS URL: {}", ws_url); - // 2. Connect via WebSocket (uses rustls TLS for wss://) - let (ws_stream, _response) = match connect_async(&ws_url).await { - Ok(r) => r, + // 2. Connect via WebSocket (uses rustls TLS for wss://). Follow HTTP 3xx + // redirects up to MAX_REDIRECT_HOPS so a `http://` config behind a + // Cloudflare-style edge that 301s to `https://` connects cleanly + // instead of looping at error-level forever. + let ws_stream = match connect_with_redirects(ws_url, shared).await { + Ok(stream) => stream, Err(e) => return ConnectionOutcome::Failed(format!("WebSocket connect: {e}")), }; @@ -404,6 +428,154 @@ fn handle_sio_packet( } } +// --------------------------------------------------------------------------- +// Redirect-following connect +// --------------------------------------------------------------------------- + +/// Connect to `ws_url`, following HTTP 3xx redirects up to `MAX_REDIRECT_HOPS`. +/// +/// Plain `connect_async` returns an error on any non-`101 Switching Protocols` +/// response, so a Cloudflare-style `http://… → https://…` 301 (which happens +/// whenever `BACKEND_URL` is configured without TLS) used to be fatal — the +/// reconnect loop would hammer the same dead URL forever at error level. +/// +/// On each redirect we: +/// 1. resolve the `Location` header against the current URL (handles relative +/// Location values), +/// 2. upgrade the scheme so the next attempt is still a WebSocket +/// (`http` → `ws`, `https` → `wss`; `ws`/`wss` pass through), +/// 3. mutate `ws_url` in place so the redirect target is pinned for +/// subsequent reconnects (no need to re-hit the redirect every retry), +/// 4. record a one-shot warning in `SharedState.error` the first time we +/// follow a redirect so the UI can surface "your `BACKEND_URL` is stale". +/// +/// On non-redirect failures the original error is returned and the caller +/// counts it toward the exponential backoff like before. +async fn connect_with_redirects( + ws_url: &mut String, + shared: &Arc, +) -> Result { + let original = ws_url.clone(); + for hop in 0..=MAX_REDIRECT_HOPS { + match connect_async(ws_url.as_str()).await { + Ok((stream, _response)) => return Ok(stream), + Err(WsError::Http(response)) if is_redirect_status(response.status()) => { + if hop == MAX_REDIRECT_HOPS { + log::error!( + "[socket] Exceeded {MAX_REDIRECT_HOPS} redirect hops starting from {original}; giving up" + ); + return Err(WsError::Http(response)); + } + let location = match extract_location_header(&response) { + Some(loc) => loc, + None => { + log::error!( + "[socket] Redirect {} from {ws_url} missing Location header", + response.status() + ); + return Err(WsError::Http(response)); + } + }; + let next_url = match resolve_redirect_target(ws_url, &location) { + Ok(url) => url, + Err(e) => { + log::error!( + "[socket] Cannot follow redirect to {location} from {ws_url}: {e}" + ); + return Err(WsError::Http(response)); + } + }; + log::warn!( + "[socket] Server redirected ({}) {} → {}", + response.status(), + ws_url, + next_url + ); + // Only persist a stale-BACKEND_URL warning for permanent + // redirects (301 / 308). Temporary redirects (302 / 307) say + // "this time, go elsewhere" — the configured BACKEND_URL is + // still correct, and surfacing a "please update config" hint + // for a transient hop would be misleading. Per CodeRabbit + // review on PR #1547. + if matches!( + response.status(), + StatusCode::MOVED_PERMANENTLY | StatusCode::PERMANENT_REDIRECT + ) { + record_redirect_warning(shared, &original, &next_url); + } + *ws_url = next_url; + } + Err(e) => return Err(e), + } + } + // Unreachable: the loop either returns Ok, returns the redirect error after + // exhausting hops, or returns a non-redirect Err. + unreachable!("connect_with_redirects exited loop without returning") +} + +/// Statuses we treat as "follow the Location and retry". +/// +/// 308 (Permanent Redirect) and 307 (Temporary Redirect) explicitly preserve +/// the method; 301/302 historically do too for upgrade requests in practice. +/// Anything else (300, 304, ...) stays an error. +fn is_redirect_status(status: StatusCode) -> bool { + matches!( + status, + StatusCode::MOVED_PERMANENTLY + | StatusCode::FOUND + | StatusCode::TEMPORARY_REDIRECT + | StatusCode::PERMANENT_REDIRECT + ) +} + +fn extract_location_header( + response: &tokio_tungstenite::tungstenite::http::Response>>, +) -> Option { + response + .headers() + .get(tokio_tungstenite::tungstenite::http::header::LOCATION) + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_string()) +} + +/// Resolve `location` against `current_ws_url` and rewrite the scheme so the +/// result is still a valid WebSocket URL. +/// +/// `location` may be absolute (`https://host/path?q=1`) or relative +/// (`/socket.io/?EIO=4`). We use the `url` crate's relative-URL parser to do +/// the join the same way browsers do, then map `http`→`ws` / `https`→`wss`. +fn resolve_redirect_target(current_ws_url: &str, location: &str) -> Result { + let base = url::Url::parse(current_ws_url).map_err(|e| format!("invalid current URL: {e}"))?; + let resolved = base + .join(location) + .map_err(|e| format!("invalid Location {location:?}: {e}"))?; + + let upgraded_scheme = match resolved.scheme() { + "http" => "ws", + "https" => "wss", + "ws" | "wss" => resolved.scheme(), + other => return Err(format!("unsupported scheme in Location: {other}")), + }; + + let mut next = resolved.clone(); + next.set_scheme(upgraded_scheme) + .map_err(|_| format!("failed to set scheme {upgraded_scheme} on {resolved}"))?; + Ok(next.to_string()) +} + +/// Persist a one-shot, user-visible warning that the backend redirected the +/// configured socket URL. Subsequent redirects in the same connect attempt +/// don't overwrite — the first hop carries the actionable signal. +fn record_redirect_warning(shared: &Arc, original: &str, resolved: &str) { + let mut slot = shared.error.write(); + if slot.is_some() { + return; + } + *slot = Some(format!( + "Backend redirected {original} → {resolved}. Update BACKEND_URL to the resolved URL to avoid the extra hop." + )); +} + #[cfg(test)] #[path = "ws_loop_tests.rs"] mod tests; diff --git a/src/openhuman/socket/ws_loop_tests.rs b/src/openhuman/socket/ws_loop_tests.rs index 4fd6d1af3..19b008401 100644 --- a/src/openhuman/socket/ws_loop_tests.rs +++ b/src/openhuman/socket/ws_loop_tests.rs @@ -1,14 +1,125 @@ use super::*; use parking_lot::RwLock; +use tokio_tungstenite::tungstenite::http::{header::LOCATION, Response, StatusCode}; fn make_shared() -> Arc { Arc::new(SharedState { webhook_router: RwLock::new(None), status: RwLock::new(ConnectionStatus::Connected), socket_id: RwLock::new(None), + error: RwLock::new(None), }) } +// ── Redirect resolution (the real fix for OPENHUMAN-TAURI-9X) ── + +#[test] +fn resolve_redirect_upgrades_http_to_ws_for_absolute_location() { + // Cloudflare's exact behaviour: ws://host/path → 301 Location: https://host:443/path. + // We must rewrite https→wss so connect_async sees a WebSocket URL. + let next = resolve_redirect_target( + "ws://api.tinyhumans.ai/socket.io/?EIO=4&transport=websocket", + "https://api.tinyhumans.ai:443/socket.io/?EIO=4&transport=websocket", + ) + .expect("scheme upgrade"); + assert!( + next.starts_with("wss://api.tinyhumans.ai"), + "expected wss:// after upgrade, got {next}" + ); + assert!(next.contains("/socket.io/?EIO=4&transport=websocket")); +} + +#[test] +fn resolve_redirect_handles_relative_location_against_current_url() { + // RFC 7230 allows a relative Location — must be resolved against the + // request URL, not treated as an error. + let next = resolve_redirect_target( + "ws://api.example.com/socket.io/?EIO=4&transport=websocket", + "/v2/socket.io/?EIO=4&transport=websocket", + ) + .expect("relative resolve"); + assert_eq!( + next, + "ws://api.example.com/v2/socket.io/?EIO=4&transport=websocket" + ); +} + +#[test] +fn resolve_redirect_preserves_ws_and_wss_schemes_verbatim() { + let next = resolve_redirect_target("wss://a.example/socket.io/", "wss://b.example/socket.io/") + .unwrap(); + assert!(next.starts_with("wss://b.example")); +} + +#[test] +fn resolve_redirect_rejects_unsupported_scheme() { + let err = resolve_redirect_target("wss://a.example/socket.io/", "ftp://elsewhere/socket.io/") + .unwrap_err(); + assert!(err.contains("ftp"), "{err}"); +} + +#[test] +fn redirect_status_matches_only_followable_codes() { + assert!(is_redirect_status(StatusCode::MOVED_PERMANENTLY)); + assert!(is_redirect_status(StatusCode::FOUND)); + assert!(is_redirect_status(StatusCode::TEMPORARY_REDIRECT)); + assert!(is_redirect_status(StatusCode::PERMANENT_REDIRECT)); + // 304 / 300 / 4xx / 5xx all stay errors that the backoff loop handles. + assert!(!is_redirect_status(StatusCode::NOT_MODIFIED)); + assert!(!is_redirect_status(StatusCode::MULTIPLE_CHOICES)); + assert!(!is_redirect_status(StatusCode::BAD_REQUEST)); + assert!(!is_redirect_status(StatusCode::BAD_GATEWAY)); +} + +#[test] +fn extract_location_header_returns_value_when_present() { + let resp = Response::builder() + .status(StatusCode::MOVED_PERMANENTLY) + .header(LOCATION, "https://api.example.com/socket.io/") + .body(None) + .unwrap(); + assert_eq!( + extract_location_header(&resp).as_deref(), + Some("https://api.example.com/socket.io/") + ); +} + +#[test] +fn extract_location_header_returns_none_when_missing() { + let resp = Response::builder() + .status(StatusCode::MOVED_PERMANENTLY) + .body(None) + .unwrap(); + assert!(extract_location_header(&resp).is_none()); +} + +#[test] +fn redirect_warning_is_recorded_once_and_pinned_to_first_hop() { + // First call records original→resolved. Second call (a second hop in the + // same attempt) must NOT overwrite — the first warning carries the + // user-actionable signal (your configured BACKEND_URL is stale). + let shared = make_shared(); + record_redirect_warning( + &shared, + "ws://api.example.com/socket.io/", + "wss://api.example.com/socket.io/", + ); + let first = shared.error.read().clone().unwrap(); + assert!(first.contains("ws://api.example.com")); + assert!(first.contains("wss://api.example.com")); + + record_redirect_warning( + &shared, + "wss://api.example.com/socket.io/", + "wss://api.example.com/v2/socket.io/", + ); + let after_second = shared.error.read().clone().unwrap(); + assert_eq!( + after_second, first, + "second redirect must not overwrite the first warning" + ); +} + // ── handle_eio_message ───────────────────────────────────────── #[test] @@ -383,3 +494,138 @@ fn connect_behavior_variants_are_distinct() { ConnectBehavior::GarbageOpenPacket => {} } } + +// ── End-to-end redirect-follow (the real fix for the 301 noise) ── + +use tokio::io::{AsyncReadExt, AsyncWriteExt}; + +/// Spawn a one-shot HTTP/1.1 server that replies with a 301 redirect to +/// `location` and closes — used to prove that `connect_with_redirects` +/// follows the redirect end-to-end through `connect_async` instead of +/// surfacing the 301 as a recurring error. +async fn spawn_mock_301_redirect(location: String) -> std::net::SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr"); + tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept"); + // Drain the incoming upgrade request so the client doesn't see RST. + let mut buf = [0u8; 4096]; + let _ = stream.read(&mut buf).await; + let response = format!( + "HTTP/1.1 301 Moved Permanently\r\n\ + Location: {location}\r\n\ + Content-Length: 0\r\n\ + Connection: close\r\n\r\n" + ); + let _ = stream.write_all(response.as_bytes()).await; + let _ = stream.shutdown().await; + }); + addr +} + +/// Driver-level proof: when the configured URL responds with a 301 pointing +/// at a working Engine.IO server, `ws_loop` follows the redirect, completes +/// the handshake, and records a one-shot warning in `SharedState.error` so +/// the UI can surface the stale-config signal. +#[tokio::test] +async fn ws_loop_follows_301_to_working_backend() { + // 1. Real EIO server on `ws://127.0.0.1:PORT`. + let (fwd_tx, mut fwd_rx) = mpsc::unbounded_channel::(); + let real_addr = spawn_mock_eio_server(ConnectBehavior::Ack, fwd_tx).await; + let real_ws_url = format!("ws://{real_addr}/socket.io/?EIO=4&transport=websocket"); + + // 2. Redirect server that 301s every request to the real EIO server. + let redirect_addr = spawn_mock_301_redirect(real_ws_url.clone()).await; + + // 3. Drive `ws_loop` with the redirect address as the base URL. This is + // exactly the production failure mode: BACKEND_URL points at a host + // that 301s the WebSocket upgrade. + let shared = make_shared(); + *shared.status.write() = ConnectionStatus::Disconnected; + let (emit_tx, emit_rx) = mpsc::unbounded_channel::(); + let (shutdown_tx, shutdown_rx) = watch::channel(false); + let internal_tx = emit_tx.clone(); + drop(emit_tx); + + let loop_shared = Arc::clone(&shared); + let handle = tokio::spawn(async move { + ws_loop( + format!("http://{redirect_addr}"), + "redirect-test-token".into(), + loop_shared, + emit_rx, + shutdown_rx, + internal_tx, + ) + .await; + }); + + // The SIO CONNECT frame arriving on the *real* server proves the redirect + // was followed and the WebSocket handshake completed against the redirect + // target — not the redirect host. + let deadline = tokio::time::Instant::now() + tokio::time::Duration::from_secs(5); + let mut saw_connect = false; + while tokio::time::Instant::now() < deadline { + if let Ok(Some(frame)) = + tokio::time::timeout(tokio::time::Duration::from_millis(200), fwd_rx.recv()).await + { + if frame.starts_with("40") && frame.contains("redirect-test-token") { + saw_connect = true; + break; + } + } + } + assert!( + saw_connect, + "redirect was not followed — SIO CONNECT never reached the real EIO server" + ); + + for _ in 0..50 { + if *shared.status.read() == ConnectionStatus::Connected { + break; + } + tokio::time::sleep(tokio::time::Duration::from_millis(50)).await; + } + assert_eq!(*shared.status.read(), ConnectionStatus::Connected); + + let warning = shared.error.read().clone(); + assert!( + warning + .as_deref() + .map(|w| w.contains("redirected") && w.contains("BACKEND_URL")) + .unwrap_or(false), + "expected redirect warning in SharedState.error, got {warning:?}" + ); + + let _ = shutdown_tx.send(true); + let _ = tokio::time::timeout(tokio::time::Duration::from_secs(5), handle).await; +} + +/// 301 without a Location header is unrecoverable — must surface as a real +/// error and not loop forever attempting to follow nothing. +#[tokio::test] +async fn connect_with_redirects_fails_when_location_missing() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr"); + tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept"); + let mut buf = [0u8; 4096]; + let _ = stream.read(&mut buf).await; + // 301 but no Location header. + let _ = stream + .write_all( + b"HTTP/1.1 301 Moved Permanently\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", + ) + .await; + let _ = stream.shutdown().await; + }); + + let shared = make_shared(); + let mut url = format!("ws://{addr}/socket.io/?EIO=4&transport=websocket"); + let err = connect_with_redirects(&mut url, &shared) + .await + .expect_err("must surface failure when Location is absent"); + assert!(matches!(err, WsError::Http(_))); + // No warning recorded because the redirect was never actually followed. + assert!(shared.error.read().is_none()); +}