plecto-server 0.14.1

//! 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 std::str::FromStr;

use hyper::header::{HeaderName, HeaderValue};
use hyper::http::Uri;
use hyper::http::Version;
use hyper::http::uri::{Authority, Scheme};
use plecto_control::{Header, HttpRequest, TrustedProxyTrust};

/// 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
/// (`X-Real-IP`, `True-Client-IP`, `CF-Connecting-IP`, `Fastly-Client-IP`, `X-Client-IP`,
/// `X-Cluster-Client-IP` — CDN-issued conventions, none of them specified by an RFC). 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",
];

/// The most `X-Forwarded-For` elements the trusted-proxy path reads before giving up (ADR 000103).
/// A real chain is one or two hops, so the bound only ever bites on a client-stuffed list — where
/// falling back to the peer is the right answer anyway. It keeps the one untrusted parse surface
/// this path opens bounded regardless of what a client sends.
const MAX_FORWARDED_ELEMENTS: usize = 32;

/// The client address a trusted front proxy named in `X-Forwarded-For` (ADR 000103), or `None`
/// when the header is absent, malformed, or names only declared hops.
///
/// Read right to left: each element was appended by the hop that received the request, so the
/// rightmost address outside `trusted` is the first one no declared proxy vouched for. Anything
/// the scan cannot classify — a non-UTF-8 value, an element that is not a bare IP, more elements
/// than the bound — returns `None`, and the caller falls back to the peer (the side a client
/// cannot forge).
fn forwarded_client(headers: &hyper::HeaderMap, trusted: &TrustedProxyTrust) -> Option<IpAddr> {
    let mut budget = MAX_FORWARDED_ELEMENTS;
    // A list split across several header lines is one list (RFC 9110 §5.3), newest hop last.
    for value in headers.get_all("x-forwarded-for").iter().rev() {
        for element in value.to_str().ok()?.rsplit(',') {
            budget = budget.checked_sub(1)?;
            let ip: IpAddr = element.trim().parse().ok()?;
            if !trusted.contains(ip) {
                return Some(ip.to_canonical());
            }
        }
    }
    None
}

/// 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 client) and
/// `X-Forwarded-Proto` (the wire scheme) afresh, and return the client this transaction is
/// attributed to. `X-Real-IP` is re-issued — not just stripped — so a backend reading that
/// de-facto convention rather than `XFF` still gets Plecto's authoritative value (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.
///
/// `trusted` is the declared `[listen.trusted_proxy]` set (ADR 000103), and `Some` only changes
/// WHICH address is issued: when `peer` is itself one of the declared front proxies, the inbound
/// `X-Forwarded-For` may name the client and is read (see [`forwarded_client`]); otherwise the
/// overwrite is unconditional, exactly as before. `peer` is the already-resolved client candidate
/// — PROXY v2 (ADR 000057) has run by the time this is called — so one rule covers both fronting
/// shapes: an L4 front restores the client below HTTP and its restored address is (correctly) not
/// a declared proxy, an L7 front leaves its own address and is. `X-Forwarded-Proto` is never taken
/// from the inbound request: the scheme stays the wire truth.
///
/// 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,
    trusted: Option<&TrustedProxyTrust>,
) -> IpAddr {
    // An IPv4 client on a dual-stack ([::]) listener arrives as an IPv4-mapped IPv6 address
    // (`::ffff:a.b.c.d`); `to_canonical` normalises it to dotted IPv4 so backends/filters that
    // all-list on the IPv4 form match — the same rule the trust containment check applies. A
    // genuine IPv6 address is kept verbatim (no brackets — XFF carries a bare address).
    let client = trusted
        .filter(|trusted| trusted.contains(peer))
        .and_then(|trusted| forwarded_client(headers, trusted))
        .unwrap_or_else(|| peer.to_canonical());

    for name in FORWARDED_HEADERS {
        // `HeaderName` lookups are case-insensitive; the constants are already lowercase.
        headers.remove(*name);
    }
    // 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.to_string()) {
        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);
    }
    client
}

