From 9a82ec1ed36dbdd3b1efe7491b549a1ec79e23cd Mon Sep 17 00:00:00 2001 From: Karakatiza666 Date: Thu, 23 Jul 2026 12:49:34 +0000 Subject: [PATCH 1/3] [pipeline-manager] Fix ad-hoc not working in Chromium-based browsers by confirming the requested subprotocol Signed-off-by: Karakatiza666 --- crates/feldera-types/src/query.rs | 7 ++ .../src/runner/interaction.rs | 92 ++++++++++++++++++- .../src/lib/services/pipelineManager.ts | 6 +- 3 files changed, 102 insertions(+), 3 deletions(-) diff --git a/crates/feldera-types/src/query.rs b/crates/feldera-types/src/query.rs index ffa280b7bab..50e3627697b 100644 --- a/crates/feldera-types/src/query.rs +++ b/crates/feldera-types/src/query.rs @@ -5,6 +5,13 @@ use utoipa::ToSchema; /// The maximum size of a WebSocket frames we're sending in bytes. pub const MAX_WS_FRAME_SIZE: usize = 1024 * 1024 * 2; +/// Non-secret WebSocket subprotocol the web console offers on every browser +/// WebSocket handshake. Browsers require the server to echo one offered +/// subprotocol, so this gives the manager a value to echo back that is not the +/// bearer token. It marks the handshake, not the endpoint (the URL path already +/// identifies that), so it is shared across all WebSocket endpoints. +pub const WS_SUBPROTOCOL: &str = "feldera-ws-v1"; + /// URL-encoded `format` argument to the `/query` endpoint. #[derive(Debug, Serialize, Deserialize, PartialEq, Clone, Copy, ToSchema)] #[serde(rename_all = "snake_case")] diff --git a/crates/pipeline-manager/src/runner/interaction.rs b/crates/pipeline-manager/src/runner/interaction.rs index e8819d78361..2cfb7f2fe09 100644 --- a/crates/pipeline-manager/src/runner/interaction.rs +++ b/crates/pipeline-manager/src/runner/interaction.rs @@ -6,13 +6,14 @@ use crate::db::types::pipeline::ExtendedPipelineDescrMonitoring; use crate::db::types::tenant::TenantId; use crate::error::ManagerError; use crate::runner::error::RunnerError; +use actix_web::http::header::{HeaderName, HeaderValue}; use actix_web::{http::Method, web::Payload, HttpRequest, HttpResponse, HttpResponseBuilder}; use actix_ws::{CloseCode, CloseReason}; use awc::error::{ConnectError, SendRequestError}; use awc::{ClientRequest, ClientResponse}; use crossbeam::sync::ShardedLock; use feldera_observability::AwcRequestTracingExt; -use feldera_types::query::MAX_WS_FRAME_SIZE; +use feldera_types::query::{MAX_WS_FRAME_SIZE, WS_SUBPROTOCOL}; use std::fmt::Display; use std::{collections::HashMap, sync::Arc, time::Duration}; use tokio::sync::Mutex; @@ -29,6 +30,31 @@ use feldera_types::runtime_status::RuntimeStatus; /// a large circuit profile. const RESPONSE_SIZE_LIMIT: usize = 50 * 1024 * 1024; +/// Pick the subprotocol to echo back on a WebSocket handshake. +/// +/// Returns `None` when the client offered none (non-browser clients such as +/// `fda` authenticate via headers and offer no subprotocol, so nothing is +/// echoed). Otherwise prefers the non-secret [`WS_SUBPROTOCOL`] sentinel and +/// falls back to the first offered protocol. +/// +/// The web console carries auth as `feldera-bearer.*`/`feldera-tenant.*` +/// subprotocols, so prefer the non-secret [`WS_SUBPROTOCOL`]. +fn select_ws_subprotocol(request: &HttpRequest) -> Option { + let offered: Vec<&str> = request + .headers() + .get("Sec-WebSocket-Protocol")? + .to_str() + .ok()? + .split(',') + .map(str::trim) + .filter(|protocol| !protocol.is_empty()) + .collect(); + if offered.contains(&WS_SUBPROTOCOL) { + return Some(WS_SUBPROTOCOL.to_string()); + } + offered.first().map(|protocol| protocol.to_string()) +} + pub(crate) struct CachedPipelineDescr { pipeline: ExtendedPipelineDescrMonitoring, instantiated: Instant, @@ -369,12 +395,26 @@ impl RunnerInteraction { let (location, _cache_hit) = self.check_pipeline(tenant_id, pipeline_name).await?; // Handle client request - let (res, mut client_tx, client_rx) = actix_ws::handle(&client_request, client_body) + let (mut res, mut client_tx, client_rx) = actix_ws::handle(&client_request, client_body) .map_err(|e| ManagerError::ApiError { api_error: ApiError::UnableToConnect { reason: format!("Unable to initiate websocket connection with client: {e}"), }, })?; + + // Echo back one of the client's offered subprotocols. Browsers that + // follow the WHATWG "establish a WebSocket connection" algorithm + // (Chromium-based) fail the connection with close code 1006 when they offered + // subprotocols but the 101 response echoes none. + if let Some(selected) = select_ws_subprotocol(&client_request) { + if let Ok(value) = HeaderValue::from_str(&selected) { + res.headers_mut().insert( + HeaderName::from_static("sec-websocket-protocol"), + value, + ); + } + } + let mut client_rx = client_rx.max_frame_size(MAX_WS_FRAME_SIZE); // Connect to the pipeline @@ -758,4 +798,52 @@ mod tests { assert_eq!(body, plain); // Dropping mock_server verifies the expect(0) assertion. } + + fn selected_protocol(subprotocols: Option<&str>) -> Option { + let headers: Vec<(&str, &str)> = subprotocols + .map(|value| vec![("Sec-WebSocket-Protocol", value)]) + .unwrap_or_default(); + let (req, _payload) = test_get_request(&headers); + select_ws_subprotocol(&req) + } + + /// A non-browser client (e.g. `fda`) offers no subprotocol, so the manager + /// echoes none: `Sec-WebSocket-Protocol` stays absent from the 101. + #[actix_web::test] + async fn ws_subprotocol_absent_when_none_offered() { + assert_eq!(selected_protocol(None), None); + } + + /// The web console offers the sentinel first; it is echoed back so Chromium + /// accepts the handshake instead of closing with code 1006. + #[actix_web::test] + async fn ws_subprotocol_prefers_sentinel() { + let offered = format!("{WS_SUBPROTOCOL}, feldera-bearer.QUJD, feldera-tenant.dA"); + // Guard against accidentally echoing the bearer token subprotocol. + assert_eq!( + selected_protocol(Some(&offered)).as_deref(), + Some(WS_SUBPROTOCOL) + ); + } + + /// The sentinel is honored regardless of the position it is offered in, + /// since browsers only require the echoed value to be one that was offered. + #[actix_web::test] + async fn ws_subprotocol_finds_sentinel_out_of_order() { + let offered = format!("feldera-bearer.QUJD, {WS_SUBPROTOCOL}"); + assert_eq!( + selected_protocol(Some(&offered)).as_deref(), + Some(WS_SUBPROTOCOL) + ); + } + + /// An older console that offers only the auth subprotocols still gets an + /// echo (the first offered), so the handshake succeeds rather than 1006. + #[actix_web::test] + async fn ws_subprotocol_falls_back_to_first_offered() { + assert_eq!( + selected_protocol(Some("feldera-bearer.QUJD, feldera-tenant.dA")).as_deref(), + Some("feldera-bearer.QUJD") + ); + } } diff --git a/js-packages/web-console/src/lib/services/pipelineManager.ts b/js-packages/web-console/src/lib/services/pipelineManager.ts index 6d83ec4d233..b6b316bf5bf 100644 --- a/js-packages/web-console/src/lib/services/pipelineManager.ts +++ b/js-packages/web-console/src/lib/services/pipelineManager.ts @@ -965,6 +965,10 @@ export const pipelineLogsStream = async ( const httpToWs = (endpoint: string) => endpoint.replace(/^http(s?):\/\//, 'ws$1://') +// Non-secret subprotocol the manager echoes back on every browser WebSocket +// handshake. Keep in sync with `WS_SUBPROTOCOL` in crates/feldera-types. +const WS_SUBPROTOCOL = 'feldera-ws-v1' + // base64url (no padding) — the only encoding whose alphabet is a valid // WebSocket subprotocol token, so it can carry the bearer token/tenant. const base64UrlEncode = (value: string): string => @@ -1011,7 +1015,7 @@ export const adHocQuery = async (pipelineName: string, query: string) => { const url = `${httpToWs(felderaEndpoint)}/v0/pipelines/${pipelineName}/query` const authHeaders = await getAuthorizationHeaders() - const protocols: string[] = [] + const protocols = [WS_SUBPROTOCOL] const token = authHeaders['Authorization']?.replace(/^Bearer /, '') if (token) { protocols.push(`feldera-bearer.${base64UrlEncode(token)}`) From 5987b8b019d9dc826c62a7a538500f94d10cea92 Mon Sep 17 00:00:00 2001 From: feldera-bot Date: Thu, 23 Jul 2026 12:59:54 +0000 Subject: [PATCH 2/3] [ci] apply automatic fixes Signed-off-by: feldera-bot --- crates/pipeline-manager/src/runner/interaction.rs | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/crates/pipeline-manager/src/runner/interaction.rs b/crates/pipeline-manager/src/runner/interaction.rs index 2cfb7f2fe09..69009d2c5da 100644 --- a/crates/pipeline-manager/src/runner/interaction.rs +++ b/crates/pipeline-manager/src/runner/interaction.rs @@ -36,7 +36,7 @@ const RESPONSE_SIZE_LIMIT: usize = 50 * 1024 * 1024; /// `fda` authenticate via headers and offer no subprotocol, so nothing is /// echoed). Otherwise prefers the non-secret [`WS_SUBPROTOCOL`] sentinel and /// falls back to the first offered protocol. -/// +/// /// The web console carries auth as `feldera-bearer.*`/`feldera-tenant.*` /// subprotocols, so prefer the non-secret [`WS_SUBPROTOCOL`]. fn select_ws_subprotocol(request: &HttpRequest) -> Option { @@ -408,10 +408,8 @@ impl RunnerInteraction { // subprotocols but the 101 response echoes none. if let Some(selected) = select_ws_subprotocol(&client_request) { if let Ok(value) = HeaderValue::from_str(&selected) { - res.headers_mut().insert( - HeaderName::from_static("sec-websocket-protocol"), - value, - ); + res.headers_mut() + .insert(HeaderName::from_static("sec-websocket-protocol"), value); } } From 1a163d8ff87d256937888f01fb37e23e6d6911e9 Mon Sep 17 00:00:00 2001 From: Karakatiza666 Date: Thu, 23 Jul 2026 14:32:18 +0000 Subject: [PATCH 3/3] review fixes Signed-off-by: Karakatiza666 --- Cargo.lock | 1 + crates/feldera-types/src/query.rs | 10 +- crates/pipeline-manager/Cargo.toml | 1 + .../src/runner/interaction.rs | 116 ++++++++++++++---- .../src/lib/services/pipelineManager.ts | 4 +- 5 files changed, 99 insertions(+), 33 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c5482cf5e74..327e0cc6127 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8851,6 +8851,7 @@ dependencies = [ "actix-cors", "actix-files", "actix-http", + "actix-test", "actix-web", "actix-web-httpauth", "actix-web-static-files", diff --git a/crates/feldera-types/src/query.rs b/crates/feldera-types/src/query.rs index 50e3627697b..02c5d86c5f3 100644 --- a/crates/feldera-types/src/query.rs +++ b/crates/feldera-types/src/query.rs @@ -5,11 +5,11 @@ use utoipa::ToSchema; /// The maximum size of a WebSocket frames we're sending in bytes. pub const MAX_WS_FRAME_SIZE: usize = 1024 * 1024 * 2; -/// Non-secret WebSocket subprotocol the web console offers on every browser -/// WebSocket handshake. Browsers require the server to echo one offered -/// subprotocol, so this gives the manager a value to echo back that is not the -/// bearer token. It marks the handshake, not the endpoint (the URL path already -/// identifies that), so it is shared across all WebSocket endpoints. +/// WebSocket subprotocol the web console offers on every browser WebSocket +/// handshake. Browsers require the server to echo one of the offered +/// subprotocols, so this gives the manager a stable value to echo back. It +/// marks the handshake, not the endpoint (the URL path already identifies +/// that), so it is shared across all WebSocket endpoints. pub const WS_SUBPROTOCOL: &str = "feldera-ws-v1"; /// URL-encoded `format` argument to the `/query` endpoint. diff --git a/crates/pipeline-manager/Cargo.toml b/crates/pipeline-manager/Cargo.toml index 5b280650274..71fb96d8ba9 100644 --- a/crates/pipeline-manager/Cargo.toml +++ b/crates/pipeline-manager/Cargo.toml @@ -145,6 +145,7 @@ proptest-derive = { workspace = true } pg-client-config = { workspace = true } base64 = { workspace = true } actix-http = { workspace = true } +actix-test = { workspace = true } serial_test = { workspace = true } wiremock = { workspace = true } feldera-types = { workspace = true, features = ["testing"] } diff --git a/crates/pipeline-manager/src/runner/interaction.rs b/crates/pipeline-manager/src/runner/interaction.rs index 69009d2c5da..5d066b49e80 100644 --- a/crates/pipeline-manager/src/runner/interaction.rs +++ b/crates/pipeline-manager/src/runner/interaction.rs @@ -6,7 +6,7 @@ use crate::db::types::pipeline::ExtendedPipelineDescrMonitoring; use crate::db::types::tenant::TenantId; use crate::error::ManagerError; use crate::runner::error::RunnerError; -use actix_web::http::header::{HeaderName, HeaderValue}; +use actix_web::http::header::{self, HeaderValue}; use actix_web::{http::Method, web::Payload, HttpRequest, HttpResponse, HttpResponseBuilder}; use actix_ws::{CloseCode, CloseReason}; use awc::error::{ConnectError, SendRequestError}; @@ -18,7 +18,7 @@ use std::fmt::Display; use std::{collections::HashMap, sync::Arc, time::Duration}; use tokio::sync::Mutex; use tokio::time::Instant; -use tracing::{error, info}; +use tracing::{error, info, warn}; use crate::db::listen_table::PIPELINE_NOTIFY_CHANNEL_CAPACITY; use crate::db::types::resources_status::ResourcesStatus; @@ -34,15 +34,17 @@ const RESPONSE_SIZE_LIMIT: usize = 50 * 1024 * 1024; /// /// Returns `None` when the client offered none (non-browser clients such as /// `fda` authenticate via headers and offer no subprotocol, so nothing is -/// echoed). Otherwise prefers the non-secret [`WS_SUBPROTOCOL`] sentinel and -/// falls back to the first offered protocol. +/// echoed). Otherwise prefers the [`WS_SUBPROTOCOL`] sentinel and falls back to +/// the first offered protocol, so an older console that offers only the +/// `feldera-bearer.*`/`feldera-tenant.*` auth subprotocols still gets an echo. /// -/// The web console carries auth as `feldera-bearer.*`/`feldera-tenant.*` -/// subprotocols, so prefer the non-secret [`WS_SUBPROTOCOL`]. +/// Subprotocol tokens are matched case-sensitively (RFC 6455): the console and +/// manager share the [`WS_SUBPROTOCOL`] constant verbatim, so no normalization +/// is needed, and the bearer/tenant tokens are case-sensitive base64url anyway. fn select_ws_subprotocol(request: &HttpRequest) -> Option { let offered: Vec<&str> = request .headers() - .get("Sec-WebSocket-Protocol")? + .get(header::SEC_WEBSOCKET_PROTOCOL)? .to_str() .ok()? .split(',') @@ -55,6 +57,27 @@ fn select_ws_subprotocol(request: &HttpRequest) -> Option { offered.first().map(|protocol| protocol.to_string()) } +/// Echo one of the client's offered subprotocols onto a WebSocket handshake +/// response (see [`select_ws_subprotocol`] for the choice). +fn echo_ws_subprotocol(response: &mut HttpResponse, request: &HttpRequest) { + let Some(selected) = select_ws_subprotocol(request) else { + return; + }; + match HeaderValue::from_str(&selected) { + // `actix_ws::handle` never sets `Sec-WebSocket-Protocol`, so insert + // (replace) rather than append: there is no prior value. + Ok(value) => { + response + .headers_mut() + .insert(header::SEC_WEBSOCKET_PROTOCOL, value); + } + // Unreachable in practice: a subprotocol token is ASCII-visible and was + // already parsed once by actix on the way in. Warn (and skip the echo) + // so a broken invariant is discoverable rather than a silent 1006. + Err(e) => warn!("failed to echo websocket subprotocol {selected:?}: {e}"), + } +} + pub(crate) struct CachedPipelineDescr { pipeline: ExtendedPipelineDescrMonitoring, instantiated: Instant, @@ -402,16 +425,7 @@ impl RunnerInteraction { }, })?; - // Echo back one of the client's offered subprotocols. Browsers that - // follow the WHATWG "establish a WebSocket connection" algorithm - // (Chromium-based) fail the connection with close code 1006 when they offered - // subprotocols but the 101 response echoes none. - if let Some(selected) = select_ws_subprotocol(&client_request) { - if let Ok(value) = HeaderValue::from_str(&selected) { - res.headers_mut() - .insert(HeaderName::from_static("sec-websocket-protocol"), value); - } - } + echo_ws_subprotocol(&mut res, &client_request); let mut client_rx = client_rx.max_frame_size(MAX_WS_FRAME_SIZE); @@ -799,7 +813,7 @@ mod tests { fn selected_protocol(subprotocols: Option<&str>) -> Option { let headers: Vec<(&str, &str)> = subprotocols - .map(|value| vec![("Sec-WebSocket-Protocol", value)]) + .map(|value| vec![(header::SEC_WEBSOCKET_PROTOCOL.as_str(), value)]) .unwrap_or_default(); let (req, _payload) = test_get_request(&headers); select_ws_subprotocol(&req) @@ -807,15 +821,15 @@ mod tests { /// A non-browser client (e.g. `fda`) offers no subprotocol, so the manager /// echoes none: `Sec-WebSocket-Protocol` stays absent from the 101. - #[actix_web::test] - async fn ws_subprotocol_absent_when_none_offered() { + #[test] + fn ws_subprotocol_absent_when_none_offered() { assert_eq!(selected_protocol(None), None); } /// The web console offers the sentinel first; it is echoed back so Chromium /// accepts the handshake instead of closing with code 1006. - #[actix_web::test] - async fn ws_subprotocol_prefers_sentinel() { + #[test] + fn ws_subprotocol_prefers_sentinel() { let offered = format!("{WS_SUBPROTOCOL}, feldera-bearer.QUJD, feldera-tenant.dA"); // Guard against accidentally echoing the bearer token subprotocol. assert_eq!( @@ -826,8 +840,8 @@ mod tests { /// The sentinel is honored regardless of the position it is offered in, /// since browsers only require the echoed value to be one that was offered. - #[actix_web::test] - async fn ws_subprotocol_finds_sentinel_out_of_order() { + #[test] + fn ws_subprotocol_finds_sentinel_out_of_order() { let offered = format!("feldera-bearer.QUJD, {WS_SUBPROTOCOL}"); assert_eq!( selected_protocol(Some(&offered)).as_deref(), @@ -837,11 +851,61 @@ mod tests { /// An older console that offers only the auth subprotocols still gets an /// echo (the first offered), so the handshake succeeds rather than 1006. - #[actix_web::test] - async fn ws_subprotocol_falls_back_to_first_offered() { + #[test] + fn ws_subprotocol_falls_back_to_first_offered() { assert_eq!( selected_protocol(Some("feldera-bearer.QUJD, feldera-tenant.dA")).as_deref(), Some("feldera-bearer.QUJD") ); } + + /// End-to-end handshake through a real actix server: a client offering + /// subprotocols gets the sentinel echoed on the 101 (so Chromium keeps the + /// connection), while a client offering none gets no header back. + #[actix_web::test] + async fn ws_handshake_echoes_offered_subprotocol() { + use actix_web::{web, App}; + setup(); // awc's TLS connector needs a rustls CryptoProvider installed. + + // Mirrors the production handshake: complete the upgrade, then echo. + async fn handler( + req: HttpRequest, + body: web::Payload, + ) -> Result { + let (mut res, _session, _stream) = actix_ws::handle(&req, body)?; + echo_ws_subprotocol(&mut res, &req); + Ok(res) + } + + let srv = actix_test::start(|| App::new().route("/query", web::get().to(handler))); + + let (with_offer, _framed) = awc::Client::new() + .ws(srv.url("/query")) + .protocols([WS_SUBPROTOCOL, "feldera-bearer.QUJD"]) + .connect() + .await + .expect("handshake with offered subprotocols must succeed"); + assert_eq!( + with_offer + .headers() + .get(header::SEC_WEBSOCKET_PROTOCOL) + .expect("101 must echo a subprotocol"), + WS_SUBPROTOCOL, + ); + + let (without_offer, _framed) = awc::Client::new() + .ws(srv.url("/query")) + .connect() + .await + .expect("handshake without subprotocols must succeed"); + assert!( + without_offer + .headers() + .get(header::SEC_WEBSOCKET_PROTOCOL) + .is_none(), + "no subprotocol offered should leave the response header absent" + ); + + srv.stop().await; + } } diff --git a/js-packages/web-console/src/lib/services/pipelineManager.ts b/js-packages/web-console/src/lib/services/pipelineManager.ts index b6b316bf5bf..4a31885d49a 100644 --- a/js-packages/web-console/src/lib/services/pipelineManager.ts +++ b/js-packages/web-console/src/lib/services/pipelineManager.ts @@ -965,8 +965,8 @@ export const pipelineLogsStream = async ( const httpToWs = (endpoint: string) => endpoint.replace(/^http(s?):\/\//, 'ws$1://') -// Non-secret subprotocol the manager echoes back on every browser WebSocket -// handshake. Keep in sync with `WS_SUBPROTOCOL` in crates/feldera-types. +// Subprotocol the manager echoes back on every browser WebSocket handshake. +// Keep in sync with `WS_SUBPROTOCOL` in crates/feldera-types. const WS_SUBPROTOCOL = 'feldera-ws-v1' // base64url (no padding) — the only encoding whose alphabet is a valid