plecto-server 0.3.7

//! E2E (tdd-workflow Phase 0) for HTTP/2 termination (ADR 000015): drive a real **HTTP/2** request
//! through `plecto-server` over TLS, negotiated via ALPN. Asserts the handshake selects `h2`, then
//! a multiplexed h2 request routes, runs the chain, and forwards to the (HTTP/1.1) upstream — the
//! request processing path is identical to slice 1, only the wire protocol differs.
//!
//! A fresh self-signed cert (rcgen) backs the listener; a rustls client offers `h2` in its ALPN
//! list (alone, or alongside `http/1.1` to pin the server's h2-first preference) and drives an
//! `hyper` HTTP/2 client connection.

use std::convert::Infallible;
use std::net::SocketAddr;
use std::sync::Arc;

use bytes::Bytes;
use http_body_util::{BodyExt, Empty, Full};
use hyper::body::Incoming;
use hyper::service::service_fn;
use hyper::{Request, Response, StatusCode};
use hyper_util::rt::{TokioExecutor, TokioIo};
use tokio::net::{TcpListener, TcpStream};
use tokio_rustls::TlsConnector;
use tokio_rustls::rustls::pki_types::{CertificateDer, PrivateKeyDer, ServerName};
use tokio_rustls::rustls::{ClientConfig, RootCertStore, crypto::aws_lc_rs};

use plecto_control::{Control, Host, Manifest, MemoryStore, ResolvedArtifact};
use plecto_host::test_support::{TestSigner, bound_sbom, filter_hello_component};
use plecto_server::serve;

/// A fresh self-signed cert for `localhost`, written to a temp dir. Returns the dir (kept alive),
/// the cert + key paths for the manifest, and the cert DER for the client's trust store.
struct TestCert {
    _dir: tempfile::TempDir,
    cert_path: String,
    key_path: String,
    cert_der: CertificateDer<'static>,
    key_der: PrivateKeyDer<'static>,
}

fn make_cert() -> TestCert {
    let generated = rcgen::generate_simple_self_signed(vec!["localhost".to_string()]).unwrap();
    let dir = tempfile::tempdir().unwrap();
    let cert_path = dir.path().join("cert.pem");
    let key_path = dir.path().join("key.pem");
    std::fs::write(&cert_path, generated.cert.pem()).unwrap();
    std::fs::write(&key_path, generated.key_pair.serialize_pem()).unwrap();
    TestCert {
        cert_der: generated.cert.der().clone(),
        key_der: PrivateKeyDer::try_from(generated.key_pair.serialize_der()).unwrap(),
        cert_path: cert_path.to_str().unwrap().to_string(),
        key_path: key_path.to_str().unwrap().to_string(),
        _dir: dir,
    }
}

/// An HTTP/1.1 upstream that echoes a fixed body — Plecto terminates h2 on the client side but
/// forwards to the upstream over HTTP/1.1 (ADR 000015: upstream stays HTTP/1.1).
async fn echo(_req: Request<Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
    Ok(Response::builder()
        .status(200)
        .header("x-from", "upstream")
        .body(Full::new(Bytes::from_static(b"upstream-ok")))
        .unwrap())
}

async fn spawn_upstream() -> SocketAddr {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        loop {
            let (stream, _) = listener.accept().await.unwrap();
            tokio::spawn(async move {
                let _ = hyper::server::conn::http1::Builder::new()
                    .serve_connection(TokioIo::new(stream), service_fn(echo))
                    .await;
            });
        }
    });
    addr
}

/// A manifest declaring filter-hello, a `/api`→echo route, and a default (host-less) `[[tls]]` cert.
fn manifest_toml(upstream: SocketAddr, digest: &str, cert: &TestCert) -> String {
    format!(
        r#"
[[filter]]
id = "fh"
source = "fh"
digest = "{digest}"
isolation = "trusted"

[[upstream]]
name = "echo"
addresses = ["{upstream}"]
[upstream.health]
path = "/healthz"
interval_ms = 50

[[route]]
filters = ["fh"]
upstream = "echo"
strip_prefix = "/api"
[route.match]
path_prefix = "/api"

[[tls]]
cert_path = "{cert_path}"
key_path = "{key_path}"
"#,
        cert_path = cert.cert_path,
        key_path = cert.key_path,
    )
}

