rust-web-server 18.14.0

//! The HTTP/3 upstream client against rws's own HTTP/3 server
//! (`h3_handler`), in-process over loopback UDP.

use super::*;
use crate::application::Application;
use crate::response::STATUS_CODE_REASON_PHRASE;
use crate::server::ConnectionInfo;
use crate::test_env::TempPath;

const BIG: usize = 2 * 1024 * 1024 + 7;

#[derive(Clone)]
struct Backend;

impl Application for Backend {
    fn execute(&self, request: &Request, _connection: &ConnectionInfo) -> Result<Response, String> {
        let header = |name: &str| request.header(name).map(|h| h.value.to_string()).unwrap_or_default();
        Ok(match request.request_uri.as_str() {
            "/hello" => Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, "hello over h3"),
            "/headers" => Response::text(
                STATUS_CODE_REASON_PHRASE.n200_ok,
                &format!("{} via={} xff={} proto={} custom={}", request.http_version, header("via"), header("x-forwarded-for"), header("x-forwarded-proto"), header("x-custom")),
            ),
            "/echo" => {
                let mut r = Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, "");
                r.content_range_list[0].body = format!("{} {}", request.method, String::from_utf8_lossy(&request.body)).into_bytes();
                r
            }
            "/big" => {
                let mut r = Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, "");
                r.content_range_list[0].body = (0..BIG).map(|i| (i % 251) as u8).collect();
                r
            }
            "/trailers" => {
                let mut r = Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, "with trailers");
                r.trailers = Some(Trailers::with(vec![Header::new("x-checksum", "abc".to_string())]));
                r
            }
            _ => Response::text(STATUS_CODE_REASON_PHRASE.n404_not_found, "missing"),
        })
    }
}

fn server_runtime() -> &'static tokio::runtime::Runtime {
    static RUNTIME: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
    RUNTIME.get_or_init(|| tokio::runtime::Builder::new_multi_thread().worker_threads(2).enable_all().build().unwrap())
}

/// An HTTP/3 server for `localhost` with a fresh self-signed certificate;
/// its address and a client configuration trusting it.
fn start_backend() -> (SocketAddr, Arc<rustls::ClientConfig>) {
    let (endpoint, tls) = start_backend_on("127.0.0.1:0".parse().unwrap(), None);
    (endpoint.local_addr().unwrap(), tls)
}

/// [`start_backend`] on `bind`, optionally with a given certificate
/// (`(cert pem, key pem, der)`); returns the endpoint so a test can close it.
fn start_backend_on(bind: SocketAddr, cert: Option<(String, String, Vec<u8>)>) -> (quinn::Endpoint, Arc<rustls::ClientConfig>) {
    static NEXT: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
    let n = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let cert_file = TempPath::new(&format!("h3_upstream_cert_{n}.pem"));
    let key_file = TempPath::new(&format!("h3_upstream_key_{n}.pem"));
    let (cert_pem, key_pem, der) = cert.unwrap_or_else(|| {
        let certified = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap();
        (certified.cert.pem(), certified.signing_key.serialize_pem(), certified.cert.der().to_vec())
    });
    std::fs::write(cert_file.as_str(), &cert_pem).unwrap();
    std::fs::write(key_file.as_str(), &key_pem).unwrap();
    let server_config = crate::tls::create_quinn_server_config(cert_file.as_str(), key_file.as_str()).unwrap();
    let endpoint = server_runtime().block_on(async move {
        let endpoint = quinn::Endpoint::server(server_config, bind).unwrap();
        let accepting = endpoint.clone();
        tokio::spawn(async move {
            while let Some(incoming) = accepting.accept().await {
                tokio::spawn(async move {
                    if let Ok(conn) = incoming.await {
                        let peer = conn.remote_address();
                        let _ = crate::h3_handler::handle_connection(conn, peer, Backend).await;
                    }
                });
            }
        });
        endpoint
    });
    let mut roots = rustls::RootCertStore::empty();
    roots.add(rustls::pki_types::CertificateDer::from(der)).unwrap();
    (endpoint, Arc::new(crate::tls_client_builder().with_root_certificates(roots).with_no_client_auth()))
}

fn target(port: u16, tls: &Arc<rustls::ClientConfig>) -> Target<'_> {
    Target { host: "localhost", port, tls, connect_timeout: Duration::from_millis(1500), read_timeout: Duration::from_secs(5) }
}

fn peer() -> crate::proxy::Peer<'static> {
    crate::proxy::Peer { ip: "203.0.113.9", tls: true }
}

fn request(raw: &str) -> Request {
    Request::parse(raw.as_bytes()).unwrap()
}

