plecto-server 0.3.5

//! Header handling for the fast path: the hop-by-hop strip (RFC 9110 §7.6.1), edge client-IP
//! propagation (X-Forwarded-* / Forwarded), the header-only request projection, and the
//! contract↔hyper conversions (including the byte-preserving pass-through, P3#6).

use std::collections::HashSet;
use std::net::IpAddr;

use hyper::header::{HeaderName, HeaderValue};
use plecto_control::{Header, HttpRequest};

/// Hop-by-hop headers a proxy must not forward (RFC 9110 §7.6.1). Stripped both ways so the
/// upstream's framing (`transfer-encoding`) and connection management never collide with the
/// fresh framing hyper computes for the leg we send.
const HOP_BY_HOP: &[&str] = &[
    "connection",
    "keep-alive",
    "proxy-connection",
    "transfer-encoding",
    "te",
    "trailer",
    "upgrade",
    // Proxy-scoped credentials/challenges are hop-by-hop (RFC 9110 §11.7.1/§11.7.2): a
    // client's `Proxy-Authorization` must not leak to the upstream, nor an upstream's
    // `Proxy-Authenticate` back to the client.
    "proxy-authorization",
    "proxy-authenticate",
];

fn is_hop_by_hop(name: &str) -> bool {
    HOP_BY_HOP.iter().any(|h| name.eq_ignore_ascii_case(h))
}

/// The set of header names DYNAMICALLY designated hop-by-hop by `Connection` (RFC 9110 §7.6.1):
/// `Connection: X-Foo, close` marks `X-Foo` connection-specific, so a proxy must not forward it.
/// Forwarding such a header is a request-smuggling / header-leak primitive, so we strip these too
/// — not just the static `HOP_BY_HOP` set (review f000005 P2#5). Tokens are lower-cased for the
/// case-insensitive name compare; `close` / `keep-alive` are inert (no such header to drop).
fn connection_named(map: &hyper::HeaderMap) -> HashSet<String> {
    let mut named = HashSet::new();
    for value in map.get_all(hyper::header::CONNECTION).iter() {
        if let Ok(s) = value.to_str() {
            for token in s.split(',') {
                let token = token.trim();
                // `close` / `keep-alive` are the overwhelmingly common tokens here and are
                // connection-management directives, not header names to strip (no such header
                // exists to drop) — skip the allocation + hash-insert for them on the hot path.
                if !token.is_empty()
                    && !token.eq_ignore_ascii_case("close")
                    && !token.eq_ignore_ascii_case("keep-alive")
                {
                    named.insert(token.to_ascii_lowercase());
                }
            }
        }
    }
    named
}

/// The client's `Upgrade` header value when the request genuinely asks for a protocol switch:
/// RFC 9110 §7.8 requires an `upgrade` option in `Connection` alongside the `Upgrade` header,
/// so both must be present — an `Upgrade` header without the `Connection` option is not an
/// upgrade request and stays subject to the plain hop-by-hop strip.
pub(crate) fn upgrade_request_header(map: &hyper::HeaderMap) -> Option<&str> {
    if !connection_named(map).contains("upgrade") {
        return None;
    }
    map.get(hyper::header::UPGRADE)?.to_str().ok()
}

/// Did the client's `TE` header ask for trailers (RFC 9110 §10.1.4)? Tokens may carry parameters
/// or weights (`trailers`, `gzip;q=0.5`), so each comma-separated token is compared by its bare
/// name. Used by the forward path (ADR 000042) to re-issue exactly `te: trailers` on an
/// h2-capable upstream leg — the gRPC proxy-compat signal — while every other TE value stays
/// stripped as hop-by-hop.
pub(crate) fn te_requests_trailers(map: &hyper::HeaderMap) -> bool {
    map.get_all(hyper::header::TE).iter().any(|value| {
        value.to_str().is_ok_and(|s| {
            s.split(',').any(|token| {
                token
                    .split(';')
                    .next()
                    .is_some_and(|name| name.trim().eq_ignore_ascii_case("trailers"))
            })
        })
    })
}