fn loaded_control(toml: &str) -> Control {
    let component = filter_hello_component();
    let signer = TestSigner::new().unwrap();
    let component_signature = signer.sign(&component).unwrap();
    let sbom = bound_sbom(&component);
    let sbom_signature = signer.sign(&sbom).unwrap();
    let mut store = MemoryStore::new();
    let digest = store.insert(
        "fh",
        ResolvedArtifact {
            component,
            component_signature,
            sbom,
            sbom_signature,
        },
    );
    let toml = toml.replace("{digest}", &digest);
    let manifest = Manifest::from_toml(&toml).unwrap();
    let host = Host::new(signer.trust_policy().unwrap()).unwrap();
    Control::load(host, &manifest, Box::new(store)).unwrap()
}

async fn spawn_proxy(control: Arc<Control>) -> SocketAddr {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        let _ = serve(control, listener).await;
    });
    addr
}

/// What the client got: the ALPN protocol the handshake selected, plus the (status, body) of one
/// HTTP/2 GET `/api/hello` driven over the negotiated connection.
struct H2Result {
    negotiated_alpn: Option<Vec<u8>>,
    status: StatusCode,
    body: String,
}

/// Connect to `proxy` trusting `root`, offering `alpn_offer` in the ClientHello, then drive one
/// HTTP/2 request and report what came back.
async fn drive_h2(
    proxy: SocketAddr,
    root: CertificateDer<'static>,
    alpn_offer: &[&[u8]],
) -> H2Result {
    let mut roots = RootCertStore::empty();
    roots.add(root).unwrap();
    let mut config = ClientConfig::builder_with_provider(Arc::new(aws_lc_rs::default_provider()))
        .with_safe_default_protocol_versions()
        .unwrap()
        .with_root_certificates(roots)
        .with_no_client_auth();
    config.alpn_protocols = alpn_offer.iter().map(|p| p.to_vec()).collect();
    let connector = TlsConnector::from(Arc::new(config));

    let tcp = TcpStream::connect(proxy).await.unwrap();
    let server_name = ServerName::try_from("localhost").unwrap();
    let tls = connector.connect(server_name, tcp).await.unwrap();
    let negotiated_alpn = tls.get_ref().1.alpn_protocol().map(<[u8]>::to_vec);

    let (mut sender, conn) =
        hyper::client::conn::http2::handshake(TokioExecutor::new(), TokioIo::new(tls))
            .await
            .unwrap();
    tokio::spawn(async move {
        let _ = conn.await;
    });

    let req = Request::builder()
        .method("GET")
        .uri("/api/hello")
        .header("host", "localhost")
        .body(Empty::<Bytes>::new())
        .unwrap();
    let resp = sender.send_request(req).await.unwrap();
    let (parts, body) = resp.into_parts();
    let bytes = body.collect().await.unwrap().to_bytes();
    H2Result {
        negotiated_alpn,
        status: parts.status,
        body: String::from_utf8_lossy(&bytes).into_owned(),
    }
}

/// Drive an h2 request, retrying past the pessimistic-start 503 window (ADR 000017): instances
/// begin unhealthy, so a forward is 503 until the upstream's first health probe lands.
async fn drive_h2_ready(
    proxy: SocketAddr,
    root: CertificateDer<'static>,
    alpn_offer: &[&[u8]],
) -> H2Result {
    for _ in 0..100 {
        let r = drive_h2(proxy, root.clone(), alpn_offer).await;
        if r.status != StatusCode::SERVICE_UNAVAILABLE {
            return r;
        }
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    }
    panic!("upstream never became healthy within the readiness window");
}

