flowscope 0.24.1

//! [`HttpProxyParser`] — the sans-IO streaming front-end, for inline
//! proxies. See the type's own documentation for the event model and
//! the forwarding contract.

use bytes::Bytes;

use super::{
    engine::{Dir, Engine, EngineEvent, EngineLimits},
    poison::HttpPoison,
    types::{RequestHead, ResponseHead, SmugglingPolicy, SwitchKind},
};
use crate::FlowSide;

/// One step of progress on a connection.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum HttpEvent {
    /// Request start line + headers, before the body.
    RequestHead(RequestHead),
    /// Response status line + headers, before the body. Emitted for
    /// every `1xx` interim as well as the final response.
    ResponseHead(ResponseHead),
    /// Body progress. `data` is decoded payload — empty when the step
    /// consumed only framing bytes, such as a chunk-size line. `raw`
    /// is every byte the step consumed, framing included.
    Body {
        dir: FlowSide,
        data: Bytes,
        raw: Bytes,
    },
    /// Trailer section after a chunked body. Emitted for every chunked
    /// message, possibly with no fields.
    Trailers {
        dir: FlowSide,
        trailers: Vec<(Bytes, Bytes)>,
        raw: Bytes,
    },
    /// One message is fully framed. On a keep-alive connection the
    /// next one may follow immediately.
    End { dir: FlowSide },
    /// The connection left HTTP behind; tunnel the rest verbatim.
    SwitchProtocols { kind: SwitchKind },
}

/// Caps for [`HttpProxyParser`].
///
/// Every field bounds something an adversary controls. The struct is
/// `#[non_exhaustive]` so it can gain knobs without breaking you,
/// which means starting from [`Default`] and assigning the fields you
/// care about:
///
/// ```
/// use flowscope::http::HttpProxyConfig;
///
/// let cfg = HttpProxyConfig::default()
///     .with_max_head_bytes(16 * 1024)
///     .with_max_pipelined(16);
/// ```
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct HttpProxyConfig {
    /// Start line + header block. Default 64 KiB.
    pub max_head_bytes: usize,
    /// Header fields per message. Default 128.
    pub max_headers: usize,
    /// One `<hex-size>[;ext]\r\n` line. Default 256 B.
    pub max_chunk_line_bytes: usize,
    /// The whole trailer section. Default 8 KiB.
    pub max_trailer_bytes: usize,
    /// Requests that may be outstanding at once. Default 64.
    pub max_pipelined: usize,
    /// What to do about an ambiguously framed message. Defaults to
    /// [`SmugglingPolicy::Strict`] — for a proxy, refusing to forward
    /// is the only response to an ambiguity that cannot be exploited.
    pub smuggling: SmugglingPolicy,
    /// How much unparsed data one direction may hold before
    /// [`push`](HttpProxyParser::push) starts refusing bytes. This is
    /// the backpressure valve, not a failure: it must be at least
    /// `max_head_bytes` for the parser to make progress. Default
    /// 256 KiB.
    pub max_buffered_bytes: usize,
}

impl HttpProxyConfig {
    /// Cap the start line + header block.
    #[must_use]
    pub fn with_max_head_bytes(mut self, n: usize) -> Self {
        self.max_head_bytes = n;
        self
    }

    /// Cap the header-field count per message.
    #[must_use]
    pub fn with_max_headers(mut self, n: usize) -> Self {
        self.max_headers = n;
        self
    }

    /// Cap one chunk-size line.
    #[must_use]
    pub fn with_max_chunk_line_bytes(mut self, n: usize) -> Self {
        self.max_chunk_line_bytes = n;
        self
    }

    /// Cap the trailer section.
    #[must_use]
    pub fn with_max_trailer_bytes(mut self, n: usize) -> Self {
        self.max_trailer_bytes = n;
        self
    }

    /// Cap how many requests may be outstanding at once.
    #[must_use]
    pub fn with_max_pipelined(mut self, n: usize) -> Self {
        self.max_pipelined = n;
        self
    }

    /// Set the ambiguous-framing policy.
    #[must_use]
    pub fn with_smuggling_policy(mut self, policy: SmugglingPolicy) -> Self {
        self.smuggling = policy;
        self
    }

    /// Cap unparsed data per direction — the backpressure threshold.
    #[must_use]
    pub fn with_max_buffered_bytes(mut self, n: usize) -> Self {
        self.max_buffered_bytes = n;
        self
    }
}

impl Default for HttpProxyConfig {
    fn default() -> Self {
        Self {
            max_head_bytes: 64 * 1024,
            max_headers: 128,
            max_chunk_line_bytes: 256,
            max_trailer_bytes: 8 * 1024,
            max_pipelined: 64,
            smuggling: SmugglingPolicy::Strict,
            max_buffered_bytes: 256 * 1024,
        }
    }
}

