v0.32.2

src/upstream/client.rs

@@ -173,12 +173,7 @@ impl RotatingClient {
         }
     }
 
-    fn rotate(
-        &self,
-        observed_generation: u64,
-        reason: &'static str,
-        triggering_request: bool,
-    ) {
+    fn rotate(&self, observed_generation: u64, reason: &'static str, triggering_request: bool) {
         if self.generation.load(Ordering::Acquire) != observed_generation
             || self
                 .rotating
@@ -272,9 +267,12 @@ fn build_raw_client(factory: &ClientFactory) -> Result<RawProxyClient> {
     let keepalive = &factory.keepalive;
     let timeout = factory.timeout;
     let is_https = url::Url::parse(&backend.url)?.scheme() == "https";
+    // WebSocket HTTP/1.1 Upgrade retains its isolated H1 client. Preserve
+    // the 0.32.2 `http2` meaning: HTTPS allows H2/H1 via ALPN, while
+    // plaintext uses H2C. Only the NEW `h2` value is strict on HTTPS.
     let protocol = if websocket {
         UpstreamProtocol::Http1
-    } else if backend.protocol == UpstreamProtocol::Http2 && is_https {
+    } else if is_https && backend.protocol == UpstreamProtocol::Http2 {
         UpstreamProtocol::Auto
     } else {
         backend.protocol
@@ -326,9 +324,20 @@ fn build_raw_client(factory: &ClientFactory) -> Result<RawProxyClient> {
     let builder = HttpsConnectorBuilder::new()
         .with_tls_config(config)
         .https_or_http();
+    // TLS Auto offers h2 then HTTP/1.1 in a single handshake. Negotiating
+    // HTTP/1.1 here happens *before* any request is sent: there is no unsafe
+    // automatic replay after a timed-out or partially delivered POST/PATCH.
+    // Cleartext Auto uses H1; H2C requires the explicit Http2 mode.
+    let protocol = if !is_https && protocol == UpstreamProtocol::Auto {
+        UpstreamProtocol::Http1
+    } else {
+        protocol
+    };
     let inner = match protocol {
         UpstreamProtocol::Http1 => builder.enable_http1().wrap_connector(http),
-        UpstreamProtocol::Http2 => builder.enable_http2().wrap_connector(http),
+        UpstreamProtocol::Http2 | UpstreamProtocol::H2 => {
+            builder.enable_http2().wrap_connector(http)
+        }
         UpstreamProtocol::Auto => builder.enable_http1().enable_http2().wrap_connector(http),
     };
     let mut builder = Client::builder(TokioExecutor::new());
@@ -343,7 +352,10 @@ fn build_raw_client(factory: &ClientFactory) -> Result<RawProxyClient> {
             0
         })
         .retry_canceled_requests(false)
-        .http2_only(protocol == UpstreamProtocol::Http2)
+        .http2_only(matches!(
+            protocol,
+            UpstreamProtocol::Http2 | UpstreamProtocol::H2
+        ))
         .timer(TokioTimer::new())
         .http2_keep_alive_interval(
             keepalive

src/upstream/protocol_tests.rs

@@ -0,0 +1,223 @@
+// Real Hyper + Rustls upstream peers, independent of the Python mock.
+async fn tls_origin(
+    offer_h2: bool,
+) -> (
+    std::net::SocketAddr,
+    tokio::sync::oneshot::Receiver<(Option<Vec<u8>>, Version)>,
+    tokio::task::JoinHandle<()>,
+) {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let pair = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();
+    let key = rustls::pki_types::PrivatePkcs8KeyDer::from(pair.signing_key.serialize_der());
+    let provider = Arc::new(rustls::crypto::ring::default_provider());
+    let mut config = rustls::ServerConfig::builder_with_provider(provider)
+        .with_safe_default_protocol_versions()
+        .unwrap()
+        .with_no_client_auth()
+        .with_single_cert(vec![pair.cert.der().clone()], key.into())
+        .unwrap();
+    config.alpn_protocols = if offer_h2 {
+        vec![b"h2".to_vec(), b"http/1.1".to_vec()]
+    } else {
+        vec![b"http/1.1".to_vec()]
+    };
+    let acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(config));
+    let (observed_tx, observed_rx) = tokio::sync::oneshot::channel();
+    let server = tokio::spawn(async move {
+        let (tcp, _) = listener.accept().await.unwrap();
+        // For an H2-only client and H1-only server, negotiation can fail
+        // before any HTTP request is delivered. That is the desired error.
+        let Ok(tls) = acceptor.accept(tcp).await else {
+            return;
+        };
+        let alpn = tls.get_ref().1.alpn_protocol().map(|value| value.to_vec());
+        let observed = Arc::new(std::sync::Mutex::new(Some(observed_tx)));
+        let negotiated_h2 = alpn.as_deref() == Some(b"h2".as_slice());
+        let service = service_fn(move |request: Request<hyper::body::Incoming>| {
+            let observed = observed.clone();
+            let alpn = alpn.clone();
+            async move {
+                if let Some(sender) = observed.lock().unwrap().take() {
+                    let _ = sender.send((alpn, request.version()));
+                }
+                Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"upstream-ok"))))
+            }
+        });
+        if negotiated_h2 {
+            let _ = hyper::server::conn::http2::Builder::new(TokioExecutor::new())
+                .serve_connection(TokioIo::new(tls), service)
+                .await;
+        } else {
+            let _ = hyper::server::conn::http1::Builder::new()
+                .serve_connection(TokioIo::new(tls), service)
+                .await;
+        }
+    });
+    (address, observed_rx, server)
+}
+
+fn protocol_backend(address: std::net::SocketAddr, scheme: &str, mode: &str) -> Backend {
+    serde_json::from_value(serde_json::json!({
+        "id": "protocol-probe",
+        "url": format!("{scheme}://{address}"),
+        "protocol": mode,
+        "allow_private": true
+    }))
+    .unwrap()
+}
+
+async fn protocol_request(
+    backend: &Backend,
+    websocket: bool,
+) -> Result<Response<hyper::body::Incoming>, hyper_util::client::legacy::Error> {
+    let client = client(backend, AddressPolicy::new(backend).unwrap(), websocket).unwrap();
+    let request = Request::builder()
+        .uri(format!("{}/probe", backend.url))
+        .body(Body::empty())
+        .unwrap();
+    tokio::time::timeout(Duration::from_secs(5), client.request(request))
+        .await
+        .expect("TLS/HTTP handshake exceeded test deadline")
+}
+
+#[tokio::test]
+async fn https_auto_chooses_h2_by_alpn_without_replaying_request() {
+    let (address, observed, server) = tls_origin(true).await;
+    let backend = protocol_backend(address, "https", "auto");
+    let response = protocol_request(&backend, false).await.unwrap();
+    assert_eq!(response.version(), Version::HTTP_2);
+    assert_eq!(
+        response.into_body().collect().await.unwrap().to_bytes(),
+        Bytes::from_static(b"upstream-ok")
+    );
+    let (alpn, version) = observed.await.unwrap();
+    assert_eq!(alpn.as_deref(), Some(b"h2".as_slice()));
+    assert_eq!(version, Version::HTTP_2);
+    server.abort();
+}
+
+#[tokio::test]
+async fn https_auto_negotiates_http1_before_sending_to_h1_only_peer() {
+    let (address, observed, server) = tls_origin(false).await;
+    let backend = protocol_backend(address, "https", "auto");
+    let response = protocol_request(&backend, false).await.unwrap();
+    assert_eq!(response.version(), Version::HTTP_11);
+    let (alpn, version) = observed.await.unwrap();
+    assert_eq!(alpn.as_deref(), Some(b"http/1.1".as_slice()));
+    assert_eq!(version, Version::HTTP_11);
+    server.abort();
+}
+
+#[tokio::test]
+async fn explicit_https_h2_requires_h2_and_refuses_h1_only_peer() {
+    let (address, _observed, server) = tls_origin(false).await;
+    let backend = protocol_backend(address, "https", "h2");
+    assert!(protocol_request(&backend, false).await.is_err());
+    server.abort();
+}
+
+#[tokio::test]
+async fn explicit_https_h2_selects_h2_via_alpn() {
+    let (address, observed, server) = tls_origin(true).await;
+    let backend = protocol_backend(address, "https", "h2");
+    let response = protocol_request(&backend, false).await.unwrap();
+    assert_eq!(response.version(), Version::HTTP_2);
+    let (alpn, version) = observed.await.unwrap();
+    assert_eq!(alpn.as_deref(), Some(b"h2".as_slice()));
+    assert_eq!(version, Version::HTTP_2);
+    server.abort();
+}
+
+#[tokio::test]
+async fn saved_legacy_https_http2_keeps_h1_fallback() {
+    let (address, observed, server) = tls_origin(false).await;
+    let backend = protocol_backend(address, "https", "http2");
+    let response = protocol_request(&backend, false).await.unwrap();
+    assert_eq!(response.version(), Version::HTTP_11);
+    let (alpn, version) = observed.await.unwrap();
+    assert_eq!(alpn.as_deref(), Some(b"http/1.1".as_slice()));
+    assert_eq!(version, Version::HTTP_11);
+    server.abort();
+}
+
+#[tokio::test]
+async fn saved_legacy_https_http2_still_prefers_h2() {
+    let (address, observed, server) = tls_origin(true).await;
+    let backend = protocol_backend(address, "https", "http2");
+    let response = protocol_request(&backend, false).await.unwrap();
+    assert_eq!(response.version(), Version::HTTP_2);
+    let (alpn, version) = observed.await.unwrap();
+    assert_eq!(alpn.as_deref(), Some(b"h2".as_slice()));
+    assert_eq!(version, Version::HTTP_2);
+    server.abort();
+}
+
+#[tokio::test]
+async fn explicit_https_http1_does_not_offer_h2_to_h2_capable_peer() {
+    let (address, observed, server) = tls_origin(true).await;
+    let backend = protocol_backend(address, "https", "http1");
+    let response = protocol_request(&backend, false).await.unwrap();
+    assert_eq!(response.version(), Version::HTTP_11);
+    let (alpn, version) = observed.await.unwrap();
+    assert_ne!(alpn.as_deref(), Some(b"h2".as_slice()));
+    assert_eq!(version, Version::HTTP_11);
+    server.abort();
+}
+
+#[tokio::test]
+async fn websocket_isolated_client_stays_on_http1_even_with_h2_backend() {
+    let (address, observed, server) = tls_origin(true).await;
+    let backend = protocol_backend(address, "https", "http2");
+    let response = protocol_request(&backend, true).await.unwrap();
+    assert_eq!(response.version(), Version::HTTP_11);
+    let (alpn, version) = observed.await.unwrap();
+    assert_ne!(alpn.as_deref(), Some(b"h2".as_slice()));
+    assert_eq!(version, Version::HTTP_11);
+    server.abort();
+}
+
+#[tokio::test]
+async fn cleartext_auto_does_not_attempt_unsafe_h2c_probe() {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let server = tokio::spawn(async move {
+        let (tcp, _) = listener.accept().await.unwrap();
+        let service = service_fn(|request: Request<hyper::body::Incoming>| async move {
+            assert_eq!(request.version(), Version::HTTP_11);
+            Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"h1"))))
+        });
+        let _ = hyper::server::conn::http1::Builder::new()
+            .serve_connection(TokioIo::new(tcp), service)
+            .await;
+    });
+    let backend = protocol_backend(address, "http", "auto");
+    let response = protocol_request(&backend, false).await.unwrap();
+    assert_eq!(response.version(), Version::HTTP_11);
+    server.abort();
+}
+
+#[test]
+fn default_protocol_and_serialization_are_backward_compatible() {
+    let backend: Backend = serde_json::from_value(serde_json::json!({
+        "id": "legacy", "url": "http://example.org", "allow_private": false
+    }))
+    .unwrap();
+    assert_eq!(backend.protocol, crate::config::UpstreamProtocol::Http1);
+    for (name, expected) in [
+        ("http1", crate::config::UpstreamProtocol::Http1),
+        ("auto", crate::config::UpstreamProtocol::Auto),
+        ("http2", crate::config::UpstreamProtocol::Http2),
+        ("h2", crate::config::UpstreamProtocol::H2),
+    ] {
+        assert_eq!(
+            serde_json::from_str::<crate::config::UpstreamProtocol>(&format!("\"{name}\""))
+                .unwrap(),
+            expected
+        );
+        assert_eq!(
+            serde_json::to_string(&expected).unwrap(),
+            format!("\"{name}\"")
+        );
+    }
+}