#[tokio::test]
async fn negotiates_h2_then_routes_and_forwards() {
    let cert = make_cert();
    let upstream = spawn_upstream().await;
    let control = loaded_control(&manifest_toml(upstream, "{digest}", &cert));
    let proxy = spawn_proxy(Arc::new(control)).await;

    // A client that advertises ONLY h2.
    let r = drive_h2_ready(proxy, cert.cert_der.clone(), &[b"h2"]).await;

    assert_eq!(
        r.negotiated_alpn.as_deref(),
        Some(b"h2".as_ref()),
        "ALPN must negotiate h2 when the client offers it"
    );
    assert_eq!(
        r.status,
        StatusCode::OK,
        "the h2 request routes + forwards 200"
    );
    assert_eq!(
        r.body, "upstream-ok",
        "the upstream body streams back over h2"
    );
}

#[tokio::test]
async fn prefers_h2_when_client_offers_both() {
    let cert = make_cert();
    let upstream = spawn_upstream().await;
    let control = loaded_control(&manifest_toml(upstream, "{digest}", &cert));
    let proxy = spawn_proxy(Arc::new(control)).await;

    // A client offering BOTH: the server's preference order (h2 first, ADR 000015) must win.
    let r = drive_h2_ready(proxy, cert.cert_der.clone(), &[b"h2", b"http/1.1"]).await;

    assert_eq!(
        r.negotiated_alpn.as_deref(),
        Some(b"h2".as_ref()),
        "with both offered, the server prefers h2"
    );
    assert_eq!(r.status, StatusCode::OK);
}

/// A repetitive, over-threshold body — compression is observable by size, not just headers
/// (mirrors tests/compression.rs, which covers the full matrix over HTTP/1.1).
fn big_text() -> String {
    "All work and no play makes the fast path a dull proxy. ".repeat(100)
}

async fn compressible(_req: Request<Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
    Ok(Response::builder()
        .status(200)
        .header("content-type", "text/html")
        .body(Full::new(Bytes::from(big_text())))
        .unwrap())
}

async fn spawn_compressible_upstream() -> SocketAddr {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        loop {
            let (stream, _) = listener.accept().await.unwrap();
            tokio::spawn(async move {
                let _ = hyper::server::conn::http1::Builder::new()
                    .serve_connection(TokioIo::new(stream), service_fn(compressible))
                    .await;
            });
        }
    });
    addr
}

/// One h2 GET with `Accept-Encoding: gzip`, returning the raw parts + wire body (no decode).
async fn drive_h2_gzip(
    proxy: SocketAddr,
    root: CertificateDer<'static>,
) -> (hyper::http::response::Parts, Bytes) {
    let mut roots = RootCertStore::empty();
    roots.add(root).unwrap();
    let mut config = ClientConfig::builder_with_provider(Arc::new(aws_lc_rs::default_provider()))
        .with_safe_default_protocol_versions()
        .unwrap()
        .with_root_certificates(roots)
        .with_no_client_auth();
    config.alpn_protocols = vec![b"h2".to_vec()];
    let connector = TlsConnector::from(Arc::new(config));

    let tcp = TcpStream::connect(proxy).await.unwrap();
    let server_name = ServerName::try_from("localhost").unwrap();
    let tls = connector.connect(server_name, tcp).await.unwrap();

    let (mut sender, conn) =
        hyper::client::conn::http2::handshake(TokioExecutor::new(), TokioIo::new(tls))
            .await
            .unwrap();
    tokio::spawn(async move {
        let _ = conn.await;
    });

    let req = Request::builder()
        .method("GET")
        .uri("/api/hello")
        .header("host", "localhost")
        .header("accept-encoding", "gzip")
        .body(Empty::<Bytes>::new())
        .unwrap();
    let resp = sender.send_request(req).await.unwrap();
    let (parts, body) = resp.into_parts();
    let bytes = body.collect().await.unwrap().to_bytes();
    (parts, bytes)
}