/// Sans-IO HTTP/1.1 parser for inline proxies.
///
/// Feed bytes per direction, drain events. No sockets, no async, no
/// clock: the caller owns I/O, and therefore owns backpressure — the
/// same discipline `rustls`, `quinn-proto`, and `h2` use. One parser
/// owns both directions of a connection, because response framing
/// depends on the matching request's method (RFC 9112 §6.3 rules
/// 1–2).
///
/// ```
/// use bytes::Bytes;
/// use flowscope::FlowSide;
/// use flowscope::http::{HttpEvent, HttpProxyParser};
///
/// let mut p = HttpProxyParser::new();
/// let wire = Bytes::from_static(
///     b"POST /submit HTTP/1.1\r\nHost: api.example.com\r\nContent-Length: 5\r\n\r\nhello",
/// );
/// let accepted = p.push(FlowSide::Initiator, &wire);
/// assert_eq!(accepted, wire.len());
///
/// // The head arrives before any body byte is looked at, so a proxy
/// // can pick a backend immediately.
/// let Some(HttpEvent::RequestHead(head)) = p.next_event() else {
///     panic!("expected a request head")
/// };
/// assert_eq!(head.host(), Some("api.example.com"));
///
/// // Body bytes are reported as spans the parser does not retain.
/// let mut body = Vec::new();
/// while let Some(ev) = p.next_event() {
///     match ev {
///         HttpEvent::Body { data, .. } => body.extend_from_slice(&data),
///         HttpEvent::End { .. } => break,
///         _ => {}
///     }
/// }
/// assert_eq!(body, b"hello");
/// ```
///
/// # Forwarding verbatim
///
/// Every event carries the exact wire bytes it consumed. Concatenating
/// [`RequestHead::raw`], each [`HttpEvent::Body`]'s `raw`, and
/// [`HttpEvent::Trailers`]' `raw` reproduces the message byte for
/// byte — so a proxy that forwards traffic unchanged relays `raw`
/// while flowscope tracks boundaries, and a proxy that rewrites
/// bodies uses the decoded `data` and re-frames.
///
/// # Bounded memory
///
/// The parser never buffers a whole body. Everything it does hold —
/// the head block, a chunk-size line, the trailer section — is capped
/// by [`HttpProxyConfig`], and breaching a cap poisons the connection
/// instead of growing. When a direction is holding
/// [`max_buffered_bytes`](HttpProxyConfig::max_buffered_bytes),
/// [`push`](Self::push) accepts fewer bytes than offered: that short
/// count is the backpressure signal, and the caller should stop
/// reading its socket until it has drained more events.
#[derive(Debug, Clone)]
pub struct HttpProxyParser {
    engine: Engine,
    config: HttpProxyConfig,
}

impl Default for HttpProxyParser {
    fn default() -> Self {
        Self::new()
    }
}

impl HttpProxyParser {
    /// Construct with default caps.
    pub fn new() -> Self {
        Self::with_config(HttpProxyConfig::default())
    }

    /// Construct with explicit caps.
    pub fn with_config(config: HttpProxyConfig) -> Self {
        let limits = EngineLimits {
            max_head_bytes: config.max_head_bytes,
            max_headers: config.max_headers,
            max_chunk_line_bytes: config.max_chunk_line_bytes,
            max_trailer_bytes: config.max_trailer_bytes,
            max_pipelined: config.max_pipelined,
            policy: config.smuggling,
        };
        Self {
            engine: Engine::new(limits),
            config,
        }
    }

    /// Feed bytes for one direction; returns how many were accepted.
    ///
    /// A short count is the backpressure signal: the direction is
    /// holding as much unparsed data as
    /// [`max_buffered_bytes`](HttpProxyConfig::max_buffered_bytes)
    /// allows. Drain events, then re-offer the remainder with
    /// `data.slice(accepted..)` — slicing a [`Bytes`] is a refcount
    /// bump, not a copy.
    ///
    /// On a zero return, check the state before deciding: a poisoned
    /// connection ([`is_poisoned`](Self::is_poisoned)) or a tunnelled
    /// one ([`is_tunnelled`](Self::is_tunnelled)) will accept nothing
    /// more — post-switch bytes are the caller's to splice, starting
    /// with [`take_tunnel_residue`](Self::take_tunnel_residue).
    /// Otherwise a zero is backpressure like any short count: drain
    /// events and re-offer. (Before 0.24 a tunnelled parser reported
    /// pushed bytes as accepted while silently dropping them.)
    pub fn push(&mut self, dir: FlowSide, data: &Bytes) -> usize {
        if self.is_poisoned() || self.engine.is_tunnelled() {
            return 0;
        }
        let dir = to_dir(dir);
        let room = self
            .config
            .max_buffered_bytes
            .saturating_sub(self.engine.buffered(dir));
        let take = room.min(data.len());
        if take > 0 {
            self.engine.push(dir, &data[..take]);
        }
        take
    }

    /// Signal end of stream on a direction.
    ///
    /// A close-delimited body ends here; anything else simply closes.
    /// A FIN between messages is normal and is not a failure.
    pub fn fin(&mut self, dir: FlowSide) {
        self.engine.fin(to_dir(dir));
    }