fn body(response: &mut Response) -> Vec<u8> {
    match response.stream_pipe.take() {
        Some(mut pipe) => {
            let mut out = Vec::new();
            pipe.read_to_end(&mut out).unwrap();
            out
        }
        None => response.content_range_list.first().map(|c| c.body.clone()).unwrap_or_default(),
    }
}

#[test]
fn requests_reach_the_backend_over_http3_and_share_a_connection() {
    let (addr, tls) = start_backend();
    let target = target(addr.port(), &tls);
    let mut hello = forward(&request("GET /hello HTTP/1.1\r\nHost: shop.example\r\n\r\n"), peer(), &target).unwrap();
    assert_eq!((hello.status_code, body(&mut hello)), (200, b"hello over h3".to_vec()));
    let mut seen = forward(&request("GET /headers HTTP/1.1\r\nHost: shop.example\r\nX-Custom: 1\r\nConnection: keep-alive\r\n\r\n"), peer(), &target).unwrap();
    assert_eq!(String::from_utf8(body(&mut seen)).unwrap(), "HTTP/3.0 via=3 rws xff=203.0.113.9 proto=https custom=1");
    let key = format!("{:x}|localhost:{}", Arc::as_ptr(&tls) as usize, addr.port());
    assert!(pool().lock().unwrap().contains_key(&key), "one pooled connection");
    let mut missing = forward(&request("GET /nothing HTTP/1.1\r\nHost: x\r\n\r\n"), peer(), &target).unwrap();
    assert_eq!((missing.status_code, body(&mut missing)), (404, b"missing".to_vec()));
}

#[test]
fn request_bodies_go_up_whole_or_streamed() {
    let (addr, tls) = start_backend();
    let target = target(addr.port(), &tls);
    let mut echo = forward(&request("POST /echo HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\n\r\nhello"), peer(), &target).unwrap();
    assert_eq!(body(&mut echo), b"POST hello");
    let mut streamed = request("PUT /echo HTTP/1.1\r\nHost: x\r\n\r\n");
    let chunks: Vec<u8> = b"streamed ".repeat(5000);
    streamed.body = Body::stream(std::io::Cursor::new(chunks.clone()));
    let mut echo = forward(&streamed, peer(), &target).unwrap();
    assert_eq!(body(&mut echo), [b"PUT ".as_slice(), &chunks].concat());
}

#[test]
fn large_responses_stream_and_trailers_arrive() {
    let (addr, tls) = start_backend();
    let target = target(addr.port(), &tls);
    let mut big = forward(&request("GET /big HTTP/1.1\r\nHost: x\r\n\r\n"), peer(), &target).unwrap();
    assert!(big.stream_pipe.is_some(), "over the buffer limit: streamed");
    let bytes = body(&mut big);
    assert_eq!(bytes.len(), BIG);
    assert!(bytes.iter().enumerate().all(|(i, b)| *b == (i % 251) as u8));
    let mut with_trailers = forward(&request("GET /trailers HTTP/1.1\r\nHost: x\r\n\r\n"), peer(), &target).unwrap();
    assert_eq!(body(&mut with_trailers), b"with trailers");
    let trailers = with_trailers.trailers.expect("trailers").get();
    assert!(trailers.iter().any(|h| h.name == "x-checksum" && h.value == "abc"), "{trailers:?}");
}

#[test]
fn connect_failures_are_told_apart() {
    // Nothing listens on this UDP port.
    let free = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
    let port = free.local_addr().unwrap().port();
    drop(free);
    let (_, tls) = start_backend();
    let started = std::time::Instant::now();
    let error = forward(&request("GET / HTTP/1.1\r\nHost: x\r\n\r\n"), peer(), &target(port, &tls)).unwrap_err();
    assert!(matches!(error, H3Error::Connect(_)), "{error}");
    assert!(started.elapsed() < Duration::from_secs(4), "bounded by the connect timeout");

    // A backend whose certificate isn't trusted: the handshake fails.
    let (addr, _) = start_backend();
    let error = forward(&request("GET / HTTP/1.1\r\nHost: x\r\n\r\n"), peer(), &target(addr.port(), &crate::proxy::tls_client_config())).unwrap_err();
    assert!(matches!(error, H3Error::Connect(_)), "{error}");
}

#[test]
fn the_async_entry_point_answers_too() {
    let (addr, tls) = start_backend();
    let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
    let mut hello = runtime.block_on(async { forward_async(&request("GET /hello HTTP/1.1\r\nHost: x\r\n\r\n"), peer(), &target(addr.port(), &tls)).await }).unwrap();
    assert_eq!(body(&mut hello), b"hello over h3");
}