/// Reject inbound authority ambiguity and leave a canonical `Host` header whenever the request
/// names an authority. This runs once at the transport-neutral proxy ingress; later filter edits
/// remain deliberate policy actions (ADR 000015).
///
/// HTTP/1.1 requires exactly one `Host`. HTTP/2 and HTTP/3 use URI authority when it is available;
/// a literal `Host`, when present, must name the same endpoint, and is the fallback when an
/// adapter exposes origin-form URI data. HTTP/1.0 retains compatibility with origin-form requests
/// that have neither field. For an absolute-form request, the URI is the canonical representation
/// retained in `Host`, as RFC 9112 §3.2.2 requires for forwarding.
pub(crate) fn normalize_ingress_authority(
    parts: &mut hyper::http::request::Parts,
    wire_scheme: &str,
) -> Result<(), ()> {
    let uri_authority = parts
        .uri
        .authority()
        .map(|authority| parse_authority(authority.as_str()))
        .transpose()?;

    let mut hosts = parts.headers.get_all(hyper::header::HOST).iter();
    let host_authority = match hosts.next() {
        Some(value) => Some(parse_authority(value.to_str().map_err(|_| ())?)?),
        None => None,
    };
    if hosts.next().is_some() {
        return Err(());
    }

    let scheme = parts.uri.scheme_str().unwrap_or(wire_scheme);
    let transport_host_fallback = !matches!(parts.version, Version::HTTP_10 | Version::HTTP_11)
        && uri_authority.is_none()
        && host_authority.is_some();
    let selected = match parts.version {
        Version::HTTP_10 => match (uri_authority.as_deref(), host_authority.as_deref()) {
            (Some(uri), Some(host)) if !authorities_match(uri, host, scheme) => return Err(()),
            (Some(uri), _) => Some(uri),
            (None, Some(host)) => Some(host),
            (None, None) => None,
        },
        Version::HTTP_11 => {
            let host = host_authority.as_deref().ok_or(())?;
            match uri_authority.as_deref() {
                Some(uri) if !authorities_match(uri, host, scheme) => return Err(()),
                Some(uri) => Some(uri),
                None => Some(host),
            }
        }
        _ => match (uri_authority.as_deref(), host_authority.as_deref()) {
            (Some(uri), Some(host)) if !authorities_match(uri, host, scheme) => return Err(()),
            (Some(uri), _) => Some(uri),
            // RFC 9113 §8.3.1 requires direct h2 clients to send `:authority` when they have
            // authority information, but an adapter can expose a valid origin-form request with
            // the corresponding single Host field. Treat it as the one authority rather than
            // inventing a second interpretation; H3 shares this transport-neutral path.
            (None, Some(host)) => Some(host),
            (None, None) => return Err(()),
        },
    };

    match selected {
        Some(authority) => {
            let value = HeaderValue::from_str(authority).map_err(|_| ())?;
            parts.headers.insert(hyper::header::HOST, value);
            // A transport adapter can expose authority solely as Host beside an origin-form Uri.
            // Materialize that same validated authority in the Uri so every later consumer reads
            // one canonical value.
            if transport_host_fallback {
                let mut uri_parts = parts.uri.clone().into_parts();
                uri_parts.scheme = Some(Scheme::from_str(wire_scheme).map_err(|_| ())?);
                uri_parts.authority = Some(Authority::from_str(authority).map_err(|_| ())?);
                parts.uri = Uri::from_parts(uri_parts).map_err(|_| ())?;
            }
        }
        None => {
            parts.headers.remove(hyper::header::HOST);
        }
    }
    Ok(())
}

fn parse_authority(value: &str) -> Result<String, ()> {
    // URI authority permits userinfo, but Host field values do not (RFC 9110 §7.2).
    if value.is_empty() || value.contains('@') {
        return Err(());
    }
    let authority = Authority::from_str(value).map_err(|_| ())?;
    // `Authority` permits a non-numeric or out-of-range port as generic URI syntax, while an
    // HTTP Host field's port is decimal `port` (RFC 3986 §3.2.3). Do not collapse either into
    // the same `None` as an absent port during equivalence checks.
    if !has_valid_port_syntax(value)
        || (authority.port().is_some() && authority.port_u16().is_none())
    {
        return Err(());
    }
    Ok(authority.to_string())
}

fn has_valid_port_syntax(authority: &str) -> bool {
    let port = if let Some(bracketed) = authority.strip_prefix('[') {
        let Some((_, suffix)) = bracketed.split_once(']') else {
            return false;
        };
        match suffix.strip_prefix(':') {
            Some(port) => port,
            None if suffix.is_empty() => return true,
            None => return false,
        }
    } else {
        match authority.rsplit_once(':') {
            Some((_, port)) => port,
            None => return true,
        }
    };
    !port.is_empty() && port.parse::<u16>().is_ok()
}