    /// Take the next event, if one is ready.
    ///
    /// `None` means more bytes are needed (or the connection is
    /// finished). Events are produced lazily — this drives the parser,
    /// so keep calling it until it returns `None`.
    pub fn next_event(&mut self) -> Option<HttpEvent> {
        // Requests first: a response cannot be framed until its
        // request's method is known.
        for dir in [Dir::Request, Dir::Response] {
            match self.engine.poll(dir) {
                Ok(Some(ev)) => {
                    if matches!(ev, EngineEvent::Head(_)) {
                        crate::obs::record_http_message(dir == Dir::Request);
                    }
                    return Some(convert(dir, ev));
                }
                Ok(None) => continue,
                Err(_) => {
                    // The direction just poisoned; count it once, on
                    // the transition, and report why through
                    // is_poisoned() / poison().
                    if let Some(reason) = self.engine.poison(dir) {
                        crate::obs::record_http_poisoned(reason);
                    }
                    continue;
                }
            }
        }
        None
    }

    /// `true` once framing was lost and parsing has stopped.
    ///
    /// An inline proxy should tear the connection down rather than
    /// keep forwarding: past a framing desync, the two endpoints no
    /// longer agree on where messages end, which is exactly the
    /// condition request smuggling exploits.
    pub fn is_poisoned(&self) -> bool {
        self.engine.is_desynced(Dir::Request) || self.engine.is_desynced(Dir::Response)
    }

    /// Why parsing stopped, if it did.
    pub fn poison(&self) -> Option<HttpPoison> {
        self.engine
            .poison(Dir::Request)
            .or_else(|| self.engine.poison(Dir::Response))
    }

    /// Stable slug for [`poison`](Self::poison).
    pub fn poison_reason(&self) -> Option<&'static str> {
        self.poison().map(HttpPoison::as_str)
    }

    /// `true` once a protocol switch (`CONNECT` tunnel, `Upgrade`
    /// 101, h2 prior knowledge) ended HTTP parsing on this
    /// connection: the remaining bytes belong to the switched-to
    /// protocol and [`push`](Self::push) accepts nothing more.
    ///
    /// Distinguishes a tunnel zero-return from a backpressure short
    /// count; [`is_done`](Self::is_done) cannot, since it is also
    /// `true` once both directions closed.
    pub fn is_tunnelled(&self) -> bool {
        self.engine.is_tunnelled()
    }

    /// Take (once) the bytes that were already buffered for a
    /// direction when the connection switched protocols — the first
    /// bytes of the switched-to protocol, coalesced into the same
    /// segment as the message that triggered the switch (a `101` plus
    /// the first WebSocket frames, an h2 preface plus SETTINGS and
    /// HEADERS in one client write). A splicing proxy must forward
    /// these BEFORE relaying anything it reads after the
    /// [`HttpEvent::SwitchProtocols`] event, or the switched-to
    /// protocol desyncs. Empty when not tunnelled, or already taken.
    /// (Before 0.24.1 these bytes were silently destroyed at the
    /// switch.)
    pub fn take_tunnel_residue(&mut self, dir: FlowSide) -> Bytes {
        self.engine.take_residue(to_dir(dir))
    }

    /// `true` once no further HTTP message can arrive, so the caller
    /// can close or recycle the connection without guessing.
    ///
    /// That is the case after a protocol switch (the bytes belong to
    /// the tunnel now), and once both directions are finished —
    /// either by end of stream or because a message said the
    /// connection closes when it completes (`Connection: close`, or
    /// HTTP/1.0 without `keep-alive`).
    pub fn is_done(&self) -> bool {
        if self.engine.is_tunnelled() {
            return true;
        }
        self.engine.is_closed(Dir::Request) && self.engine.is_closed(Dir::Response)
    }

    /// Bytes held but not yet parsed on a direction. Bounded by
    /// [`max_buffered_bytes`](HttpProxyConfig::max_buffered_bytes) —
    /// a body never accumulates here.
    pub fn buffered(&self, dir: FlowSide) -> usize {
        self.engine.buffered(to_dir(dir))
    }
}

/// [`HttpProxyParser`] as a [`SessionParser`](crate::SessionParser),
/// so the streaming event stream can ride flowscope's own plumbing — the typed `Driver`, the
/// pcap replay helpers, the `emit` writers.
///
/// Use this when flowscope drives the bytes. Use [`HttpProxyParser`]
/// directly when *you* own the sockets: the trait has no way to
/// express a short read, so the backpressure signal
/// ([`push`](HttpProxyParser::push)'s accepted count) is not available
/// through this adapter, and bytes are copied once at the boundary
/// because the trait hands out `&[u8]`.
///
/// Unlike the passive [`HttpParser`](super::HttpParser), this reports
/// a framing failure as a poisoned parser. The driver responds by
/// dropping the parser and emitting
/// [`Event::ParserClosed`](crate::driver::Event::ParserClosed) with
/// [`EndReason::ParseError`](crate::EndReason::ParseError), rather
/// than continuing to feed a parser that has lost track of where
/// messages end. The TCP flow itself keeps going — closing the
/// connection is the caller's decision, since the caller owns the
/// socket.
#[derive(Debug, Clone)]
pub struct HttpProxySession {
    inner: HttpProxyParser,
}

impl Default for HttpProxySession {
    fn default() -> Self {
        Self::new()
    }
}

impl HttpProxySession {
    pub fn new() -> Self {
        Self {
            inner: HttpProxyParser::new(),
        }
    }

    pub fn with_config(config: HttpProxyConfig) -> Self {
        Self {
            inner: HttpProxyParser::with_config(config),
        }
    }

