From 51dc49337393561f4235a0aac1386bdcbba357b0 Mon Sep 17 00:00:00 2001 From: Justin Bradfield Date: Tue, 28 Jul 2026 14:10:47 -0500 Subject: [PATCH 1/4] SQL-522: serve HTTP/2 on environmentd and balancerd HTTP endpoints Negotiate HTTP/2 or HTTP/1.1 via TLS ALPN and serve both protocols with hyper's auto builder (h2c via preface sniffing on plaintext listeners). HTTP/1.1-only clients are unaffected. pgwire listeners share the TLS context, but pgwire clients do not send ALPN, so the negotiation callback no-ops for them. balancerd remains a byte proxy for HTTPS: HTTP/2 frames pass through to environmentd unchanged. Its raw HTTP/1.1 502 fallback is now skipped for clients that negotiated h2, and its internal HTTP server serves h2c. Also accept WebSockets over HTTP/2 (RFC 8441 extended CONNECT): environmentd advertises SETTINGS_ENABLE_CONNECT_PROTOCOL and the /api/experimental/sql route accepts CONNECT in addition to GET. This requires axum >= 0.8.9, as 0.8.8 has a method-routing bug that registers CONNECT handlers under OPTIONS. The axum bump pulls tungstenite 0.29, which duplicates the 0.28 pinned by kube-client, so deny.toml gains skip entries for the old versions. reqwest gains the native-tls-alpn feature: without it the native-tls backend never offers ALPN and clients silently stay on HTTP/1.1, which the new tests would catch in CI. Tests: HTTP/2 over TLS with HTTP/1.1 downgrade, h2c prior knowledge, WebSocket authentication (valid and invalid credentials) over both HTTP/2 and HTTP/1.1, and h2/h1.1/h2c assertions through balancerd. Co-Authored-By: Claude Opus 4.5 --- Cargo.lock | 41 ++++- Cargo.toml | 15 +- deny.toml | 4 + src/balancerd/src/lib.rs | 87 ++++++----- src/balancerd/tests/server.rs | 33 ++++ src/environmentd/Cargo.toml | 1 + src/environmentd/src/http.rs | 24 ++- src/environmentd/tests/server.rs | 261 +++++++++++++++++++++++++++++++ src/server-core/src/lib.rs | 9 +- 9 files changed, 418 insertions(+), 57 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7c108680f1bf6..e279382788a93 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1317,9 +1317,9 @@ dependencies = [ [[package]] name = "axum" -version = "0.8.8" +version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b52af3cb4058c895d37317bb27508dccc8e5f2d39454016b297bf4a400597b8" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", "base64 0.22.1", @@ -1344,7 +1344,7 @@ dependencies = [ "sha1", "sync_wrapper", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.29.0", "tower 0.5.3", "tower-layer", "tower-service", @@ -5382,7 +5382,7 @@ dependencies = [ "serde_yaml", "thiserror 2.0.18", "tokio", - "tokio-tungstenite", + "tokio-tungstenite 0.28.0", "tokio-util", "tower 0.5.3", "tower-http", @@ -7175,6 +7175,7 @@ dependencies = [ "tokio-metrics", "tokio-postgres", "tokio-stream", + "tokio-tungstenite 0.29.0", "tower 0.5.3", "tower-http", "tower-sessions", @@ -7183,7 +7184,7 @@ dependencies = [ "tracing-capture", "tracing-opentelemetry", "tracing-subscriber", - "tungstenite", + "tungstenite 0.29.0", "url", "uuid", ] @@ -13304,7 +13305,19 @@ dependencies = [ "futures-util", "log", "tokio", - "tungstenite", + "tungstenite 0.28.0", +] + +[[package]] +name = "tokio-tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f72a05e828585856dacd553fba484c242c46e391fb0e58917c942ee9202915c" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite 0.29.0", ] [[package]] @@ -13775,6 +13788,22 @@ dependencies = [ "utf-8", ] +[[package]] +name = "tungstenite" +version = "0.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c01152af293afb9c7c2a57e4b559c5620b421f6d133261c60dd2d0cdb38e6b8" +dependencies = [ + "bytes", + "data-encoding", + "http 1.4.2", + "httparse", + "log", + "rand 0.9.4", + "sha1", + "thiserror 2.0.18", +] + [[package]] name = "turmoil" version = "0.7.2" diff --git a/Cargo.toml b/Cargo.toml index cd86e168ab06d..d8932577e750f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -308,7 +308,9 @@ aws-smithy-runtime = { version = "1.9.8", features = ["connector-hyper-0-14-x"] aws-smithy-runtime-api = "1.10.0" aws-smithy-types = { version = "1.1.8", features = ["byte-stream-poll-next"] } aws-types = "1.3.9" -axum = { version = "0.8.8", features = ["ws"] } +# 0.8.9 minimum: 0.8.8 has a method-routing bug that registers CONNECT +# handlers (needed for WebSockets over HTTP/2) under OPTIONS. +axum = { version = "0.8.9", features = ["http2", "ws"] } axum-extra = { version = "0.12.5", features = ["typed-header"] } axum-server = { version = "0.8.0", features = ["tls-rustls"] } azure_core = "0.21.0" @@ -386,12 +388,12 @@ http = "1.4.0" http-body-util = "0.1.3" httparse = "1.8.0" humantime = "2.3.0" -hyper = { version = "1.9.0", features = ["http1", "server"] } +hyper = { version = "1.9.0", features = ["client", "http1", "http2", "server"] } # hyper 0.14 is used by AWS SDK. Used to override the DNS resolver used by the SDK, # which can only be done by constructing the HTTP Client hyper-0-14 = { package = "hyper", version = "0.14", features = ["client", "tcp"] } hyper-openssl = "0.10.2" -hyper-util = "0.1.20" +hyper-util = { version = "0.1.20", features = ["server-auto", "tokio"] } tower-service = "0.3.3" iceberg = "0.9.0" iceberg-catalog-rest = "0.9.0" @@ -482,7 +484,9 @@ rdkafka = { version = "0.29.0", features = ["cmake-build", "libz-static", "ssl-v rdkafka-sys = { version = "4.3.0", features = ["cmake-build", "libz-static", "ssl-vendored", "zstd"] } regex = "1.12.3" regex-syntax = "0.8.10" -reqwest = { version = "0.12.28", features = ["blocking", "charset", "cookies", "default-tls", "http2", "json", "native-tls-vendored", "stream"] } +# `native-tls-alpn` is required for reqwest to negotiate HTTP/2 via ALPN with +# the native-tls backend; without it clients silently stay on HTTP/1.1. +reqwest = { version = "0.12.28", features = ["blocking", "charset", "cookies", "default-tls", "http2", "json", "native-tls-alpn", "native-tls-vendored", "stream"] } reqwest-middleware = { version = "0.4.2", features = ["json"] } reqwest-retry = "0.8.0" rlimit = "0.11.0" @@ -543,6 +547,7 @@ tokio-openssl = "0.6.5" tokio-postgres = "0.7.15" tokio-stream = "0.1.18" tokio-test = "0.4.5" +tokio-tungstenite = "0.29.0" tokio-util = "0.7.18" toml = "0.8.22" toml_edit = { version = "0.22.26", features = ["serde"] } @@ -559,7 +564,7 @@ tracing-capture = "0.1.0" tracing-core = "0.1.35" tracing-opentelemetry = "0.32.1" tracing-subscriber = { version = "0.3.23", features = ["env-filter", "fmt", "json", "tracing-log"] } -tungstenite = "0.28.0" +tungstenite = "0.29.0" turmoil = "0.7.2" uncased = "0.9.10" unicode-normalization = "0.1.25" diff --git a/deny.toml b/deny.toml index 9224be13166cd..edbf41c3a5328 100644 --- a/deny.toml +++ b/deny.toml @@ -159,6 +159,10 @@ skip = [ { name = "rand_core", version = "0.10.1" }, { name = "getrandom", version = "0.4.2" }, { name = "cpufeatures", version = "0.3.0" }, + # Held back by kube-client; axum 0.8.9 uses tungstenite 0.29 for + # WebSockets. + { name = "tungstenite", version = "0.28.0" }, + { name = "tokio-tungstenite", version = "0.28.0" }, ] [[bans.deny]] diff --git a/src/balancerd/src/lib.rs b/src/balancerd/src/lib.rs index 8de55ba9c8045..f215925545b8b 100644 --- a/src/balancerd/src/lib.rs +++ b/src/balancerd/src/lib.rs @@ -30,7 +30,6 @@ use anyhow::Context; use axum::response::IntoResponse; use axum::{Router, routing}; use bytes::BytesMut; -use futures::TryFutureExt; use futures::stream::BoxStream; use hickory_resolver::config::LookupIpStrategy; use hickory_resolver::lookup_ip::LookupIp; @@ -39,7 +38,7 @@ use hickory_resolver::proto::rr::{RData, RecordType}; use hickory_resolver::system_conf::read_system_conf; use hickory_resolver::{Resolver, TokioResolver}; use hyper::StatusCode; -use hyper_util::rt::TokioIo; +use hyper_util::rt::{TokioExecutor, TokioIo}; use launchdarkly_server_sdk as ld; use mz_build_info::{BuildInfo, build_info}; use mz_dyncfg::ConfigSet; @@ -472,8 +471,11 @@ impl mz_server_core::Server for InternalHttpServer { let conn = TokioIo::new(conn); Box::pin(async { - let http = hyper::server::conn::http1::Builder::new(); - http.serve_connection(conn, service).err_into().await + // Serve HTTP/1.1 or HTTP/2 (h2c via preface sniffing). + let http = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new()); + http.serve_connection(conn, service) + .await + .map_err(|e| anyhow::anyhow!(e)) }) } } @@ -1209,28 +1211,32 @@ impl mz_server_core::Server for HttpsBalancer { let active_guard = inner_metrics.active_connections(); let result: Result<_, anyhow::Error> = Box::pin(async move { let peer_addr = peer_addr.context("fetching peer addr")?; - let (mut client_stream, servername): (Box, Option) = - match tls_context { - Some(tls_context) => { - let mut ssl_stream = - SslStream::new(Ssl::new(&tls_context.get())?, conn)?; - if let Err(e) = Pin::new(&mut ssl_stream).accept().await { - let _ = ssl_stream.get_mut().shutdown().await; - return Err(e.into()); - } - let servername: Option = - ssl_stream.ssl().servername(NameType::HOST_NAME).map(|sn| { - match sn.split_once('.') { - Some((left, _right)) => left, - None => sn, - } - .into() - }); - debug!("Found sni servername: {servername:?} (https)"); - (Box::new(ssl_stream), servername) + let (mut client_stream, servername, client_h2): ( + Box, + Option, + bool, + ) = match tls_context { + Some(tls_context) => { + let mut ssl_stream = SslStream::new(Ssl::new(&tls_context.get())?, conn)?; + if let Err(e) = Pin::new(&mut ssl_stream).accept().await { + let _ = ssl_stream.get_mut().shutdown().await; + return Err(e.into()); } - _ => (Box::new(conn), None), - }; + let servername: Option = + ssl_stream.ssl().servername(NameType::HOST_NAME).map(|sn| { + match sn.split_once('.') { + Some((left, _right)) => left, + None => sn, + } + .into() + }); + debug!("Found sni servername: {servername:?} (https)"); + let client_h2 = + ssl_stream.ssl().selected_alpn_protocol() == Some(b"h2".as_slice()); + (Box::new(ssl_stream), servername, client_h2) + } + _ => (Box::new(conn), None, false), + }; let resolved = Self::resolve(&resolver, &resolve_template, port, servername.as_deref()) .await?; @@ -1242,23 +1248,28 @@ impl mz_server_core::Server for HttpsBalancer { Ok(stream) => stream, Err(e) => { error!("failed to connect to upstream server: {e}"); - let body = "upstream server not available"; // We know this is an HTTPs stream (see name // HttpsBalancer), but we actually don't care what type // of traffic it is and we only use raw tcp streams.In // order to respond with HTTP we have to write this as a - // raw http message. - let response = format!( - "HTTP/1.1 502 Bad Gateway\r\n\ - Content-Type: text/plain\r\n\ - Content-Length: {}\r\n\ - Connection: close\r\n\ - \r\n\ - {}", - body.len(), - body - ); - let _ = client_stream.write_all(response.as_bytes()).await; + // raw http message. This raw message is only + // intelligible to HTTP/1 clients, though: clients that + // negotiated HTTP/2 via ALPN just get a closed + // connection. + if !client_h2 { + let body = "upstream server not available"; + let response = format!( + "HTTP/1.1 502 Bad Gateway\r\n\ + Content-Type: text/plain\r\n\ + Content-Length: {}\r\n\ + Connection: close\r\n\ + \r\n\ + {}", + body.len(), + body + ); + let _ = client_stream.write_all(response.as_bytes()).await; + } let _ = client_stream.shutdown().await; return Ok(()); } diff --git a/src/balancerd/tests/server.rs b/src/balancerd/tests/server.rs index aab1e72f15588..56c40ea1916ed 100644 --- a/src/balancerd/tests/server.rs +++ b/src/balancerd/tests/server.rs @@ -288,6 +288,29 @@ async fn test_balancer() { let resp_x509 = X509::from_der(tlsinfo.peer_certificate().unwrap()).unwrap(); let server_x509 = X509::from_pem(&std::fs::read(&server_cert).unwrap()).unwrap(); assert_eq!(resp_x509, server_x509); + // The default client negotiates HTTP/2 with balancerd via ALPN; the + // HTTP/2 stream is byte-proxied through to environmentd. + assert_eq!(resp.version(), reqwest::Version::HTTP_2); + assert_contains!(resp.text().await.unwrap(), "12234"); + + // HTTP/1.1-only clients are still served. + let http1_client = reqwest::Client::builder() + .add_root_certificate( + reqwest::Certificate::from_pem(&ca.cert.to_pem().unwrap()).unwrap(), + ) + .pool_max_idle_per_host(0) + .http1_only() + .build() + .unwrap(); + let resp = http1_client + .post(&https_url) + .header("Content-Type", "application/json") + .basic_auth(frontegg_user, Some(&frontegg_password)) + .body(body) + .send() + .await + .unwrap(); + assert_eq!(resp.version(), reqwest::Version::HTTP_11); assert_contains!(resp.text().await.unwrap(), "12234"); // Generate new certs. Install only the key, reload, and make sure the old cert is still in @@ -406,5 +429,15 @@ async fn test_balancer() { }) .await .unwrap(); + + // The internal HTTP server serves h2c (HTTP/2 with prior knowledge) + // alongside HTTP/1.1. + let h2c_client = reqwest::Client::builder() + .http2_prior_knowledge() + .build() + .unwrap(); + let resp = h2c_client.get(&metrics_url).send().await.unwrap(); + assert_eq!(resp.version(), reqwest::Version::HTTP_2); + assert!(resp.status().is_success()); } } diff --git a/src/environmentd/Cargo.toml b/src/environmentd/Cargo.toml index 1e569b44889ee..1aebbe6459bfc 100644 --- a/src/environmentd/Cargo.toml +++ b/src/environmentd/Cargo.toml @@ -150,6 +150,7 @@ serde_urlencoded.workspace = true similar-asserts.workspace = true timely.workspace = true tokio-postgres = { workspace = true, features = ["with-chrono-0_4", "with-serde_json-1"] } +tokio-tungstenite.workspace = true [build-dependencies] anyhow.workspace = true diff --git a/src/environmentd/src/http.rs b/src/environmentd/src/http.rs index 04b3fb28252e4..b81a1e9c96fa3 100644 --- a/src/environmentd/src/http.rs +++ b/src/environmentd/src/http.rs @@ -68,7 +68,7 @@ use axum::extract::{ConnectInfo, DefaultBodyLimit, FromRequestParts, Query, Requ use axum::middleware::{self, Next}; use axum::response::{IntoResponse, Redirect, Response}; use axum::{Extension, Json, Router, routing}; -use futures::future::{Shared, TryFutureExt}; +use futures::future::Shared; use headers::authorization::{Authorization, Basic, Bearer}; use headers::{HeaderMapExt, HeaderName}; use http::header::{AUTHORIZATION, CONTENT_TYPE}; @@ -76,7 +76,7 @@ use http::uri::Scheme; use http::{HeaderMap, HeaderValue, Method, StatusCode, Uri}; use hyper_openssl::SslStream; use hyper_openssl::client::legacy::MaybeHttpsStream; -use hyper_util::rt::TokioIo; +use hyper_util::rt::{TokioExecutor, TokioIo}; use mz_adapter::session::{Session as AdapterSession, SessionConfig as AdapterSessionConfig}; use mz_adapter::{AdapterError, AdapterNotice, Client, SessionClient, WebhookAppenderCache}; use mz_adapter_types::dyncfgs::OIDC_GROUP_CLAIM; @@ -321,7 +321,12 @@ impl HttpServer { base_router = base_router.merge(base_group); let mut ws_router = Router::new() - .route("/api/experimental/sql", routing::get(sql::handle_sql_ws)) + // WebSockets arrive as a GET with the `Upgrade` header on + // HTTP/1.1 and as an extended CONNECT (RFC 8441) on HTTP/2. + .route( + "/api/experimental/sql", + routing::get(sql::handle_sql_ws).connect(sql::handle_sql_ws), + ) .with_state(WsState { frontegg, oidc_rx: oidc_rx.clone(), @@ -720,11 +725,16 @@ impl Server for HttpServer { .into_make_service_with_connect_info::(); let tower_svc = make_tower_svc.call(peer_addr).await.unwrap(); let hyper_svc = hyper::service::service_fn(|req| tower_svc.clone().call(req)); - let http = hyper::server::conn::http1::Builder::new(); - http.serve_connection(conn, hyper_svc) - .with_upgrades() - .err_into() + // Serve HTTP/1.1 or HTTP/2, detected via TLS ALPN or, on + // plaintext connections, by sniffing the HTTP/2 preface. + let mut http = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new()); + // Advertise RFC 8441 extended CONNECT so that clients can open + // WebSockets over HTTP/2. Clients that don't support it (or that + // negotiated HTTP/1.1) still use the HTTP/1.1 upgrade path. + http.http2().enable_connect_protocol(); + http.serve_connection_with_upgrades(conn, hyper_svc) .await + .map_err(|e| anyhow::anyhow!(e)) }) } } diff --git a/src/environmentd/tests/server.rs b/src/environmentd/tests/server.rs index d6828fb290372..dfe20423f757f 100644 --- a/src/environmentd/tests/server.rs +++ b/src/environmentd/tests/server.rs @@ -7583,3 +7583,264 @@ fn test_shutdown_with_inflight_writes() { tracing::info!("round {round} survived"); } } + +// Test that the HTTP server negotiates HTTP/2 over TLS via ALPN and that +// HTTP/1.1-only clients can still connect. +#[mz_ore::test(tokio::test(flavor = "multi_thread", worker_threads = 1))] +#[cfg_attr(miri, ignore)] // too slow +async fn test_http2_tls() { + let ca = Ca::new_root("test ca").unwrap(); + let (server_cert, server_key) = ca + .request_cert("server", vec![IpAddr::V4(Ipv4Addr::LOCALHOST)]) + .unwrap(); + let server = test_util::TestHarness::default() + .with_tls(server_cert, server_key) + .start() + .await; + + let https_url = Url::parse(&format!("https://{}/api/sql", server.http_local_addr())).unwrap(); + let json: serde_json::Value = serde_json::from_str(r#"{ "query": "SELECT 42;" }"#).unwrap(); + let ca_cert = reqwest::Certificate::from_pem(&ca.cert.to_pem().unwrap()).unwrap(); + + // A client that supports HTTP/2 negotiates it via ALPN. + let client = reqwest::Client::builder() + .add_root_certificate(ca_cert.clone()) + .build() + .unwrap(); + let response = client + .post(https_url.clone()) + .json(&json) + .send() + .await + .unwrap(); + assert_eq!(response.version(), reqwest::Version::HTTP_2); + assert!(response.status().is_success()); + assert_contains!(response.text().await.unwrap(), "42"); + + // An HTTP/1.1-only client is still served. + let client = reqwest::Client::builder() + .add_root_certificate(ca_cert) + .http1_only() + .build() + .unwrap(); + let response = client.post(https_url).json(&json).send().await.unwrap(); + assert_eq!(response.version(), reqwest::Version::HTTP_11); + assert!(response.status().is_success()); + assert_contains!(response.text().await.unwrap(), "42"); +} + +// Test that plaintext listeners serve both HTTP/1.1 and HTTP/2 (h2c via +// prior knowledge). +#[mz_ore::test(tokio::test(flavor = "multi_thread", worker_threads = 1))] +#[cfg_attr(miri, ignore)] // too slow +async fn test_http2_cleartext() { + let server = test_util::TestHarness::default().start().await; + + let http_url = Url::parse(&format!("http://{}/api/sql", server.http_local_addr())).unwrap(); + let json: serde_json::Value = serde_json::from_str(r#"{ "query": "SELECT 42;" }"#).unwrap(); + + // HTTP/2 with prior knowledge (h2c). + let client = reqwest::Client::builder() + .http2_prior_knowledge() + .build() + .unwrap(); + let response = client + .post(http_url.clone()) + .json(&json) + .send() + .await + .unwrap(); + assert_eq!(response.version(), reqwest::Version::HTTP_2); + assert!(response.status().is_success()); + assert_contains!(response.text().await.unwrap(), "42"); + + // Plain HTTP/1.1 is unchanged. + let response = reqwest::Client::new() + .post(http_url) + .json(&json) + .send() + .await + .unwrap(); + assert_eq!(response.version(), reqwest::Version::HTTP_11); + assert!(response.status().is_success()); + assert_contains!(response.text().await.unwrap(), "42"); +} + +// Test WebSockets over HTTP/2 (RFC 8441 extended CONNECT): authentication +// works over an HTTP/2 stream, bad credentials are rejected, and the HTTP/1.1 +// WebSocket upgrade (the downgrade path) keeps working against the same +// server. +#[mz_ore::test(tokio::test(flavor = "multi_thread", worker_threads = 1))] +#[cfg_attr(miri, ignore)] // too slow +async fn test_http2_websocket_auth() { + use futures::{SinkExt, StreamExt}; + use mz_auth::password::Password; + + type WsStream = + tokio_tungstenite::WebSocketStream>; + + // Reads messages until the initial ReadyForQuery, mirroring + // `test_util::auth_with_ws_impl` for async streams. + async fn read_until_ready(ws: &mut WsStream) -> Result, anyhow::Error> { + let mut msgs = Vec::new(); + loop { + let msg = ws + .next() + .await + .ok_or_else(|| anyhow::anyhow!("ws stream ended"))??; + match msg { + Message::Text(text) => { + let msg: WebSocketResponse = serde_json::from_str(&text).unwrap(); + match msg { + WebSocketResponse::ReadyForQuery(_) => return Ok(msgs), + msg => msgs.push(msg), + } + } + Message::Ping(_) => continue, + Message::Close(frame) => anyhow::bail!("ws closed: {frame:?}"), + other => panic!("unexpected message: {other:?}"), + } + } + } + + let server = test_util::TestHarness::default() + .with_system_parameter_default("enable_password_auth".to_string(), "true".to_string()) + .with_password_auth(Password("mz_system_password".to_owned())) + .start() + .await; + + // Opens a WebSocket over an HTTP/2 stream via extended CONNECT. + let connect_ws_http2 = || async { + let tcp = tokio::net::TcpStream::connect(server.http_local_addr()) + .await + .unwrap(); + let (mut sender, conn) = + hyper::client::conn::http2::Builder::new(hyper_util::rt::TokioExecutor::new()) + .handshake::<_, http_body_util::Empty>(hyper_util::rt::TokioIo::new( + tcp, + )) + .await + .unwrap(); + // The client can only send extended CONNECT after the server's + // SETTINGS frame advertising it has arrived, so drive the connection + // until it has been processed. + let mut conn = Box::pin(conn); + let deadline = Instant::now() + Duration::from_secs(10); + while !conn.is_extended_connect_protocol_enabled() { + assert!( + Instant::now() < deadline, + "server did not advertise RFC 8441 extended CONNECT" + ); + let _ = futures::poll!(conn.as_mut()); + tokio::task::yield_now().await; + } + task::spawn(|| "h2_ws_conn", async move { + let _ = conn.await; + }); + let req = Request::builder() + .method("CONNECT") + .extension(hyper::ext::Protocol::from_static("websocket")) + .uri("/api/experimental/sql") + .header("host", server.http_local_addr().to_string()) + .header("sec-websocket-version", "13") + .body(http_body_util::Empty::::new()) + .unwrap(); + let response = sender.send_request(req).await.unwrap(); + assert_eq!(response.version(), http::Version::HTTP_2); + assert_eq!( + response.status(), + StatusCode::OK, + "headers: {:?}", + response.headers() + ); + let upgraded = hyper::upgrade::on(response).await.unwrap(); + tokio_tungstenite::WebSocketStream::from_raw_socket( + hyper_util::rt::TokioIo::new(upgraded), + tungstenite::protocol::Role::Client, + None, + ) + .await + }; + + let auth_json = |password: &str| { + serde_json::to_string(&WebSocketAuth::Basic { + user: "mz_system".into(), + password: Password(password.to_owned()), + options: BTreeMap::default(), + }) + .unwrap() + }; + + // WebSocket over HTTP/2 with valid credentials: authenticates and runs a + // query. + let mut ws = connect_ws_http2().await; + ws.send(Message::Text(auth_json("mz_system_password").into())) + .await + .unwrap(); + read_until_ready(&mut ws).await.unwrap(); + ws.send(Message::Text(r#"{"query": "SELECT 'row42'"}"#.into())) + .await + .unwrap(); + let mut saw_row = false; + loop { + match ws.next().await.unwrap().unwrap() { + Message::Text(text) => match serde_json::from_str(&text).unwrap() { + WebSocketResponse::Row(row) => { + assert_eq!(row, vec![serde_json::json!("row42")]); + saw_row = true; + } + WebSocketResponse::ReadyForQuery(_) => break, + _ => {} + }, + Message::Ping(_) => continue, + other => panic!("unexpected message: {other:?}"), + } + } + assert!(saw_row, "expected a row from SELECT over HTTP/2 WebSocket"); + + // WebSocket over HTTP/2 with bad credentials: the server closes the + // socket without revealing detail. + let mut ws = connect_ws_http2().await; + ws.send(Message::Text(auth_json("wrong_password").into())) + .await + .unwrap(); + let err = read_until_ready(&mut ws).await.unwrap_err(); + assert_contains!(err.to_string(), "unauthorized"); + + // Downgrade: the same server still serves HTTP/1.1 WebSocket upgrades, + // with the same authentication behavior. + let (mut ws, _resp) = tungstenite::connect(server.ws_addr()).unwrap(); + test_util::auth_with_ws_impl( + &mut ws, + Message::Text(auth_json("mz_system_password").into()), + ) + .unwrap(); + ws.send(Message::Text(r#"{"query": "SELECT 'row42'"}"#.into())) + .unwrap(); + let mut saw_row = false; + loop { + match ws.read().unwrap() { + Message::Text(text) => match serde_json::from_str(&text).unwrap() { + WebSocketResponse::Row(row) => { + assert_eq!(row, vec![serde_json::json!("row42")]); + saw_row = true; + } + WebSocketResponse::ReadyForQuery(_) => break, + _ => {} + }, + Message::Ping(_) => continue, + other => panic!("unexpected message: {other:?}"), + } + } + assert!( + saw_row, + "expected a row from SELECT over HTTP/1.1 WebSocket" + ); + + // HTTP/1.1 with bad credentials is also still rejected. + let (mut ws, _resp) = tungstenite::connect(server.ws_addr()).unwrap(); + let err = + test_util::auth_with_ws_impl(&mut ws, Message::Text(auth_json("wrong_password").into())) + .unwrap_err(); + assert_contains!(format!("{err:?}"), "unauthorized"); +} diff --git a/src/server-core/src/lib.rs b/src/server-core/src/lib.rs index 68969452e4e85..09e3d0b3a6d5c 100644 --- a/src/server-core/src/lib.rs +++ b/src/server-core/src/lib.rs @@ -29,7 +29,7 @@ use mz_ore::error::ErrorExt; use mz_ore::netio::AsyncReady; use mz_ore::option::OptionExt; use mz_ore::task::JoinSetExt; -use openssl::ssl::{SslAcceptor, SslContext, SslFiletype, SslMethod}; +use openssl::ssl::{AlpnError, SslAcceptor, SslContext, SslFiletype, SslMethod, select_next_proto}; use proxy_header::{ParseConfig, ProxiedAddress, ProxyHeader}; use schemars::JsonSchema; use scopeguard::ScopeGuard; @@ -492,6 +492,13 @@ impl TlsCertConfig { // ciphers. We once tried to use the modern preset, but it was // incompatible with Fivetran, and presumably other JDBC-based tools. let mut builder = SslAcceptor::mozilla_intermediate_v5(SslMethod::tls())?; + // Negotiate HTTP/2 or HTTP/1.1 via ALPN for clients that request it. + // This context is shared with pgwire listeners, but pgwire clients do + // not send the ALPN extension, in which case `NOACK` omits ALPN from + // the handshake entirely rather than rejecting the connection. + builder.set_alpn_select_callback(|_ssl, client_protos| { + select_next_proto(b"\x02h2\x08http/1.1", client_protos).ok_or(AlpnError::NOACK) + }); builder.set_certificate_chain_file(&self.cert)?; builder.set_private_key_file(&self.key, SslFiletype::PEM)?; Ok(builder.build().into_context()) From 55b282804b156e97e026775b7f01329f1cf4c826 Mon Sep 17 00:00:00 2001 From: Justin Bradfield Date: Fri, 7 Aug 2026 16:12:28 -0500 Subject: [PATCH 2/4] SQL-522: add dyncfg to control HTTP/2 ALPN on balancerd balancerd is a byte proxy: it terminates TLS and forwards the decrypted stream to environmentd. If balancerd advertises HTTP/2 via ALPN before environmentd supports it, clients negotiate h2 but environmentd receives frames it cannot parse ("invalid HTTP version parsed (found HTTP2 preface)"). Add `balancerd_https_enable_http2_alpn` dyncfg (default false) to control whether balancerd advertises h2. During upgrade: 1. Upgrade all environmentd instances (they now support h2) 2. Enable the dyncfg via LaunchDarkly 3. Restart balancerd instances (the dyncfg is read at startup) Also add `enable_http2_alpn` parameter to `TlsCertConfig::load_context` so callers can control ALPN advertisement. environmentd passes `true` since it handles HTTP/2 directly via hyper's auto builder. Co-Authored-By: Claude Opus 4.5 --- src/balancerd/src/dyncfgs.rs | 14 ++++++++++++++ src/balancerd/src/lib.rs | 11 ++++++++--- src/environmentd/src/lib.rs | 7 ++++++- src/server-core/src/lib.rs | 23 ++++++++++++++++++----- 4 files changed, 46 insertions(+), 9 deletions(-) diff --git a/src/balancerd/src/dyncfgs.rs b/src/balancerd/src/dyncfgs.rs index 6803c3b7497ef..50c47e22bcb10 100644 --- a/src/balancerd/src/dyncfgs.rs +++ b/src/balancerd/src/dyncfgs.rs @@ -43,6 +43,19 @@ pub const INJECT_PROXY_PROTOCOL_HEADER_HTTP: Config = Config::new( "Whether to inject tcp proxy protocol headers to downstream http servers.", ); +/// Whether to advertise HTTP/2 via ALPN on the HTTPS listener. +/// +/// balancerd is a byte proxy: it terminates TLS and forwards the decrypted +/// stream to environmentd. If this is enabled before environmentd supports +/// HTTP/2, clients negotiate h2 but environmentd receives frames it cannot +/// parse. Enable only after all environmentd instances support HTTP/2. +pub const HTTPS_ENABLE_HTTP2_ALPN: Config = Config::new( + "balancerd_https_enable_http2_alpn", + false, + "Whether to advertise HTTP/2 via ALPN on the HTTPS listener. \ + Enable only after all environmentd instances support HTTP/2.", +); + /// Sets the filter to apply to stderr logging. pub const LOGGING_FILTER: Config<&str> = Config::new( "balancerd_log_filter", @@ -98,6 +111,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { .add(&SIGTERM_CONNECTION_WAIT) .add(&SIGTERM_LISTEN_WAIT) .add(&INJECT_PROXY_PROTOCOL_HEADER_HTTP) + .add(&HTTPS_ENABLE_HTTP2_ALPN) .add(&LOGGING_FILTER) .add(&OPENTELEMETRY_FILTER) .add(&LOGGING_FILTER_DEFAULTS) diff --git a/src/balancerd/src/lib.rs b/src/balancerd/src/lib.rs index f215925545b8b..5d8f102f3a6f4 100644 --- a/src/balancerd/src/lib.rs +++ b/src/balancerd/src/lib.rs @@ -76,8 +76,8 @@ use uuid::Uuid; use crate::codec::{BackendMessage, FramedConn}; use crate::dyncfgs::{ - INJECT_PROXY_PROTOCOL_HEADER_HTTP, SIGTERM_CONNECTION_WAIT, SIGTERM_LISTEN_WAIT, - has_tracing_config_update, tracing_config, + HTTPS_ENABLE_HTTP2_ALPN, INJECT_PROXY_PROTOCOL_HEADER_HTTP, SIGTERM_CONNECTION_WAIT, + SIGTERM_LISTEN_WAIT, has_tracing_config_update, tracing_config, }; /// Balancer build information. @@ -301,7 +301,12 @@ impl BalancerService { pub async fn serve(self) -> Result<(), anyhow::Error> { let (pgwire_tls, https_tls) = match &self.cfg.tls { Some(tls) => { - let context = tls.reloading_context(self.cfg.reload_certs)?; + // Controlled by dyncfg: only advertise HTTP/2 via ALPN when the + // upstream environmentd is known to support it. balancerd is a + // byte proxy, so if we advertise h2 before environmentd supports + // it, clients send h2 frames that environmentd cannot parse. + let enable_http2_alpn = HTTPS_ENABLE_HTTP2_ALPN.get(&self.configs); + let context = tls.reloading_context(self.cfg.reload_certs, enable_http2_alpn)?; ( Some(ReloadingTlsConfig { context: context.clone(), diff --git a/src/environmentd/src/lib.rs b/src/environmentd/src/lib.rs index 76a2b0e3ef7bf..484804be52a17 100644 --- a/src/environmentd/src/lib.rs +++ b/src/environmentd/src/lib.rs @@ -365,7 +365,12 @@ impl Listeners { // Validate TLS configuration, if present. let tls_reloading_context = match config.tls { - Some(tls_config) => Some(tls_config.reloading_context(config.tls_reload_certs)?), + Some(tls_config) => { + // environmentd handles HTTP/2 directly via hyper's auto builder, + // so advertise h2 via ALPN. + let enable_http2_alpn = true; + Some(tls_config.reloading_context(config.tls_reload_certs, enable_http2_alpn)?) + } None => None, }; diff --git a/src/server-core/src/lib.rs b/src/server-core/src/lib.rs index 09e3d0b3a6d5c..0e445ff8c7640 100644 --- a/src/server-core/src/lib.rs +++ b/src/server-core/src/lib.rs @@ -484,7 +484,14 @@ pub struct TlsCertConfig { impl TlsCertConfig { /// Returns the SSL context to use in TlsConfigs. - pub fn load_context(&self) -> Result { + /// + /// When `enable_http2_alpn` is true, the context advertises both HTTP/2 and + /// HTTP/1.1 via ALPN. Callers that terminate TLS but proxy the decrypted + /// bytes to an upstream server (like balancerd) should pass `false` unless + /// the upstream is known to support HTTP/2. Otherwise, clients negotiate + /// HTTP/2 with the proxy but the upstream receives h2 frames it cannot + /// parse. + pub fn load_context(&self, enable_http2_alpn: bool) -> Result { // Mozilla publishes three presets: old, intermediate, and modern. They // recommend the intermediate preset for general purpose servers, which // is what we use, as it is compatible with nearly every client released @@ -496,8 +503,13 @@ impl TlsCertConfig { // This context is shared with pgwire listeners, but pgwire clients do // not send the ALPN extension, in which case `NOACK` omits ALPN from // the handshake entirely rather than rejecting the connection. - builder.set_alpn_select_callback(|_ssl, client_protos| { - select_next_proto(b"\x02h2\x08http/1.1", client_protos).ok_or(AlpnError::NOACK) + let alpn_list: &[u8] = if enable_http2_alpn { + b"\x02h2\x08http/1.1" + } else { + b"\x08http/1.1" + }; + builder.set_alpn_select_callback(move |_ssl, client_protos| { + select_next_proto(alpn_list, client_protos).ok_or(AlpnError::NOACK) }); builder.set_certificate_chain_file(&self.cert)?; builder.set_private_key_file(&self.key, SslFiletype::PEM)?; @@ -512,13 +524,14 @@ impl TlsCertConfig { pub fn reloading_context( &self, mut ticker: ReloadTrigger, + enable_http2_alpn: bool, ) -> Result { - let context = Arc::new(RwLock::new(self.load_context()?)); + let context = Arc::new(RwLock::new(self.load_context(enable_http2_alpn)?)); let updater_context = Arc::clone(&context); let config = self.clone(); mz_ore::task::spawn(|| "TlsCertConfig reloading_context", async move { while let Some(chan) = ticker.next().await { - let result = match config.load_context() { + let result = match config.load_context(enable_http2_alpn) { Ok(ctx) => { *updater_context.write().expect("poisoned") = ctx; Ok(()) From e5f9cde03298aa15dfa555021f784b11a4dd3fe1 Mon Sep 17 00:00:00 2001 From: Justin Bradfield Date: Fri, 7 Aug 2026 16:19:25 -0500 Subject: [PATCH 3/4] SQL-522: register balancerd_https_enable_http2_alpn with test flag lints bin/lint-test-flags requires every mz_dyncfg::Config to be known to parallel-workload's FlipFlagsAction and to mzcompose's system parameter lists. balancerd dyncfgs are not SQL-settable system parameters, so add the new flag alongside the other balancerd_* entries in the uninteresting lists. Co-Authored-By: Claude Opus 5 --- misc/python/materialize/mzcompose/__init__.py | 1 + misc/python/materialize/parallel_workload/action.py | 1 + 2 files changed, 2 insertions(+) diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index f622f03de3c8d..2604d1dac7418 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -665,6 +665,7 @@ def get_default_system_parameters( "balancerd_sigterm_connection_wait", "balancerd_sigterm_listen_wait", "balancerd_inject_proxy_protocol_header_http", + "balancerd_https_enable_http2_alpn", "balancerd_log_filter", "balancerd_opentelemetry_filter", "balancerd_log_filter_defaults", diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 1ef968a1bd9de..6ca7d8dfb8aa0 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -2042,6 +2042,7 @@ def __init__( "balancerd_sigterm_connection_wait", "balancerd_sigterm_listen_wait", "balancerd_inject_proxy_protocol_header_http", + "balancerd_https_enable_http2_alpn", "balancerd_log_filter", "balancerd_opentelemetry_filter", "balancerd_log_filter_defaults", From 780cf8b1f25d51b3eeaab17113b0b8af08ee0677 Mon Sep 17 00:00:00 2001 From: Justin Bradfield Date: Fri, 7 Aug 2026 21:09:11 -0500 Subject: [PATCH 4/4] SQL-522: enable balancerd HTTP/2 ALPN in tests The new balancerd_https_enable_http2_alpn dyncfg defaults off, which broke the balancerd server test's assertion that a request through balancerd comes back as HTTP/2. Follow the convention that a new flag defaults off in production but on in the test configuration: teach set_defaults to accept the flag so it can be set via --default-config, and have the test enable it. Also document that the flag is read once when the TLS context is built, so a change only takes effect after balancerd restarts. Co-Authored-By: Claude Opus 5 --- src/balancerd/src/dyncfgs.rs | 20 ++++++++++++++++---- src/balancerd/tests/server.rs | 8 +++++++- 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/src/balancerd/src/dyncfgs.rs b/src/balancerd/src/dyncfgs.rs index 50c47e22bcb10..61dc2a8b64196 100644 --- a/src/balancerd/src/dyncfgs.rs +++ b/src/balancerd/src/dyncfgs.rs @@ -46,14 +46,21 @@ pub const INJECT_PROXY_PROTOCOL_HEADER_HTTP: Config = Config::new( /// Whether to advertise HTTP/2 via ALPN on the HTTPS listener. /// /// balancerd is a byte proxy: it terminates TLS and forwards the decrypted -/// stream to environmentd. If this is enabled before environmentd supports -/// HTTP/2, clients negotiate h2 but environmentd receives frames it cannot -/// parse. Enable only after all environmentd instances support HTTP/2. +/// stream to environmentd. ALPN is answered during the client handshake, +/// before balancerd has connected upstream, so it cannot discover whether +/// environmentd speaks HTTP/2 in time. Enabling this while environmentd is +/// still HTTP/1.1-only makes clients negotiate h2 and send frames environmentd +/// rejects with "invalid HTTP version parsed (found HTTP2 preface)". Enable +/// only after every environmentd instance supports HTTP/2. +/// +/// NOTE: read once when the TLS context is built at startup, so a change only +/// takes effect after balancerd restarts. pub const HTTPS_ENABLE_HTTP2_ALPN: Config = Config::new( "balancerd_https_enable_http2_alpn", false, "Whether to advertise HTTP/2 via ALPN on the HTTPS listener. \ - Enable only after all environmentd instances support HTTP/2.", + Enable only after all environmentd instances support HTTP/2. \ + Takes effect on balancerd restart.", ); /// Sets the filter to apply to stderr logging. @@ -138,6 +145,11 @@ pub(crate) fn set_defaults( INJECT_PROXY_PROTOCOL_HEADER_HTTP.name(), mz_dyncfg::ConfigVal::Bool(bool::from_str(v)?), ) + } else if k.as_str() == HTTPS_ENABLE_HTTP2_ALPN.name() { + config_updates.add_dynamic( + HTTPS_ENABLE_HTTP2_ALPN.name(), + mz_dyncfg::ConfigVal::Bool(bool::from_str(v)?), + ) } else { return Err(anyhow!("Invalid default config value {k}")); } diff --git a/src/balancerd/tests/server.rs b/src/balancerd/tests/server.rs index 56c40ea1916ed..79b16f30ad67c 100644 --- a/src/balancerd/tests/server.rs +++ b/src/balancerd/tests/server.rs @@ -201,7 +201,13 @@ async fn test_balancer() { None, None, TracingHandle::disabled(), - vec![], + // Advertise HTTP/2 via ALPN. This defaults off in production so a + // balancerd that upgrades ahead of environmentd does not offer h2 + // to clients before environmentd can parse it. + vec![( + "balancerd_https_enable_http2_alpn".to_string(), + "true".to_string(), + )], ); let balancer_server = BalancerService::new(balancer_cfg).await.unwrap(); let balancer_pgwire_listen = balancer_server.pgwire.0.local_addr();