/// Forwarding / client-IP headers a client could spoof: RFC 7239 `Forwarded`, the de-facto
/// `X-Forwarded-*`, and the de-facto client-IP family that many backends and CDNs trust (nginx's
/// `X-Real-IP`, Akamai/Cloudflare `True-Client-IP`, `CF-Connecting-IP`, `Fastly-Client-IP`,
/// `X-Client-IP`, `X-Cluster-Client-IP`). As an EDGE proxy Plecto strips this whole family on
/// ingress and sets its own (review f000005 P2#3 / ADR 000018 + 000022), so an untrusted client
/// cannot forge its source IP / scheme for an IP-based filter or the upstream — stripping `XFF`
/// alone would leave a spoofed `X-Real-IP` to fool a backend that reads it instead.
const FORWARDED_HEADERS: &[&str] = &[
    "forwarded",
    "x-forwarded-for",
    "x-forwarded-proto",
    "x-forwarded-host",
    "x-real-ip",
    "true-client-ip",
    "cf-connecting-ip",
    "fastly-client-ip",
    "x-client-ip",
    "x-cluster-client-ip",
];

/// Edge-proxy client-IP propagation: drop any client-supplied forwarding / client-IP headers
/// (`FORWARDED_HEADERS`), then set `X-Forwarded-For` and `X-Real-IP` (the real connection peer) and
/// `X-Forwarded-Proto` (the wire scheme) afresh. `X-Real-IP` is re-issued — not just stripped — so a
/// backend reading the nginx convention rather than `XFF` still gets Plecto's authoritative peer
/// (ADR 000022 widens ADR 000018's "issue For+Proto only"). The chain (so IP-based rate-limit / auth
/// filters can trust them) and the upstream then see only Plecto's values, never the client's claim.
/// A trusted-proxy *append* mode (Plecto behind another LB) is a manifest knob deferred to a later
/// slice; overwrite is the safe default.
///
/// Operates directly on the inbound hyper `HeaderMap`, BEFORE the contract projection: the
/// corrected headers then flow to the chain via `to_http_request` and to the filterless direct
/// forward path verbatim — one application, both consumers.
pub(crate) fn set_forwarded(headers: &mut hyper::HeaderMap, peer: IpAddr, scheme: &str) {
    for name in FORWARDED_HEADERS {
        // `HeaderName` lookups are case-insensitive; the constants are already lowercase.
        headers.remove(*name);
    }
    // An IPv4 client on a dual-stack ([::]) listener arrives as an IPv4-mapped IPv6 address
    // (`::ffff:a.b.c.d`); normalise it to dotted IPv4 so backends/filters that all-list on the IPv4
    // form match, and the value matches what nginx/Envoy would emit. A genuine IPv6 peer is kept
    // verbatim (no brackets — XFF carries a bare address).
    let client_ip = match peer {
        IpAddr::V6(v6) => match v6.to_ipv4_mapped() {
            Some(v4) => v4.to_string(),
            None => v6.to_string(),
        },
        IpAddr::V4(v4) => v4.to_string(),
    };
    // An IP address in decimal/hex-free ASCII is always a valid header value; stay total on the
    // data plane regardless (drop rather than panic if that were ever not the case).
    if let Ok(ip_value) = HeaderValue::from_str(&client_ip) {
        headers.insert("x-forwarded-for", ip_value.clone());
        headers.insert("x-real-ip", ip_value);
    }
    if let Ok(proto) = HeaderValue::from_str(scheme) {
        headers.insert("x-forwarded-proto", proto);
    }
}

