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 ffa280b7bab..02c5d86c5f3 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; +/// 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. #[derive(Debug, Serialize, Deserialize, PartialEq, Clone, Copy, ToSchema)] #[serde(rename_all = "snake_case")] 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 e8819d78361..5d066b49e80 100644 --- a/crates/pipeline-manager/src/runner/interaction.rs +++ b/crates/pipeline-manager/src/runner/interaction.rs @@ -6,18 +6,19 @@ 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::{self, 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; 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; @@ -29,6 +30,54 @@ 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 [`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. +/// +/// 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(header::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()) +} + +/// 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, @@ -369,12 +418,15 @@ 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_ws_subprotocol(&mut res, &client_request); + let mut client_rx = client_rx.max_frame_size(MAX_WS_FRAME_SIZE); // Connect to the pipeline @@ -758,4 +810,102 @@ 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![(header::SEC_WEBSOCKET_PROTOCOL.as_str(), 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. + #[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. + #[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!( + 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. + #[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(), + 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. + #[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 6d83ec4d233..4a31885d49a 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://') +// 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)}`)