src/upstream/tests.rs

@@ -80,38 +80,42 @@ mod tests {
 
     #[tokio::test]
     async fn explicit_h2c_backend_uses_http2_prior_knowledge() {
-        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
-        let address = listener.local_addr().unwrap();
-        let server = tokio::spawn(async move {
-            let (stream, _) = listener.accept().await.unwrap();
-            let service = service_fn(|request: Request<hyper::body::Incoming>| async move {
-                assert_eq!(request.version(), Version::HTTP_2);
-                Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"h2-ok"))))
+        // The persisted HTTP/2 value and the new strict H2 value both retain
+        // plaintext HTTP/2 prior knowledge, with no HTTP/1.1 probe.
+        for mode in ["http2", "h2"] {
+            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+            let address = listener.local_addr().unwrap();
+            let server = tokio::spawn(async move {
+                let (stream, _) = listener.accept().await.unwrap();
+                let service = service_fn(|request: Request<hyper::body::Incoming>| async move {
+                    assert_eq!(request.version(), Version::HTTP_2);
+                    Ok::<_, Infallible>(Response::new(Full::new(Bytes::from_static(b"h2-ok"))))
+                });
+                let _ = hyper::server::conn::http2::Builder::new(TokioExecutor::new())
+                    .serve_connection(TokioIo::new(stream), service)
+                    .await;
             });
-            let _ = hyper::server::conn::http2::Builder::new(TokioExecutor::new())
-                .serve_connection(TokioIo::new(stream), service)
-                .await;
-        });
-        let backend: Backend = serde_json::from_value(serde_json::json!({
-            "id": "h2",
-            "url": format!("http://{address}"),
-            "protocol": "http2",
-            "allow_private": true
-        }))
-        .unwrap();
-        let client = client(&backend, AddressPolicy::new(&backend).unwrap(), false).unwrap();
-        let request = Request::builder()
-            .version(Version::HTTP_2)
-            .uri(format!("http://{address}/probe"))
-            .body(Body::empty())
+            let backend: Backend = serde_json::from_value(serde_json::json!({
+                "id": "h2",
+                "url": format!("http://{address}"),
+                "protocol": mode,
+                "allow_private": true
+            }))
             .unwrap();
-        let response = client.request(request).await.unwrap();
-        assert_eq!(response.version(), Version::HTTP_2);
-        assert_eq!(
-            response.into_body().collect().await.unwrap().to_bytes(),
-            Bytes::from_static(b"h2-ok")
-        );
-        server.abort();
+            let client = client(&backend, AddressPolicy::new(&backend).unwrap(), false).unwrap();
+            let request = Request::builder()
+                .version(Version::HTTP_2)
+                .uri(format!("http://{address}/probe"))
+                .body(Body::empty())
+                .unwrap();
+            let response = client.request(request).await.unwrap();
+            assert_eq!(response.version(), Version::HTTP_2);
+            assert_eq!(
+                response.into_body().collect().await.unwrap().to_bytes(),
+                Bytes::from_static(b"h2-ok")
+            );
+            server.abort();
+        }
     }
 
     #[test]
@@ -164,4 +168,5 @@ mod tests {
         let addresses = (1..=33).map(|n| format!("10.0.0.{n}:0").parse().unwrap());
         assert!(policy.validate_answers(addresses).is_err());
     }
+    include!("protocol_tests.rs");
 }