/// Build a header-only `HttpRequest` (the chain's view) from the inbound request parts. The body
/// is handled separately (streamed), so it is absent here — the v0.1 contract is header-only.
///
/// `scheme` is the connection-level truth (`"https"` when the fast path terminated TLS on this
/// connection, `"http"` for plaintext) — not the request URI's scheme, which a client can spoof.
pub(crate) fn to_http_request(parts: &hyper::http::request::Parts, scheme: &str) -> HttpRequest {
    let path = parts
        .uri
        .path_and_query()
        .map(|pq| pq.as_str().to_string())
        .unwrap_or_else(|| "/".to_string());
    HttpRequest {
        method: parts.method.as_str().to_string(),
        path,
        authority: request_authority(parts),
        scheme: scheme.to_string(),
        headers: headers_to_vec(&parts.headers),
    }
}

/// The request's authority: for HTTP/2 the `:authority` pseudo-header lands in the URI; for
/// HTTP/1.1 it is the Host header. Prefer the URI authority (h2), falling back to Host, then to
/// empty. Shared by the contract projection above and the access-log capture.
pub(crate) fn request_authority(parts: &hyper::http::request::Parts) -> String {
    parts
        .uri
        .authority()
        .map(|a| a.to_string())
        .or_else(|| {
            parts
                .headers
                .get(hyper::header::HOST)
                .and_then(|v| v.to_str().ok())
                .map(str::to_string)
        })
        .unwrap_or_default()
}

/// Convert a hyper `HeaderMap` to the contract's `Vec<Header>` (byte-preserving). Both the static
/// hop-by-hop set AND any header dynamically named by `Connection` are dropped (RFC 9110 §7.6.1) —
/// this is the single ingress point for both the request (from `parts`) and the response (from the
/// upstream parts), so a connection-specific header can never be carried into the contract and
/// forwarded.
pub(crate) fn headers_to_vec(map: &hyper::HeaderMap) -> Vec<Header> {
    let named = connection_named(map);
    map.iter()
        .filter(|(name, _)| {
            // `HeaderName::as_str()` is already lowercase (the `http` crate normalizes on parse),
            // and `named` holds lowered tokens — compare directly, no per-header allocation.
            !is_hop_by_hop(name.as_str()) && (named.is_empty() || !named.contains(name.as_str()))
        })
        .map(|(name, value)| Header {
            name: name.as_str().to_string(),
            value: value.as_bytes().to_vec(),
        })
        .collect()
}

/// Copy contract headers into a hyper `HeaderMap`, skipping hop-by-hop and any that fail hyper's
/// validation (a malformed name/value is dropped, never panics — data-plane no-panic).
pub(crate) fn copy_headers(dst: Option<&mut hyper::HeaderMap>, headers: &[Header]) {
    copy_headers_filtered(dst, headers, false);
}

/// Like [`copy_headers`], but also drops `Content-Length` — for synthesised (buffered) responses
/// where the host owns framing from `full(body)` and a guest-supplied length must not desync the
/// wire (ADR 000073 review). `Transfer-Encoding` is already hop-by-hop.
pub(crate) fn copy_headers_synth(dst: Option<&mut hyper::HeaderMap>, headers: &[Header]) {
    copy_headers_filtered(dst, headers, true);
}

fn copy_headers_filtered(
    dst: Option<&mut hyper::HeaderMap>,
    headers: &[Header],
    strip_content_length: bool,
) {
    let Some(dst) = dst else { return };
    for h in headers {
        if is_hop_by_hop(&h.name) {
            continue;
        }
        if strip_content_length && h.name.eq_ignore_ascii_case("content-length") {
            continue;
        }
        if let (Ok(name), Ok(value)) = (
            HeaderName::from_bytes(h.name.as_bytes()),
            HeaderValue::from_bytes(&h.value),
        ) {
            dst.append(name, value);
        }
    }
}