    /// The underlying parser, for state the trait does not expose.
    pub fn parser(&self) -> &HttpProxyParser {
        &self.inner
    }

    fn feed(&mut self, dir: FlowSide, bytes: &[u8], out: &mut Vec<HttpEvent>) {
        if bytes.is_empty() {
            return;
        }
        // The trait gives us a borrowed slice, so one copy here is
        // unavoidable; everything downstream is refcounted views of
        // it. Feed in a loop because the parser bounds how much it
        // will hold at once.
        let mut data = Bytes::copy_from_slice(bytes);
        loop {
            let accepted = self.inner.push(dir, &data);
            while let Some(ev) = self.inner.next_event() {
                out.push(ev);
            }
            if accepted == 0 || accepted == data.len() {
                break;
            }
            data = data.slice(accepted..);
        }
    }
}

impl crate::SessionParser for HttpProxySession {
    type Message = HttpEvent;

    fn feed_initiator(&mut self, bytes: &[u8], _ts: crate::Timestamp, out: &mut Vec<HttpEvent>) {
        self.feed(FlowSide::Initiator, bytes, out);
    }

    fn feed_responder(&mut self, bytes: &[u8], _ts: crate::Timestamp, out: &mut Vec<HttpEvent>) {
        self.feed(FlowSide::Responder, bytes, out);
    }

    fn fin_initiator(&mut self, out: &mut Vec<HttpEvent>) {
        self.inner.fin(FlowSide::Initiator);
        while let Some(ev) = self.inner.next_event() {
            out.push(ev);
        }
    }

    fn fin_responder(&mut self, out: &mut Vec<HttpEvent>) {
        self.inner.fin(FlowSide::Responder);
        while let Some(ev) = self.inner.next_event() {
            out.push(ev);
        }
    }

    fn rst_initiator(&mut self) {
        self.inner = HttpProxyParser::with_config(self.inner.config.clone());
    }

    fn rst_responder(&mut self) {
        self.inner = HttpProxyParser::with_config(self.inner.config.clone());
    }

    fn parser_kind(&self) -> crate::ParserKind {
        crate::ParserKind::Http1
    }

    fn is_poisoned(&self) -> bool {
        self.inner.is_poisoned()
    }

    fn poison_reason(&self) -> Option<&str> {
        self.inner.poison_reason()
    }

    fn is_done(&self) -> bool {
        self.inner.is_done()
    }
}

fn to_dir(side: FlowSide) -> Dir {
    match side {
        FlowSide::Initiator => Dir::Request,
        FlowSide::Responder => Dir::Response,
    }
}

fn to_side(dir: Dir) -> FlowSide {
    match dir {
        Dir::Request => FlowSide::Initiator,
        Dir::Response => FlowSide::Responder,
    }
}