#[tokio::test]
async fn h2_compresses_the_streamed_response_body() {
    // Compression wraps the one `ResponseBody` inside `proxy_core` (ADR 000074) — this pins that
    // hyper's h2 DATA framing carries the compressed stream unchanged (no Content-Length games).
    let cert = make_cert();
    let upstream = spawn_compressible_upstream().await;
    let toml = format!(
        r#"
[[upstream]]
name = "echo"
addresses = ["{upstream}"]
[upstream.health]
path = "/healthz"
interval_ms = 50

[[route]]
upstream = "echo"
[route.match]
path_prefix = "/api"
[route.compression]

[[tls]]
cert_path = "{cert_path}"
key_path = "{key_path}"
"#,
        cert_path = cert.cert_path,
        key_path = cert.key_path,
    );
    let control = loaded_control(&toml);
    let proxy = spawn_proxy(Arc::new(control)).await;

    let (parts, bytes) = {
        let mut result = None;
        for _ in 0..100 {
            let (parts, bytes) = drive_h2_gzip(proxy, cert.cert_der.clone()).await;
            if parts.status != StatusCode::SERVICE_UNAVAILABLE {
                result = Some((parts, bytes));
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        result.expect("upstream never became healthy within the readiness window")
    };

    assert_eq!(parts.status, StatusCode::OK);
    assert_eq!(
        parts.headers.get("content-encoding").map(|v| v.as_bytes()),
        Some(b"gzip".as_slice()),
        "the negotiated coding rides h2 response HEADERS"
    );
    assert!(
        bytes.len() < big_text().len(),
        "the h2 DATA frames carry compressed bytes"
    );
    let mut out = Vec::new();
    std::io::Read::read_to_end(&mut flate2::read::GzDecoder::new(bytes.as_ref()), &mut out)
        .unwrap();
    assert_eq!(out, big_text().as_bytes());
}

// ----- ADR 000078: downstream client-certificate verification on the h2 path -----

/// Like [`drive_h2`], but over a caller-built `ClientConfig`, and reporting failure instead of
/// panicking — an anonymous client against a client-auth listener is EXPECTED to fail, at the
/// connect or on the first request (TLS 1.3 post-handshake alert), and both fold into `Err`.
async fn try_h2_get(proxy: SocketAddr, config: Arc<ClientConfig>) -> Result<StatusCode, String> {
    let connector = TlsConnector::from(config);
    let tcp = TcpStream::connect(proxy).await.map_err(|e| e.to_string())?;
    let server_name = ServerName::try_from("localhost").unwrap();
    let tls = connector
        .connect(server_name, tcp)
        .await
        .map_err(|e| e.to_string())?;
    let (mut sender, conn) =
        hyper::client::conn::http2::handshake(TokioExecutor::new(), TokioIo::new(tls))
            .await
            .map_err(|e| e.to_string())?;
    tokio::spawn(async move {
        let _ = conn.await;
    });
    let req = Request::builder()
        .method("GET")
        .uri("/api/hello")
        .header("host", "localhost")
        .body(Empty::<Bytes>::new())
        .unwrap();
    let resp = sender.send_request(req).await.map_err(|e| e.to_string())?;
    Ok(resp.status())
}

/// Reflects the resolved Host header it received into the response body (`NONE` if absent), so a
/// test can see exactly what the upstream leg was sent.
async fn host_echo(req: Request<Incoming>) -> Result<Response<Full<Bytes>>, Infallible> {
    let body = req
        .headers()
        .get(hyper::header::HOST)
        .and_then(|v| v.to_str().ok())
        .map(|v| Bytes::copy_from_slice(v.as_bytes()))
        .unwrap_or_else(|| Bytes::from_static(b"NONE"));
    Ok(Response::builder()
        .status(200)
        .body(Full::new(body))
        .unwrap())
}

async fn spawn_host_echo_upstream() -> SocketAddr {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        loop {
            let (stream, _) = listener.accept().await.unwrap();
            tokio::spawn(async move {
                let _ = hyper::server::conn::http1::Builder::new()
                    .serve_connection(TokioIo::new(stream), service_fn(host_echo))
                    .await;
            });
        }
    });
    addr
}