/// Copy a hyper `HeaderMap` into another directly — the filterless fast path. Drops the static
/// hop-by-hop set AND any header dynamically named by `Connection` (RFC 9110 §7.6.1), exactly like
/// `headers_to_vec` + `copy_headers` compose, but without the contract projection: the original
/// bytes forward verbatim (`HeaderName`/`HeaderValue` clones are refcounted, no copy).
pub(crate) fn copy_headers_direct(dst: Option<&mut hyper::HeaderMap>, src: &hyper::HeaderMap) {
    let Some(dst) = dst else { return };
    let named = connection_named(src);
    for (name, value) in src {
        if is_hop_by_hop(name.as_str()) || (!named.is_empty() && named.contains(name.as_str())) {
            continue;
        }
        dst.append(name.clone(), value.clone());
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use hyper::Request;

    fn parts(authority_in_uri: bool) -> hyper::http::request::Parts {
        let uri = if authority_in_uri {
            "https://h2.example/api/x"
        } else {
            "/api/x"
        };
        Request::builder()
            .method("GET")
            .uri(uri)
            .header("host", "h1.example")
            .body(())
            .unwrap()
            .into_parts()
            .0
    }

    #[test]
    fn scheme_reflects_tls_termination_not_a_hardcoded_value() {
        // A TLS-terminated connection must surface `https` to the chain; plaintext surfaces `http`.
        // (The scheme is connection truth — what the fast path terminated — so a filter that, say,
        // redirects http→https can trust it.)
        assert_eq!(to_http_request(&parts(false), "https").scheme, "https");
        assert_eq!(to_http_request(&parts(false), "http").scheme, "http");
    }

    #[test]
    fn authority_prefers_h2_uri_authority_then_falls_back_to_host() {
        // HTTP/2 carries the host in the URI (`:authority`); HTTP/1.1 carries it in the Host header.
        assert_eq!(
            to_http_request(&parts(true), "https").authority,
            "h2.example"
        );
        assert_eq!(
            to_http_request(&parts(false), "http").authority,
            "h1.example"
        );
    }

    fn header(name: &str, value: &str) -> Header {
        Header {
            name: name.to_string(),
            value: value.as_bytes().to_vec(),
        }
    }

    #[test]
    fn hop_by_hop_set_is_recognised_case_insensitively() {
        // The exact RFC 9110 §7.6.1 connection-management set a proxy must never forward, plus
        // `transfer-encoding` — forwarding a client's `Transfer-Encoding`/`Connection` next to the
        // fresh framing hyper computes is the classic request-smuggling primitive (CWE-444).
        for h in [
            "connection",
            "Keep-Alive",
            "PROXY-CONNECTION",
            "Transfer-Encoding",
            "te",
            "Trailer",
            "upgrade",
            "Proxy-Authorization",
            "Proxy-Authenticate",
        ] {
            assert!(is_hop_by_hop(h), "{h} must be treated as hop-by-hop");
        }
        // a normal end-to-end header is not hop-by-hop.
        assert!(!is_hop_by_hop("x-api-key"));
        assert!(!is_hop_by_hop("content-type"));
    }

    #[test]
    fn te_requests_trailers_matches_the_token_not_the_whole_value() {
        // RFC 9110 §10.1.4: TE is a comma-separated list whose tokens may carry parameters.
        // The forward path re-issues `te: trailers` only when the client actually asked for
        // trailers (ADR 000042) — never for other transfer codings.
        for value in [
            "trailers",
            "Trailers",
            "gzip, trailers",
            "trailers;q=1",
            "gzip;q=0.5, trailers",
        ] {
            let mut map = hyper::HeaderMap::new();
            map.insert(hyper::header::TE, HeaderValue::from_str(value).unwrap());
            assert!(te_requests_trailers(&map), "{value:?} requests trailers");
        }
        for value in ["gzip", "compress, deflate", "trailersx"] {
            let mut map = hyper::HeaderMap::new();
            map.insert(hyper::header::TE, HeaderValue::from_str(value).unwrap());
            assert!(!te_requests_trailers(&map), "{value:?} must not match");
        }
        assert!(
            !te_requests_trailers(&hyper::HeaderMap::new()),
            "no TE header, no trailers request"
        );
    }

    #[test]
    fn headers_to_vec_strips_hop_by_hop_on_ingress() {
        // What the chain (and ultimately the upstream) sees must already be free of connection-
        // management headers: stripping them on the way in is half the smuggling defence (the
        // other half is `copy_headers` on the way out).
        let mut map = hyper::HeaderMap::new();
        map.insert("x-keep", HeaderValue::from_static("1"));
        map.insert(
            hyper::header::CONNECTION,
            HeaderValue::from_static("keep-alive"),
        );
        map.insert(
            hyper::header::TRANSFER_ENCODING,
            HeaderValue::from_static("chunked"),
        );
        map.insert(hyper::header::TE, HeaderValue::from_static("trailers"));

        let out = headers_to_vec(&map);
        assert!(
            out.iter().all(|h| !is_hop_by_hop(&h.name)),
            "no hop-by-hop header may survive the ingress conversion"
        );
        assert!(
            out.iter().any(|h| h.name == "x-keep"),
            "an end-to-end header is preserved"
        );
    }

    #[test]
    fn headers_to_vec_strips_connection_named_headers() {
        // RFC 9110 §7.6.1 (review f000005 P2#5): a header NAMED by `Connection` is connection-
        // specific and must not be forwarded. A client using `Connection: x-secret` to smuggle
        // `x-secret` past the proxy is defeated; inert tokens (`close`) are ignored.
        let mut map = hyper::HeaderMap::new();
        map.insert(
            hyper::header::CONNECTION,
            HeaderValue::from_static("X-Secret, close"),
        );
        map.append("x-secret", HeaderValue::from_static("leak"));
        map.insert("x-keep", HeaderValue::from_static("1"));

        let out = headers_to_vec(&map);
        assert!(
            !out.iter().any(|h| h.name.eq_ignore_ascii_case("x-secret")),
            "a Connection-named header must be stripped"
        );
        assert!(
            !out.iter()
                .any(|h| h.name.eq_ignore_ascii_case("connection")),
            "Connection itself is hop-by-hop"
        );
        assert!(
            out.iter().any(|h| h.name == "x-keep"),
            "an unrelated end-to-end header survives"
        );
    }

    #[test]
    fn copy_headers_forwards_non_utf8_bytes() {
        // P3#6: the contract carries header values as bytes, so pass-through headers stay
        // byte-exact on egress — not re-encoded from a lossy string and not dropped.
        let raw: &[u8] = &[0xC3, 0x28]; // invalid UTF-8
        let contract = vec![Header {
            name: "x-blob".to_string(),
            value: raw.to_vec(),
        }];

        let mut dst = hyper::HeaderMap::new();
        copy_headers(Some(&mut dst), &contract);

        assert_eq!(
            dst.get("x-blob").map(|v| v.as_bytes()),
            Some(raw),
            "a byte-valued header forwards byte-for-byte"
        );
    }

    #[test]
    fn set_forwarded_overwrites_spoofed_client_headers() {
        // Edge model (review f000005 P2#3 / ADR 000018 + 000022): the whole de-facto client-IP
        // family — X-Forwarded-For / Forwarded / X-Real-IP / CDN headers — is STRIPPED and the
        // peer's value re-issued, never appended-to or trusted, so an untrusted client cannot forge
        // its source IP. X-Forwarded-For and X-Real-IP carry the real peer; X-Forwarded-Proto the
        // wire scheme; a stripped CDN header (CF-Connecting-IP) is NOT re-issued.
        let mut headers = hyper::HeaderMap::new();
        headers.append("x-forwarded-for", HeaderValue::from_static("9.9.9.9"));
        headers.append("x-forwarded-for", HeaderValue::from_static("8.8.4.4"));
        headers.insert("forwarded", HeaderValue::from_static("for=10.0.0.1"));
        headers.insert("x-real-ip", HeaderValue::from_static("9.9.9.9"));
        headers.insert("cf-connecting-ip", HeaderValue::from_static("8.8.8.8"));
        headers.insert("x-keep", HeaderValue::from_static("1"));
        set_forwarded(&mut headers, "203.0.113.5".parse().unwrap(), "https");

        let xff: Vec<&HeaderValue> = headers.get_all("x-forwarded-for").iter().collect();
        assert_eq!(
            xff,
            vec!["203.0.113.5"],
            "the spoofed XFF (all values) is replaced by the real peer (one value, not appended)"
        );
        assert_eq!(
            headers.get("x-real-ip").map(HeaderValue::as_bytes),
            Some(b"203.0.113.5".as_ref()),
            "the spoofed X-Real-IP is replaced by the real peer (re-issued)"
        );
        assert!(
            !headers.contains_key("forwarded"),
            "a spoofed Forwarded header is stripped"
        );
        assert!(
            !headers.contains_key("cf-connecting-ip"),
            "a spoofed CDN client-IP header is stripped and not re-issued"
        );
        assert_eq!(
            headers
                .get("x-forwarded-proto")
                .and_then(|v| v.to_str().ok()),
            Some("https"),
            "X-Forwarded-Proto reflects the connection scheme"
        );
        assert!(
            headers.contains_key("x-keep"),
            "an unrelated header is left intact"
        );
    }

    #[test]
    fn set_forwarded_normalises_ipv4_mapped_peer() {
        // An IPv4 client on a dual-stack ([::]) listener arrives as an IPv4-mapped IPv6 peer
        // (`::ffff:a.b.c.d`); X-Forwarded-For / X-Real-IP must carry the dotted IPv4 form so a
        // backend all-listing on the IPv4 address matches (ADR 000022).
        let mut headers = hyper::HeaderMap::new();
        set_forwarded(&mut headers, "::ffff:203.0.113.5".parse().unwrap(), "https");
        for name in ["x-forwarded-for", "x-real-ip"] {
            assert_eq!(
                headers.get(name).and_then(|v| v.to_str().ok()),
                Some("203.0.113.5"),
                "an IPv4-mapped peer normalises to dotted IPv4 in {name}"
            );
        }

        // A genuine IPv6 peer is preserved verbatim (no brackets in XFF).
        let mut headers = hyper::HeaderMap::new();
        set_forwarded(&mut headers, "2001:db8::1".parse().unwrap(), "https");
        assert_eq!(
            headers.get("x-forwarded-for").and_then(|v| v.to_str().ok()),
            Some("2001:db8::1"),
            "a real IPv6 peer is kept as-is"
        );
    }

    #[test]
    fn copy_headers_drops_hop_by_hop_crlf_and_malformed_names() {
        // Egress side: a filter (or a buggy/hostile one) must not be able to smuggle framing or
        // inject a header via an embedded CRLF (CWE-113) or a malformed name. `copy_headers` drops
        // each silently and never panics (data-plane discipline) — and the rest still copies.
        let mut dst = hyper::HeaderMap::new();
        copy_headers(
            Some(&mut dst),
            &[
                header("x-ok", "fine"),
                header("transfer-encoding", "chunked"), // hop-by-hop → dropped
                header("x-evil", "a\r\nInjected: pwned"), // CRLF in value → dropped
                header("bad name", "x"),                // space in name → invalid → dropped
                header("", "x"),                        // empty name → invalid → dropped
            ],
        );

        assert_eq!(
            dst.get("x-ok").and_then(|v| v.to_str().ok()),
            Some("fine"),
            "a valid end-to-end header is copied"
        );
        assert!(
            !dst.contains_key("transfer-encoding"),
            "a filter cannot re-introduce a hop-by-hop header"
        );
        assert!(
            !dst.contains_key("x-evil") && !dst.contains_key("injected"),
            "a CRLF-bearing value is rejected, not split into a second header"
        );
        assert_eq!(dst.len(), 1, "only the one valid header survives");
    }
}