fn authorities_match(left: &str, right: &str, scheme: &str) -> bool {
    let Ok(left) = Authority::from_str(left) else {
        return false;
    };
    let Ok(right) = Authority::from_str(right) else {
        return false;
    };
    let left_host = left.host().strip_suffix('.').unwrap_or(left.host());
    let right_host = right.host().strip_suffix('.').unwrap_or(right.host());
    if !left_host.eq_ignore_ascii_case(right_host) {
        return false;
    }
    match (left.port_u16(), right.port_u16()) {
        (Some(left), Some(right)) => left == right,
        (None, None) => true,
        (Some(port), None) | (None, Some(port)) => {
            (scheme.eq_ignore_ascii_case("http") && port == 80)
                || (scheme.eq_ignore_ascii_case("https") && port == 443)
        }
    }
}

/// 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_with_query: 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 request_parts(version: Version, uri: &str, hosts: &[&str]) -> hyper::http::request::Parts {
        let mut builder = Request::builder().method("GET").version(version).uri(uri);
        for host in hosts {
            builder = builder.header(hyper::header::HOST, *host);
        }
        builder.body(()).unwrap().into_parts().0
    }

    #[test]
    fn ingress_authority_rejects_ambiguous_or_invalid_http11_host() {
        for (uri, hosts) in [
            ("http://public.example/", vec!["protected.example"]),
            ("/", vec!["public.example", "public.example"]),
            ("/", vec![]),
            ("/", vec!["user@public.example"]),
            ("/", vec!["public.example:abc"]),
            ("/", vec!["public.example:65536"]),
            ("/", vec!["public.example:"]),
        ] {
            let mut parts = request_parts(Version::HTTP_11, uri, &hosts);
            assert!(
                normalize_ingress_authority(&mut parts, "http").is_err(),
                "{uri:?} with {hosts:?} must be rejected"
            );
        }
    }

    #[test]
    fn ingress_authority_canonicalizes_matching_authorities_for_all_transports() {
        let cases = [
            (
                Version::HTTP_11,
                "http://EXAMPLE.test./",
                "example.test:80",
                "EXAMPLE.test.",
            ),
            (
                Version::HTTP_2,
                "https://[2001:db8::1]/",
                "[2001:DB8::1]:443",
                "[2001:db8::1]",
            ),
        ];
        for (version, uri, host, want) in cases {
            let mut parts = request_parts(version, uri, &[host]);
            normalize_ingress_authority(&mut parts, "https").unwrap();
            assert_eq!(
                parts
                    .headers
                    .get(hyper::header::HOST)
                    .and_then(|v| v.to_str().ok()),
                Some(want),
                "{uri:?} must replace its matching literal Host with the URI authority"
            );
        }

        let mut h2_without_host = request_parts(Version::HTTP_2, "https://h2.example/", &[]);
        normalize_ingress_authority(&mut h2_without_host, "https").unwrap();
        assert_eq!(
            h2_without_host
                .headers
                .get(hyper::header::HOST)
                .and_then(|v| v.to_str().ok()),
            Some("h2.example")
        );

        for version in [Version::HTTP_2, Version::HTTP_3] {
            let mut origin_form = request_parts(version, "/from-adapter", &["h2.example"]);
            normalize_ingress_authority(&mut origin_form, "https").unwrap();
            assert_eq!(
                origin_form
                    .headers
                    .get(hyper::header::HOST)
                    .and_then(|v| v.to_str().ok()),
                Some("h2.example"),
                "a single Host remains the authority when {version:?} adapter URI is origin-form"
            );
            assert_eq!(
                origin_form.uri.authority().map(Authority::as_str),
                Some("h2.example")
            );
        }
    }

    #[test]
    fn ingress_authority_rejects_non_default_port_conflicts_and_preserves_http10_compatibility() {
        let mut non_default = request_parts(
            Version::HTTP_11,
            "http://example.test/",
            &["example.test:8080"],
        );
        assert!(normalize_ingress_authority(&mut non_default, "http").is_err());

        let mut legacy = request_parts(Version::HTTP_10, "/", &[]);
        normalize_ingress_authority(&mut legacy, "http").unwrap();
        assert!(legacy.headers.get(hyper::header::HOST).is_none());
        assert_eq!(request_authority(&legacy), "");
    }

    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 hop_by_hop_list_matches_the_host_contract_strip_list() {
        // Two crates carry this list — the egress strip here and the guest-output drop in
        // plecto-host's contract mappers. Neither can depend on the other for the constant
        // (fast path / extension plane separation), so a drift between them would silently
        // open a smuggling-adjacent gap (CWE-444): a name one side strips and the other
        // forwards. This assertion is the only guard.
        assert_eq!(HOP_BY_HOP, plecto_host::HOP_BY_HOP_GUEST_HEADERS);
    }

    #[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", None);

        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",
            None,
        );
        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", None);
        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"
        );
    }

    /// The parsed `[listen.trusted_proxy]` trust for `cidrs`, built through the manifest — the
    /// one parse entrypoint, so these tests share the CIDR canonicalisation the fast path uses.
    fn trusted(cidrs: &str) -> plecto_control::TrustedProxyTrust {
        plecto_control::Manifest::from_toml(&format!(
            "[listen.trusted_proxy]\ntrusted = [{cidrs}]\n"
        ))
        .unwrap()
        .listen
        .trusted_proxy_trust()
        .unwrap()
        .expect("the section is declared")
    }

    fn xff(headers: &hyper::HeaderMap) -> Option<&str> {
        headers.get("x-forwarded-for").and_then(|v| v.to_str().ok())
    }

    #[test]
    fn set_forwarded_restores_the_client_named_by_a_trusted_proxy() {
        // ADR 000103: a peer inside the declared CIDRs may name the client. The list is read
        // right to left — each element was appended by the hop that received the request — so
        // the first address no declared proxy vouched for is the client.
        let trust = trusted(r#""10.0.0.0/8", "192.0.2.0/24""#);
        let mut headers = hyper::HeaderMap::new();
        headers.insert(
            "x-forwarded-for",
            HeaderValue::from_static("198.51.100.7, 192.0.2.5, 10.0.0.1"),
        );
        headers.insert("x-real-ip", HeaderValue::from_static("9.9.9.9"));

        let client = set_forwarded(
            &mut headers,
            "10.0.0.1".parse().unwrap(),
            "https",
            Some(&trust),
        );

        assert_eq!(client, "198.51.100.7".parse::<IpAddr>().unwrap());
        assert_eq!(
            xff(&headers),
            Some("198.51.100.7"),
            "the re-issued XFF carries the restored client, not the inbound list"
        );
        assert_eq!(
            headers.get("x-real-ip").and_then(|v| v.to_str().ok()),
            Some("198.51.100.7"),
            "X-Real-IP is re-issued from the same restored client (ADR 000022 stays)"
        );
    }

    #[test]
    fn set_forwarded_reads_repeated_forwarded_headers_as_one_list() {
        // A list split across several header lines is one list (RFC 9110 §5.3): the rightmost
        // element of the LAST line is the newest hop, so the scan starts there.
        let trust = trusted(r#""10.0.0.0/8""#);
        let mut headers = hyper::HeaderMap::new();
        headers.append("x-forwarded-for", HeaderValue::from_static("203.0.113.9"));
        headers.append(
            "x-forwarded-for",
            HeaderValue::from_static("198.51.100.7, 10.0.0.2"),
        );

        let client = set_forwarded(
            &mut headers,
            "10.0.0.1".parse().unwrap(),
            "http",
            Some(&trust),
        );

        assert_eq!(client, "198.51.100.7".parse::<IpAddr>().unwrap());
        assert_eq!(xff(&headers), Some("198.51.100.7"));
    }

    #[test]
    fn set_forwarded_falls_back_to_the_peer_when_every_element_is_trusted() {
        // Nothing outside the declared set means nobody named a client — fall back to the peer,
        // the side that cannot be forged (ADR 000103 decision 5).
        let trust = trusted(r#""10.0.0.0/8""#);
        let mut headers = hyper::HeaderMap::new();
        headers.insert(
            "x-forwarded-for",
            HeaderValue::from_static("10.0.0.9, 10.0.0.1"),
        );

        let client = set_forwarded(
            &mut headers,
            "10.0.0.1".parse().unwrap(),
            "http",
            Some(&trust),
        );

        assert_eq!(client, "10.0.0.1".parse::<IpAddr>().unwrap());
        assert_eq!(xff(&headers), Some("10.0.0.1"));
    }

    #[test]
    fn set_forwarded_falls_back_to_the_peer_on_a_malformed_or_absent_list() {
        // Untrusted input on the trusted path: an unparseable element, an empty value, a bare
        // trailing separator, a port-suffixed element, and a missing header all fall back to the
        // peer rather than guessing — and none of them panics (data-plane discipline).
        let trust = trusted(r#""10.0.0.0/8""#);
        for value in [
            Some("not-an-ip"),
            Some(""),
            Some("198.51.100.7,"),
            Some("198.51.100.7, , 10.0.0.1"),
            Some("198.51.100.7:443"),
            Some("for=198.51.100.7"),
            None,
        ] {
            let mut headers = hyper::HeaderMap::new();
            if let Some(value) = value {
                headers.insert("x-forwarded-for", HeaderValue::from_str(value).unwrap());
            }
            let client = set_forwarded(
                &mut headers,
                "10.0.0.1".parse().unwrap(),
                "http",
                Some(&trust),
            );
            assert_eq!(
                client,
                "10.0.0.1".parse::<IpAddr>().unwrap(),
                "{value:?} must fall back to the peer"
            );
            assert_eq!(xff(&headers), Some("10.0.0.1"), "{value:?}");
        }
    }

    #[test]
    fn set_forwarded_bounds_the_elements_it_scans() {
        // The one untrusted parse surface the trusted path opens is bounded (ADR 000103): a
        // stuffed list runs the scan out of budget, which falls back to the peer.
        let trust = trusted(r#""10.0.0.0/8""#);
        let stuffed = std::iter::repeat_n("10.0.0.1", MAX_FORWARDED_ELEMENTS + 1)
            .collect::<Vec<_>>()
            .join(",");
        let mut headers = hyper::HeaderMap::new();
        headers.insert(
            "x-forwarded-for",
            HeaderValue::from_str(&format!("198.51.100.7,{stuffed}")).unwrap(),
        );

        let client = set_forwarded(
            &mut headers,
            "10.0.0.1".parse().unwrap(),
            "http",
            Some(&trust),
        );

        assert_eq!(client, "10.0.0.1".parse::<IpAddr>().unwrap());
    }

    #[test]
    fn set_forwarded_ignores_x_forwarded_for_from_an_untrusted_peer() {
        // The security half of ADR 000103: the declaration decides who may speak. A peer outside
        // the CIDRs gets the unchanged edge behaviour — the forged list is dropped, the peer
        // re-issued — no matter how plausible the claim looks.
        let trust = trusted(r#""10.0.0.0/8""#);
        for trust_arg in [Some(&trust), None] {
            let mut headers = hyper::HeaderMap::new();
            headers.insert(
                "x-forwarded-for",
                HeaderValue::from_static("198.51.100.7, 10.0.0.1"),
            );
            headers.insert("x-real-ip", HeaderValue::from_static("198.51.100.7"));

            let client = set_forwarded(
                &mut headers,
                "203.0.113.5".parse().unwrap(),
                "https",
                trust_arg,
            );

            assert_eq!(client, "203.0.113.5".parse::<IpAddr>().unwrap());
            assert_eq!(
                xff(&headers),
                Some("203.0.113.5"),
                "an untrusted peer cannot name its own client"
            );
            assert_eq!(
                headers.get("x-real-ip").and_then(|v| v.to_str().ok()),
                Some("203.0.113.5")
            );
        }
    }

    #[test]
    fn set_forwarded_normalises_an_ipv4_mapped_restored_client() {
        // The same canonicalisation the peer gets (ADR 000022) applies to a restored client, and
        // a v4 CIDR matches an IPv4-mapped hop — otherwise a dual-stack chain would silently
        // stop matching the declared set.
        let trust = trusted(r#""10.0.0.0/8""#);
        let mut headers = hyper::HeaderMap::new();
        headers.insert(
            "x-forwarded-for",
            HeaderValue::from_static("::ffff:198.51.100.7, ::ffff:10.0.0.2"),
        );

        let client = set_forwarded(
            &mut headers,
            "::ffff:10.0.0.1".parse().unwrap(),
            "https",
            Some(&trust),
        );

        assert_eq!(client, "198.51.100.7".parse::<IpAddr>().unwrap());
        assert_eq!(xff(&headers), Some("198.51.100.7"));
    }

    #[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");
    }
}