/// A filterless `/api`→echo route (no `[[filter]]`) with a default `[[tls]]` cert — the shape the
/// multi-replica reference (`plecto/examples/multi-replica/`) actually runs.
fn manifest_toml_filterless(upstream: SocketAddr, cert: &TestCert) -> String {
    format!(
        r#"
[[upstream]]
name = "echo"
addresses = ["{upstream}"]
[upstream.health]
path = "/healthz"
interval_ms = 50

[[route]]
upstream = "echo"
strip_prefix = "/api"
[route.match]
path_prefix = "/api"

[[tls]]
cert_path = "{cert_path}"
key_path = "{key_path}"
"#,
        cert_path = cert.cert_path,
        key_path = cert.key_path,
    )
}

fn loaded_control_filterless(toml: &str) -> Control {
    let signer = TestSigner::new().unwrap();
    let manifest = Manifest::from_toml(toml).unwrap();
    let host = Host::new(signer.trust_policy().unwrap()).unwrap();
    Control::load(host, &manifest, Box::new(MemoryStore::new())).unwrap()
}

/// Like [`drive_h2`], but sends an absolute-form URI with NO literal `host` header — what a real h2
/// client does (RFC 9113: the authority lives only in `:authority`). `drive_h2`'s `.header("host",
/// ...)` is a build convenience that (unlike a real h2 client) also reaches the wire as a regular
/// header field, which would mask this bug.
async fn drive_h2_no_host_header(proxy: SocketAddr, root: CertificateDer<'static>) -> H2Result {
    let mut roots = RootCertStore::empty();
    roots.add(root).unwrap();
    let mut config = ClientConfig::builder_with_provider(Arc::new(aws_lc_rs::default_provider()))
        .with_safe_default_protocol_versions()
        .unwrap()
        .with_root_certificates(roots)
        .with_no_client_auth();
    config.alpn_protocols = vec![b"h2".to_vec()];
    let connector = TlsConnector::from(Arc::new(config));

    let tcp = TcpStream::connect(proxy).await.unwrap();
    let server_name = ServerName::try_from("localhost").unwrap();
    let tls = connector.connect(server_name, tcp).await.unwrap();

    let (mut sender, conn) =
        hyper::client::conn::http2::handshake(TokioExecutor::new(), TokioIo::new(tls))
            .await
            .unwrap();
    tokio::spawn(async move {
        let _ = conn.await;
    });

    let req = Request::builder()
        .method("GET")
        .uri("https://localhost/api/hello")
        .body(Empty::<Bytes>::new())
        .unwrap();
    let resp = sender.send_request(req).await.unwrap();
    let (parts, body) = resp.into_parts();
    let bytes = body.collect().await.unwrap().to_bytes();
    H2Result {
        negotiated_alpn: None,
        status: parts.status,
        body: String::from_utf8_lossy(&bytes).into_owned(),
    }
}

/// Drive [`drive_h2_no_host_header`], retrying past the pessimistic-start 503 window (ADR 000017).
async fn drive_h2_no_host_header_ready(
    proxy: SocketAddr,
    root: CertificateDer<'static>,
) -> H2Result {
    for _ in 0..100 {
        let r = drive_h2_no_host_header(proxy, root.clone()).await;
        if r.status != StatusCode::SERVICE_UNAVAILABLE {
            return r;
        }
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    }
    panic!("upstream never became healthy within the readiness window");
}