#[test]
fn a_closed_connection_is_replaced_by_a_new_one() {
    let certified = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap();
    let cert = (certified.cert.pem(), certified.signing_key.serialize_pem(), certified.cert.der().to_vec());
    let (first, tls) = start_backend_on("127.0.0.1:0".parse().unwrap(), Some(cert.clone()));
    let addr = first.local_addr().unwrap();
    let mut hello = forward(&request("GET /hello HTTP/1.1\r\nHost: x\r\n\r\n"), peer(), &target(addr.port(), &tls)).unwrap();
    assert_eq!(body(&mut hello), b"hello over h3");
    // The backend restarts: its connections are closed, a new one listens
    // on the same port with the same certificate.
    first.close(0u32.into(), b"restart");
    std::thread::sleep(Duration::from_millis(300));
    drop(first);
    std::thread::sleep(Duration::from_millis(100));
    let (_second, _) = start_backend_on(addr, Some(cert));
    let mut again = forward(&request("GET /hello HTTP/1.1\r\nHost: x\r\n\r\n"), peer(), &target(addr.port(), &tls)).unwrap();
    assert_eq!(body(&mut again), b"hello over h3");
}

// ── config-file routes: [[upstream]] protocol = "h3" ──────────────────────────

fn config_app(toml: &str) -> crate::proxy_config::ConfigDrivenApp {
    let report = crate::config_check::check_str(toml);
    assert!(report.errors.is_empty(), "{:?}", report.errors);
    let config = crate::proxy_config::ProxyConfig::from_str(toml);
    assert!(config.routes[0].middleware.is_empty(), "h3 keeps the async forwarder");
    crate::proxy_config::builder::build(config).0
}

fn h3_toml(port: u16, ca_file: &str) -> String {
    format!(
        "[[upstream]]\nname = \"app\"\nprotocol = \"h3\"\nbackends = [\"https://localhost:{port}\"]\ntls_ca_file = \"{ca_file}\"\n\n\
         [[route]]\n[route.match]\npath = \"/*\"\n[route.action]\ntype = \"proxy\"\nupstream = \"app\"\nconnect_timeout_ms = 1000\n"
    )
}

#[test]
fn config_proxy_routes_reach_an_h3_upstream_on_both_paths() {
    let certified = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap();
    let ca = TempPath::new("h3_upstream_route_ca.pem");
    std::fs::write(ca.as_str(), certified.cert.pem()).unwrap();
    let (endpoint, _) = start_backend_on("127.0.0.1:0".parse().unwrap(), Some((certified.cert.pem(), certified.signing_key.serialize_pem(), certified.cert.der().to_vec())));
    let app = config_app(&h3_toml(endpoint.local_addr().unwrap().port(), ca.as_str()));
    let conn = ConnectionInfo { tls: true, ..ConnectionInfo::default() };
    let get = request("GET /headers HTTP/1.1\r\nHost: shop.example\r\n\r\n");
    let mut sync = app.execute(&get, &conn).unwrap();
    assert!(String::from_utf8(body(&mut sync)).unwrap().starts_with("HTTP/3.0 via=3 rws"), "sync path over HTTP/3");
    let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
    let mut async_response = runtime.block_on(async { app.execute_async(&get, &conn).expect("async route").await }).unwrap();
    assert!(String::from_utf8(body(&mut async_response)).unwrap().starts_with("HTTP/3.0"), "async path over HTTP/3");
}

#[test]
fn a_backend_quic_cannot_reach_is_served_over_tcp_and_remembered() {
    // HTTPS on TCP only: nothing on that UDP port.
    let (port, _, accepted, pem) = crate::proxy::tests::tls_backend_with_pem(b"HTTP/1.1 200 OK\r\nContent-Length: 3\r\n\r\ntcp", 100);
    let ca = TempPath::new("h3_upstream_fallback_ca.pem");
    std::fs::write(ca.as_str(), pem).unwrap();
    let app = config_app(&h3_toml(port, ca.as_str()).replace("name = \"app\"", "name = \"fallback-test\"").replace("upstream = \"app\"", "upstream = \"fallback-test\""));
    let get = request("GET /x HTTP/1.1\r\nHost: x\r\n\r\n");
    let count = || {
        crate::proxy_config::upstream::prometheus_text()
            .lines()
            .find(|l| l.starts_with("rws_upstream_h3_fallbacks_total{upstream=\"fallback-test\"}"))
            .map(|l| l.rsplit(' ').next().unwrap().to_string())
    };
    let started = std::time::Instant::now();
    let mut first = app.execute(&get, &ConnectionInfo::default()).unwrap();
    assert_eq!((first.status_code, body(&mut first)), (200, b"tcp".to_vec()));
    assert!(started.elapsed() < Duration::from_secs(3), "bounded by connect_timeout_ms");
    assert_eq!(count().as_deref(), Some("1"));
    let started = std::time::Instant::now();
    let mut second = app.execute(&get, &ConnectionInfo::default()).unwrap();
    assert_eq!(body(&mut second), b"tcp");
    assert!(started.elapsed() < Duration::from_millis(500), "no second QUIC attempt");
    assert_eq!(count().as_deref(), Some("1"), "remembered: no new fallback");
    assert!(accepted.load(std::sync::atomic::Ordering::Relaxed) >= 1);
}