fn convert(dir: Dir, ev: EngineEvent) -> HttpEvent {
    let side = to_side(dir);
    match ev {
        EngineEvent::Head(h) => match dir {
            Dir::Request => HttpEvent::RequestHead(RequestHead {
                method: h.method,
                path: h.path,
                version: h.version,
                headers: h.headers,
                framing: h.framing,
                applied: h.applied,
                raw: h.raw,
            }),
            Dir::Response => HttpEvent::ResponseHead(ResponseHead {
                status: h.status,
                reason: h.reason,
                version: h.version,
                headers: h.headers,
                framing: h.framing,
                interim: h.interim,
                applied: h.applied,
                raw: h.raw,
            }),
        },
        EngineEvent::Body { decoded, raw } => HttpEvent::Body {
            dir: side,
            data: decoded,
            raw,
        },
        EngineEvent::Trailers { fields, raw } => HttpEvent::Trailers {
            dir: side,
            trailers: fields,
            raw,
        },
        EngineEvent::End => HttpEvent::End { dir: side },
        EngineEvent::Switch(kind) => HttpEvent::SwitchProtocols { kind },
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::http::BodyFraming;

    fn push_all(p: &mut HttpProxyParser, dir: FlowSide, bytes: &[u8]) {
        let data = Bytes::copy_from_slice(bytes);
        let n = p.push(dir, &data);
        assert_eq!(n, data.len(), "test inputs fit the default caps");
    }

    fn drain(p: &mut HttpProxyParser) -> Vec<HttpEvent> {
        let mut out = Vec::new();
        while let Some(ev) = p.next_event() {
            out.push(ev);
        }
        out
    }

    /// Concatenated `raw` of every event, which must reproduce the
    /// wire bytes exactly.
    fn raw_of(evs: &[HttpEvent]) -> Vec<u8> {
        let mut v = Vec::new();
        for ev in evs {
            match ev {
                HttpEvent::RequestHead(h) => v.extend_from_slice(&h.raw),
                HttpEvent::ResponseHead(h) => v.extend_from_slice(&h.raw),
                HttpEvent::Body { raw, .. } => v.extend_from_slice(raw),
                HttpEvent::Trailers { raw, .. } => v.extend_from_slice(raw),
                _ => {}
            }
        }
        v
    }

    fn decoded_of(evs: &[HttpEvent]) -> Vec<u8> {
        let mut v = Vec::new();
        for ev in evs {
            if let HttpEvent::Body { data, .. } = ev {
                v.extend_from_slice(data);
            }
        }
        v
    }

    #[test]
    fn head_arrives_before_any_body_byte() {
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"POST /submit HTTP/1.1\r\nHost: api.example.com\r\nContent-Length: 1000\r\n\r\n",
        );
        let ev = p.next_event().expect("head must be ready");
        match ev {
            HttpEvent::RequestHead(h) => {
                assert_eq!(h.method.as_ref(), b"POST");
                assert_eq!(h.host(), Some("api.example.com"));
                assert_eq!(h.framing, BodyFraming::ContentLength(1000));
            }
            other => panic!("expected RequestHead, got {other:?}"),
        }
        assert!(p.next_event().is_none(), "no body has been fed yet");
    }

    #[test]
    fn body_is_never_accumulated() {
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"POST /u HTTP/1.1\r\nContent-Length: 100000\r\n\r\n",
        );
        let _ = drain(&mut p);
        // Push the whole body through in slices, draining as we go.
        let chunk = Bytes::from_static(&[b'x'; 4096]);
        let mut sent = 0usize;
        while sent < 100_000 {
            let slice = chunk.slice(..chunk.len().min(100_000 - sent));
            let n = p.push(FlowSide::Initiator, &slice);
            sent += n;
            let _ = drain(&mut p);
            assert!(
                p.buffered(FlowSide::Initiator) <= HttpProxyConfig::default().max_buffered_bytes,
                "the parser must never accumulate a body"
            );
        }
        assert_eq!(p.buffered(FlowSide::Initiator), 0);
    }

    #[test]
    fn raw_spans_reproduce_the_wire() {
        let wire: &[u8] = b"POST /u HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n\
                            5\r\nhello\r\n6\r\n world\r\n0\r\nX-Sum: 9\r\n\r\n";
        let mut p = HttpProxyParser::new();
        push_all(&mut p, FlowSide::Initiator, wire);
        let evs = drain(&mut p);
        assert_eq!(raw_of(&evs), wire);
        assert_eq!(decoded_of(&evs), b"hello world");

        let trailers: Vec<_> = evs
            .iter()
            .filter_map(|e| match e {
                HttpEvent::Trailers { trailers, .. } => Some(trailers.clone()),
                _ => None,
            })
            .collect();
        assert_eq!(trailers.len(), 1);
        assert_eq!(trailers[0][0].0.as_ref(), b"X-Sum");
        assert_eq!(trailers[0][0].1.as_ref(), b"9");
        assert!(matches!(evs.last(), Some(HttpEvent::End { .. })));
    }

    #[test]
    fn split_feeds_produce_the_same_events() {
        let wire: &[u8] = b"POST /u HTTP/1.1\r\nContent-Length: 11\r\n\r\nhello world\
                            GET /next HTTP/1.1\r\nHost: h\r\n\r\n";
        let mut whole = HttpProxyParser::new();
        push_all(&mut whole, FlowSide::Initiator, wire);
        let a = drain(&mut whole);

        let mut drip = HttpProxyParser::new();
        let mut b = Vec::new();
        for byte in wire {
            push_all(&mut drip, FlowSide::Initiator, std::slice::from_ref(byte));
            b.extend(drain(&mut drip));
        }
        assert_eq!(raw_of(&a), raw_of(&b));
        assert_eq!(decoded_of(&a), decoded_of(&b));
        assert_eq!(raw_of(&b), wire);
    }

    #[test]
    fn pipelined_requests_each_end_cleanly() {
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"POST /a HTTP/1.1\r\nContent-Length: 3\r\n\r\nAAA\
              POST /b HTTP/1.1\r\nContent-Length: 2\r\n\r\nBB\
              GET /c HTTP/1.1\r\n\r\n",
        );
        let evs = drain(&mut p);
        let paths: Vec<String> = evs
            .iter()
            .filter_map(|e| match e {
                HttpEvent::RequestHead(h) => Some(h.path_str().unwrap().to_string()),
                _ => None,
            })
            .collect();
        assert_eq!(paths, vec!["/a", "/b", "/c"]);
        let ends = evs
            .iter()
            .filter(|e| matches!(e, HttpEvent::End { .. }))
            .count();
        assert_eq!(ends, 3, "one End per message");
    }

    #[test]
    fn response_framing_uses_the_request_method() {
        let mut p = HttpProxyParser::new();
        push_all(&mut p, FlowSide::Initiator, b"HEAD /x HTTP/1.1\r\n\r\n");
        let _ = drain(&mut p);
        push_all(
            &mut p,
            FlowSide::Responder,
            b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n",
        );
        let evs = drain(&mut p);
        match &evs[0] {
            HttpEvent::ResponseHead(h) => assert_eq!(h.framing, BodyFraming::None),
            other => panic!("expected ResponseHead, got {other:?}"),
        }
        assert!(matches!(evs[1], HttpEvent::End { .. }));
    }

    #[test]
    fn backpressure_short_count_when_caller_does_not_drain() {
        let cfg = HttpProxyConfig {
            max_buffered_bytes: 4096,
            ..HttpProxyConfig::default()
        };
        let mut p = HttpProxyParser::with_config(cfg);
        let big = Bytes::from(vec![b'x'; 8192]);
        // Nothing has been parsed, so only the cap is accepted.
        let n = p.push(FlowSide::Initiator, &big);
        assert_eq!(n, 4096);
        // Re-offering the remainder is a refcount slice, not a copy.
        assert_eq!(p.push(FlowSide::Initiator, &big.slice(n..)), 0);
    }

    #[test]
    fn framing_desync_poisons_with_a_typed_reason() {
        let cfg = HttpProxyConfig {
            max_head_bytes: 128,
            ..HttpProxyConfig::default()
        };
        let mut p = HttpProxyParser::with_config(cfg);
        push_all(&mut p, FlowSide::Initiator, &[b'A'; 200]);
        let _ = drain(&mut p);
        assert!(p.is_poisoned());
        assert_eq!(p.poison(), Some(HttpPoison::HeadOverflow));
        assert_eq!(p.poison_reason(), Some("head-overflow"));
        // A poisoned connection accepts nothing further.
        assert_eq!(
            p.push(
                FlowSide::Initiator,
                &Bytes::from_static(b"GET / HTTP/1.1\r\n\r\n")
            ),
            0
        );
    }

    #[test]
    fn bad_chunk_size_is_reported_as_such() {
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"POST /u HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\nzz\r\n",
        );
        let _ = drain(&mut p);
        assert_eq!(p.poison(), Some(HttpPoison::InvalidChunkSize));
    }

    // ── interim responses, tunnels, completion (#162) ─────────────

    fn statuses(evs: &[HttpEvent]) -> Vec<(u16, bool)> {
        evs.iter()
            .filter_map(|e| match e {
                HttpEvent::ResponseHead(h) => Some((h.status, h.interim)),
                _ => None,
            })
            .collect()
    }

    #[test]
    fn interim_responses_precede_the_final_one() {
        // 100-continue then the real response. The interim must be
        // reported (a proxy forwards it) but must not complete the
        // exchange or consume the request.
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"POST /u HTTP/1.1\r\nExpect: 100-continue\r\nContent-Length: 2\r\n\r\n",
        );
        let _ = drain(&mut p);
        push_all(
            &mut p,
            FlowSide::Responder,
            b"HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok",
        );
        let evs = drain(&mut p);
        assert_eq!(statuses(&evs), vec![(100, true), (200, false)]);
        assert_eq!(decoded_of(&evs), b"ok");
        // Exactly one End: the interim did not complete anything.
        assert_eq!(
            evs.iter()
                .filter(|e| matches!(e, HttpEvent::End { .. }))
                .count(),
            1
        );
    }

    #[test]
    fn multiple_interims_then_final() {
        // 103 Early Hints (RFC 8297) can repeat before the final.
        let mut p = HttpProxyParser::new();
        push_all(&mut p, FlowSide::Initiator, b"GET /a HTTP/1.1\r\n\r\n");
        let _ = drain(&mut p);
        push_all(
            &mut p,
            FlowSide::Responder,
            b"HTTP/1.1 103 Early Hints\r\nLink: </s.css>\r\n\r\n\
              HTTP/1.1 103 Early Hints\r\nLink: </t.css>\r\n\r\n\
              HTTP/1.1 200 OK\r\nContent-Length: 1\r\n\r\nx",
        );
        let evs = drain(&mut p);
        assert_eq!(statuses(&evs), vec![(103, true), (103, true), (200, false)]);
    }

    #[test]
    fn response_reader_runs_before_the_request_body_is_sent() {
        // The 100-continue deadlock: the server answers while the
        // request body is still outstanding. Framing the response
        // must not wait on the request completing.
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"POST /u HTTP/1.1\r\nExpect: 100-continue\r\nContent-Length: 1000000\r\n\r\n",
        );
        let _ = drain(&mut p);
        push_all(
            &mut p,
            FlowSide::Responder,
            b"HTTP/1.1 100 Continue\r\n\r\n",
        );
        let evs = drain(&mut p);
        assert_eq!(statuses(&evs), vec![(100, true)]);
    }

    // 0.24.1: a half-close must not un-tunnel the connection — fin()
    // previously replaced Tunnel with Closed, after which push() went back
    // to accepting-and-dropping bytes (the exact bug the tunnelled-push
    // contract exists to prevent).
    #[test]
    fn fin_preserves_the_tunnel() {
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n\r\n",
        );
        let _ = drain(&mut p);
        push_all(
            &mut p,
            FlowSide::Responder,
            b"HTTP/1.1 200 Connection Established\r\n\r\n",
        );
        let _ = drain(&mut p);
        assert!(p.is_tunnelled());

        p.fin(FlowSide::Initiator);
        assert!(p.is_tunnelled(), "a tunnel outlives a half-close");
        assert!(p.is_done());
        let late = Bytes::from_static(b"tunnel bytes");
        assert_eq!(
            p.push(FlowSide::Responder, &late),
            0,
            "post-fin tunnel must still refuse bytes, not drop them"
        );
    }

    // 0.24.1: bytes coalesced into the same segment as the message that
    // triggers the switch are the switched-to protocol's first bytes — they
    // must be retrievable, not destroyed.
    #[test]
    fn tunnel_residue_after_upgrade_is_retrievable() {
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"GET /chat HTTP/1.1\r\nHost: h\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\n",
        );
        let _ = drain(&mut p);
        // The 101 and the first WebSocket frame arrive in ONE segment.
        push_all(
            &mut p,
            FlowSide::Responder,
            b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\n\x82\x05FRAME",
        );
        let evs = drain(&mut p);
        assert!(matches!(
            evs.last(),
            Some(HttpEvent::SwitchProtocols { .. })
        ));
        assert_eq!(
            p.take_tunnel_residue(FlowSide::Responder).as_ref(),
            b"\x82\x05FRAME",
            "the coalesced first frame must survive the switch"
        );
        assert!(
            p.take_tunnel_residue(FlowSide::Responder).is_empty(),
            "residue is taken once"
        );
        assert!(p.take_tunnel_residue(FlowSide::Initiator).is_empty());
    }

    #[test]
    fn tunnel_residue_h2_preface_is_byte_exact() {
        let mut p = HttpProxyParser::new();
        // Prior-knowledge h2: preface + SETTINGS in one client write. The
        // preface itself is never consumed as HTTP, so the residue is the
        // entire stream — a splicing proxy relays it verbatim.
        let wire: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n\x00\x00\x00\x04\x00\x00\x00\x00\x00";
        push_all(&mut p, FlowSide::Initiator, wire);
        let evs = drain(&mut p);
        assert!(matches!(
            evs.first(),
            Some(HttpEvent::SwitchProtocols {
                kind: SwitchKind::Http2PriorKnowledge
            })
        ));
        assert_eq!(p.take_tunnel_residue(FlowSide::Initiator).as_ref(), wire);
    }

    #[test]
    fn no_residue_without_a_switch() {
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"GET / HTTP/1.1\r\nHost: h\r\n\r\n",
        );
        let _ = drain(&mut p);
        assert!(p.take_tunnel_residue(FlowSide::Initiator).is_empty());
        assert!(p.take_tunnel_residue(FlowSide::Responder).is_empty());
    }

    #[test]
    fn connect_tunnel_switches_and_stops_parsing() {
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n\r\n",
        );
        let _ = drain(&mut p);
        push_all(
            &mut p,
            FlowSide::Responder,
            b"HTTP/1.1 200 Connection Established\r\n\r\n",
        );
        let evs = drain(&mut p);
        assert!(matches!(
            evs.last(),
            Some(HttpEvent::SwitchProtocols {
                kind: SwitchKind::ConnectTunnel
            })
        ));
        // Tunnelled bytes are the caller's to splice: push refuses
        // them (0, not a silent drop), and is_tunnelled — not
        // poisoned — says why.
        assert!(p.is_tunnelled());
        let refused = Bytes::from_static(b"\x16\x03\x01\x02\x00\x01");
        assert_eq!(p.push(FlowSide::Initiator, &refused), 0);
        assert!(p.next_event().is_none());
        assert!(!p.is_poisoned());
    }

    #[test]
    fn failed_connect_is_a_normal_response() {
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"CONNECT example.com:443 HTTP/1.1\r\n\r\n",
        );
        let _ = drain(&mut p);
        push_all(
            &mut p,
            FlowSide::Responder,
            b"HTTP/1.1 403 Forbidden\r\nContent-Length: 2\r\n\r\nno",
        );
        let evs = drain(&mut p);
        assert!(
            !evs.iter()
                .any(|e| matches!(e, HttpEvent::SwitchProtocols { .. })),
            "a rejected CONNECT does not open a tunnel"
        );
        assert_eq!(decoded_of(&evs), b"no");
    }

    #[test]
    fn websocket_upgrade_switches_with_the_protocol_token() {
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"GET /chat HTTP/1.1\r\nHost: h\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\n",
        );
        let _ = drain(&mut p);
        push_all(
            &mut p,
            FlowSide::Responder,
            b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\n",
        );
        let evs = drain(&mut p);
        match evs.last() {
            Some(HttpEvent::SwitchProtocols {
                kind: SwitchKind::Upgrade { protocol },
            }) => assert_eq!(protocol.as_ref(), b"websocket"),
            other => panic!("expected an Upgrade switch, got {other:?}"),
        }
        // The 101 head itself is still reported, so a proxy forwards it.
        assert_eq!(statuses(&evs), vec![(101, false)]);
    }

    #[test]
    fn http2_preface_is_recognised_not_mistaken_for_a_request() {
        let mut p = HttpProxyParser::new();
        push_all(
            &mut p,
            FlowSide::Initiator,
            b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n\x00\x00\x00\x04\x00\x00\x00\x00\x00",
        );
        let evs = drain(&mut p);
        assert!(matches!(
            evs.first(),
            Some(HttpEvent::SwitchProtocols {
                kind: SwitchKind::Http2PriorKnowledge
            })
        ));
        assert!(!p.is_poisoned(), "h2 traffic is not a framing failure");
    }

    #[test]
    fn partial_preface_waits_instead_of_guessing() {
        let mut p = HttpProxyParser::new();
        push_all(&mut p, FlowSide::Initiator, b"PRI * HTTP/2.0\r\n");
        assert!(
            p.next_event().is_none(),
            "a prefix of the preface must not decide either way"
        );
        push_all(&mut p, FlowSide::Initiator, b"\r\nSM\r\n\r\n");
        assert!(matches!(
            p.next_event(),
            Some(HttpEvent::SwitchProtocols {
                kind: SwitchKind::Http2PriorKnowledge
            })
        ));
    }

    #[test]
    fn connection_close_completes_the_direction() {
        let mut p = HttpProxyParser::new();
        push_all(&mut p, FlowSide::Initiator, b"GET /a HTTP/1.1\r\n\r\n");
        let _ = drain(&mut p);
        push_all(
            &mut p,
            FlowSide::Responder,
            b"HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 2\r\n\r\nhi",
        );
        let evs = drain(&mut p);
        assert_eq!(decoded_of(&evs), b"hi");
        p.fin(FlowSide::Initiator);
        assert!(
            p.is_done(),
            "after a Connection: close response the connection is finished"
        );
    }

    #[test]
    fn http_1_0_response_without_keep_alive_completes() {
        let mut p = HttpProxyParser::new();
        push_all(&mut p, FlowSide::Initiator, b"GET /a HTTP/1.0\r\n\r\n");
        let _ = drain(&mut p);
        push_all(
            &mut p,
            FlowSide::Responder,
            b"HTTP/1.0 200 OK\r\nContent-Length: 2\r\n\r\nhi",
        );
        let _ = drain(&mut p);
        p.fin(FlowSide::Initiator);
        assert!(p.is_done(), "HTTP/1.0 defaults to closing");
    }

    // ── SessionParser adapter (#164) ──────────────────────────────

    #[test]
    fn adapter_emits_the_same_events_through_the_trait() {
        use crate::SessionParser;
        let mut s = HttpProxySession::new();
        let mut out = Vec::new();
        s.feed_initiator(
            b"POST /a HTTP/1.1\r\nHost: h\r\nContent-Length: 5\r\n\r\nhello",
            crate::Timestamp::default(),
            &mut out,
        );
        assert!(matches!(out[0], HttpEvent::RequestHead(_)));
        assert_eq!(decoded_of(&out), b"hello");
        assert!(matches!(out.last(), Some(HttpEvent::End { .. })));
    }

    #[test]
    fn adapter_reports_poison_so_the_driver_ends_the_flow() {
        use crate::SessionParser;
        let mut s = HttpProxySession::new();
        let mut out = Vec::new();
        // Content-Length and Transfer-Encoding together.
        s.feed_initiator(
            b"POST /a HTTP/1.1\r\nContent-Length: 6\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\n",
            crate::Timestamp::default(),
            &mut out,
        );
        assert!(s.is_poisoned());
        assert_eq!(
            s.poison_reason(),
            Some("content-length-with-transfer-encoding")
        );
    }

    #[test]
    fn adapter_survives_a_feed_larger_than_the_buffer_cap() {
        // The trait hands over a whole slice at once with no way to
        // say "not yet", so the adapter has to loop internally
        // rather than silently dropping the tail.
        use crate::SessionParser;
        let cfg = HttpProxyConfig {
            max_buffered_bytes: 512,
            ..HttpProxyConfig::default()
        };
        let mut s = HttpProxySession::with_config(cfg);
        let mut wire = b"POST /a HTTP/1.1\r\nContent-Length: 4096\r\n\r\n".to_vec();
        wire.extend(std::iter::repeat_n(b'x', 4096));
        let mut out = Vec::new();
        s.feed_initiator(&wire, crate::Timestamp::default(), &mut out);
        assert_eq!(decoded_of(&out).len(), 4096, "no body bytes may be lost");
        assert!(matches!(out.last(), Some(HttpEvent::End { .. })));
    }

    #[test]
    fn adapter_reset_clears_state() {
        use crate::SessionParser;
        let mut s = HttpProxySession::new();
        let mut out = Vec::new();
        s.feed_initiator(
            b"POST /a HTTP/1.1\r\nContent-Length: 6\r\nTransfer-Encoding: chunked\r\n\r\n",
            crate::Timestamp::default(),
            &mut out,
        );
        assert!(s.is_poisoned());
        s.rst_initiator();
        assert!(!s.is_poisoned(), "a reset connection starts clean");
    }

    #[test]
    fn clean_fin_is_not_a_poison() {
        let mut p = HttpProxyParser::new();
        push_all(&mut p, FlowSide::Initiator, b"GET /a HTTP/1.1\r\n\r\n");
        let _ = drain(&mut p);
        p.fin(FlowSide::Initiator);
        p.fin(FlowSide::Responder);
        assert!(!p.is_poisoned());
        assert!(p.is_done());
    }

    #[test]
    fn close_delimited_response_flushes_at_fin() {
        let mut p = HttpProxyParser::new();
        push_all(&mut p, FlowSide::Initiator, b"GET /a HTTP/1.1\r\n\r\n");
        let _ = drain(&mut p);
        push_all(&mut p, FlowSide::Responder, b"HTTP/1.1 200 OK\r\n\r\nhello");
        let evs = drain(&mut p);
        assert_eq!(decoded_of(&evs), b"hello");
        p.fin(FlowSide::Responder);
        // Nothing was withheld: the body was reported as it arrived.
        assert!(!p.is_poisoned());
    }
}