/// Regression (docs/servey f00002, external multi-replica run): a real h2 client carries no literal
/// `Host` header — RFC 9113 puts the authority in `:authority` — and the upstream leg is always
/// HTTP/1.1 (ADR 000042). Before this fix, the forwarded header set had no `Host` at all in that
/// case, so hyper's h1 upstream client synthesized one from the destination URI: the upstream saw
/// ITS OWN resolved address instead of the client's original authority. It must see the client's
/// authority instead, exactly like an HTTP/1.1 client's literal `Host` is already forwarded verbatim.
#[tokio::test]
async fn h2_client_forwards_original_authority_as_host_not_upstream_address() {
    let cert = make_cert();
    let upstream = spawn_host_echo_upstream().await;
    let toml = manifest_toml_filterless(upstream, &cert);
    let control = Arc::new(loaded_control_filterless(&toml));
    let proxy = spawn_proxy(control).await;

    let result = drive_h2_no_host_header_ready(proxy, cert.cert_der.clone()).await;

    assert_eq!(result.status, StatusCode::OK);
    assert_eq!(
        result.body, "localhost",
        "the upstream must see the client's original authority, not its own resolved address"
    );
}

/// Downstream mTLS on the h2 path (ADR 000078): the SAME TCP acceptor + `ServerConfig` that
/// terminate HTTP/1.1 enforce client auth for an h2-negotiating client — authenticated is
/// served over h2, anonymous is refused at the TLS layer.
#[tokio::test]
async fn client_auth_listener_serves_h2_only_to_an_authenticated_client() {
    let cert = make_cert();
    let identity = make_cert_for_client();
    let upstream = spawn_upstream().await;
    let toml = format!(
        "{}\n[listen.client_auth]\nca_path = \"{}\"\n",
        manifest_toml(upstream, "{digest}", &cert),
        identity.cert_path
    );
    let control = loaded_control(&toml);
    let proxy = spawn_proxy(Arc::new(control)).await;

    let mut roots = RootCertStore::empty();
    roots.add(cert.cert_der.clone()).unwrap();
    let mut authed = ClientConfig::builder_with_provider(Arc::new(aws_lc_rs::default_provider()))
        .with_safe_default_protocol_versions()
        .unwrap()
        .with_root_certificates(roots.clone())
        .with_client_auth_cert(
            vec![identity.cert_der.clone()],
            identity.key_der.clone_key(),
        )
        .unwrap();
    authed.alpn_protocols = vec![b"h2".to_vec()];
    let authed = Arc::new(authed);
    let status = tokio::time::timeout(std::time::Duration::from_secs(10), async {
        loop {
            match try_h2_get(proxy, authed.clone()).await {
                Ok(StatusCode::SERVICE_UNAVAILABLE) => {
                    tokio::time::sleep(std::time::Duration::from_millis(20)).await;
                }
                other => break other,
            }
        }
    })
    .await
    .expect("upstream never became healthy")
    .expect("an authenticated h2 client must be served");
    assert_eq!(status, StatusCode::OK, "authenticated h2 client gets 200");

    let mut anon = ClientConfig::builder_with_provider(Arc::new(aws_lc_rs::default_provider()))
        .with_safe_default_protocol_versions()
        .unwrap()
        .with_root_certificates(roots)
        .with_no_client_auth();
    anon.alpn_protocols = vec![b"h2".to_vec()];
    assert!(
        try_h2_get(proxy, Arc::new(anon)).await.is_err(),
        "an anonymous h2 client must be refused at the TLS layer"
    );
}

/// A client identity for the mTLS tests: [`make_cert`] shape, distinct hostname.
fn make_cert_for_client() -> TestCert {
    let generated = rcgen::generate_simple_self_signed(vec!["plecto-client".to_string()]).unwrap();
    let dir = tempfile::tempdir().unwrap();
    let cert_path = dir.path().join("cert.pem");
    let key_path = dir.path().join("key.pem");
    std::fs::write(&cert_path, generated.cert.pem()).unwrap();
    std::fs::write(&key_path, generated.key_pair.serialize_pem()).unwrap();
    TestCert {
        cert_der: generated.cert.der().clone(),
        key_der: PrivateKeyDer::try_from(generated.key_pair.serialize_der()).unwrap(),
        cert_path: cert_path.to_str().unwrap().to_string(),
        key_path: key_path.to_str().unwrap().to_string(),
        _dir: dir,
    }
}