#[test]
fn rws_check_reports_protocol_mistakes() {
    let base = |protocol: &str, backend: &str| {
        format!("[[upstream]]\nname = \"u\"\nprotocol = \"{protocol}\"\nbackends = [\"{backend}\"]\n\n[[route]]\n[route.match]\npath = \"/\"\n[route.action]\ntype = \"proxy\"\nupstream = \"u\"\n")
    };
    let errors = |toml: String| crate::config_check::check_str_offline(&toml).errors.join("\n");
    assert!(errors(base("h3", "https://10.0.0.5:443")).is_empty());
    assert!(errors(base("http1", "http://10.0.0.5:80")).is_empty());
    assert!(errors(base("h3", "http://10.0.0.5:80")).contains("needs https:// backends"));
    assert!(errors(base("quic", "https://10.0.0.5:443")).contains("protocol must be"));
}

// ── H3ReverseProxy ──────────────────────────────────────────────────────────────

#[test]
fn the_library_proxy_forwards_over_http3_and_falls_back_to_tcp() {
    use crate::middleware::WithMiddleware;
    let certified = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap();
    let ca = TempPath::new("h3_reverse_proxy_ca.pem");
    std::fs::write(ca.as_str(), certified.cert.pem()).unwrap();
    let (endpoint, _) = start_backend_on("127.0.0.1:0".parse().unwrap(), Some((certified.cert.pem(), certified.signing_key.serialize_pem(), certified.cert.der().to_vec())));
    let proxy = crate::proxy::H3ReverseProxy::new([format!("https://localhost:{}", endpoint.local_addr().unwrap().port())])
        .path_prefix("/headers")
        .tls_ca_file(ca.as_str())
        .unwrap();
    let app = WithMiddleware::new(crate::app::App::with_state(()).get("/local", |_, _, _, _| Response::text(STATUS_CODE_REASON_PHRASE.n200_ok, "local"))).wrap(proxy);
    let client = crate::test_client::TestClient::new(app);
    assert!(client.get("/headers").send().body_text().starts_with("HTTP/3.0 via=3 rws"));
    assert_eq!(client.get("/local").send().body_text(), "local", "outside the prefix: passed through");

    let (port, _, _, pem) = crate::proxy::tests::tls_backend_with_pem(b"HTTP/1.1 200 OK\r\nContent-Length: 3\r\n\r\ntcp", 100);
    let tcp_ca = TempPath::new("h3_reverse_proxy_tcp_ca.pem");
    std::fs::write(tcp_ca.as_str(), pem).unwrap();
    let fallback = crate::proxy::H3ReverseProxy::new([format!("https://localhost:{port}")])
        .name("library-fallback-test")
        .connect_timeout_ms(1000)
        .tls_ca_file(tcp_ca.as_str())
        .unwrap();
    let client = crate::test_client::TestClient::new(WithMiddleware::new(crate::app::App::new()).wrap(fallback));
    assert_eq!(client.get("/x").send().body_text(), "tcp");
    assert_eq!(client.get("/x").send().body_text(), "tcp");
    assert!(crate::proxy_config::upstream::prometheus_text().contains("rws_upstream_h3_fallbacks_total{upstream=\"library-fallback-test\"} 1"));
}

#[test]
fn every_resolved_address_is_tried_at_once() {
    // Where `localhost` resolves to both families, listen only on the one
    // that comes second: trying addresses in turn would spend the whole
    // connect timeout on the silent first one (CI runners resolve ::1
    // first; QUIC to a port nobody answers on just stays quiet).
    use std::net::ToSocketAddrs;
    let resolved: Vec<SocketAddr> = ("localhost", 0).to_socket_addrs().map(|a| a.collect()).unwrap_or_default();
    let Some(first) = resolved.first() else { return };
    let Some(second) = resolved.iter().find(|a| a.is_ipv6() != first.is_ipv6()) else {
        eprintln!("localhost has one address family here ({resolved:?}); skipped");
        return;
    };
    let bind = SocketAddr::new(second.ip(), 0);
    let (endpoint, tls) = start_backend_on(bind, None);
    let started = std::time::Instant::now();
    let mut hello = forward(&request("GET /hello HTTP/1.1\r\nHost: x\r\n\r\n"), peer(), &target(endpoint.local_addr().unwrap().port(), &tls)).unwrap();
    assert_eq!(body(&mut hello), b"hello over h3");
    assert!(started.elapsed() < Duration::from_millis(1000), "didn't wait out the silent address: {:?}", started.elapsed());
}