cellos_supervisor/sni_proxy/h2/

mod.rs

//! HTTP/2 cleartext (h2c) frame parser — pure logic, no I/O.
//!
//! ## Phases
//!
//! - **Phase 3a (P3c)** — h2c preface detection + first-HEADERS-frame
//!   `:authority` extraction with **static-table-only HPACK**.
//!   CONTINUATION-fragmented HEADERS, dynamic-table refs, and Huffman
//!   literals all rejected.
//!
//! - **Phase 3g** — full HPACK dynamic-table state, Huffman literal
//!   decoding (RFC 7541 §5.2 + Appendix B), and CONTINUATION-fragmented
//!   HEADERS reassembly. Closes the v0.4.0 honest residual *"full HPACK
//!   dynamic-table state + Huffman literal decoding (currently rejected
//!   as `l7_h2_unparseable_headers`)."*
//!
//!   The stateless [`extract_h2_authority`] entry point is preserved with
//!   identical behaviour for backward compatibility — it constructs a
//!   one-shot [`HpackDecoder`] internally. Callers that need the dynamic
//!   table to evolve across multiple frames on the same connection use
//!   [`HpackDecoder::decode_block`] + [`reassemble_header_block`] directly
//!   (see the proxy wiring in `sni_proxy::handle_connection`).
//!
//! ## Honest scope (Phase 3g)
//!
//! - **h2c only.** TLS termination is ADR-0004 territory.
//! - **First-stream allow/deny only.** Connection-wide allow/deny is
//!   locked to the first `:authority` extracted; multi-stream per-request
//!   enforcement (different `:authority` per stream on the same connection)
//!   is future Phase 3g.1.
//! - **No HTTP/3 / QUIC.** Phase 4 separate.
//! - **Frame size cap.** RFC 7540 §6.5.2 default `SETTINGS_MAX_FRAME_SIZE`
//!   is 16 KiB. Frames whose declared length exceeds that are rejected as
//!   [`H2ParseError::OversizedFrame`].
//! - **Header block cap.** Aggregate HEADERS + CONTINUATION payload cannot
//!   exceed [`MAX_HEADER_BLOCK_SIZE`] (64 KiB) — defence against memory
//!   amplification via long fragmentation chains.
//! - **HPACK table size cap.** `SETTINGS_HEADER_TABLE_SIZE` updates are
//!   bounded at 64 KiB (see [`hpack::dynamic_table::MAX_TABLE_SIZE`]).
//!
//! [RFC 7540]: https://www.rfc-editor.org/rfc/rfc7540
//! [RFC 7541]: https://www.rfc-editor.org/rfc/rfc7541

pub mod error;
pub mod frame;
pub mod hpack;

pub use error::{H2ParseError, ReassemblerOverflowKind};
#[allow(unused_imports)]
pub use frame::{FrameHeader, DEFAULT_MAX_FRAME_SIZE};
#[allow(unused_imports)]
pub use hpack::{AuthorityProvenance, DecodedAuthority, HpackDecoder};

use frame::{
    parse_one_frame, strip_headers_padding_and_priority, FLAG_END_HEADERS, FRAME_TYPE_CONTINUATION,
    FRAME_TYPE_HEADERS, FRAME_TYPE_SETTINGS,
};
use std::collections::HashMap;

/// HTTP/2 connection preface (RFC 7540 §3.5).
pub const HTTP2_PREFACE: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";

/// Defensive cap on aggregate HEADERS + CONTINUATION payload. SETTINGS
/// `MAX_HEADER_LIST_SIZE` (RFC 7540 §6.5.2) governs this on the wire but
/// the proxy doesn't see SETTINGS that haven't reached the client; pick a
/// generous bound that admits realistic browsers (~16 KiB cookies + the
/// rest of the request line) while bounding adversarial fragmentation.
pub const MAX_HEADER_BLOCK_SIZE: usize = 64 * 1024;

/// Returns `true` when `bytes` starts with the canonical 24-byte HTTP/2
/// cleartext connection preface.
pub fn is_h2c_preface(bytes: &[u8]) -> bool {
    bytes.len() >= HTTP2_PREFACE.len() && &bytes[..HTTP2_PREFACE.len()] == HTTP2_PREFACE
}

/// Output of [`reassemble_header_block`]: the concatenated header-block
/// payload + the slice of input remaining after the END_HEADERS-bearing
/// frame.
pub type ReassembledBlock<'a> = (Vec<u8>, &'a [u8]);

/// Reassemble a header block from a stream of bytes that begins with a
/// HEADERS frame and may continue with one or more CONTINUATION frames
/// (RFC 7540 §4.3 + §6.10). Returns the concatenated header-block fragment
/// (after stripping padding + priority from the initial HEADERS) and the
/// slice of `buf` after the last CONTINUATION.
///
/// Returns `Ok(None)` if the buffer ends mid-frame (caller may wait for
/// more bytes). Returns `Err(InterleavedFrame)` if a non-CONTINUATION
/// frame appears between the initial HEADERS and the END_HEADERS-bearing
/// CONTINUATION. Returns `Err(HpackOversizedHeaderBlock)` if the
/// concatenated payloads exceed [`MAX_HEADER_BLOCK_SIZE`].
pub fn reassemble_header_block(buf: &[u8]) -> Result<Option<ReassembledBlock<'_>>, H2ParseError> {
    let mut cursor = buf;
    let mut accumulated: Vec<u8> = Vec::new();
    let mut first_frame = true;

    loop {
        let (header, payload, rest) = match parse_one_frame(cursor)? {
            Some(p) => p,
            None => return Ok(None),
        };

        if first_frame {
            if header.frame_type != FRAME_TYPE_HEADERS {
                return Err(H2ParseError::UnexpectedFirstFrame {
                    frame_type: header.frame_type,
                });
            }
            let block = strip_headers_padding_and_priority(payload, header.flags)?;
            if accumulated.len() + block.len() > MAX_HEADER_BLOCK_SIZE {
                return Err(H2ParseError::HpackOversizedHeaderBlock {
                    total: accumulated.len() + block.len(),
                    max: MAX_HEADER_BLOCK_SIZE,
                });
            }
            accumulated.extend_from_slice(block);
            cursor = rest;
            if header.flags & FLAG_END_HEADERS != 0 {
                return Ok(Some((accumulated, cursor)));
            }
            first_frame = false;
        } else {
            if header.frame_type != FRAME_TYPE_CONTINUATION {
                return Err(H2ParseError::InterleavedFrame {
                    frame_type: header.frame_type,
                });
            }
            // CONTINUATION carries no padding / priority.
            if accumulated.len() + payload.len() > MAX_HEADER_BLOCK_SIZE {
                return Err(H2ParseError::HpackOversizedHeaderBlock {
                    total: accumulated.len() + payload.len(),
                    max: MAX_HEADER_BLOCK_SIZE,
                });
            }
            accumulated.extend_from_slice(payload);
            cursor = rest;
            if header.flags & FLAG_END_HEADERS != 0 {
                return Ok(Some((accumulated, cursor)));
            }
        }
    }
}

/// Stateless one-shot extraction of `:authority` from h2c bytes that
/// begin **after** the 24-byte preface. This API is preserved from P3c
/// for backward compat; under the hood it constructs a one-shot
/// [`HpackDecoder`] and runs it against the reassembled header block.
///
/// Returns:
/// - `Ok(Some(authority))` — first HEADERS frame's `:authority`,
///   normalised (lowercased + port-stripped).
/// - `Ok(None)` — buffer too short, or no `:authority` in the block.
/// - `Err(...)` — see [`H2ParseError`].
pub fn extract_h2_authority(after_preface: &[u8]) -> Result<Option<String>, H2ParseError> {
    extract_h2_authority_with(after_preface, &mut HpackDecoder::new())
        .map(|opt| opt.map(|d| d.value))
}

/// Like [`extract_h2_authority`] but parameterised over a caller-owned
/// [`HpackDecoder`] so the dynamic table can persist across multiple
/// invocations on the same h2 connection. Returns the [`DecodedAuthority`]
/// (value + provenance) so the caller can pick the right reason code.
pub fn extract_h2_authority_with(
    after_preface: &[u8],
    decoder: &mut HpackDecoder,
) -> Result<Option<DecodedAuthority>, H2ParseError> {
    let mut cursor = after_preface;

    // Step 1: skip a single optional non-ACK SETTINGS frame.
    if let Some((header, _payload, rest)) = parse_one_frame(cursor)? {
        if header.frame_type == FRAME_TYPE_SETTINGS {
            cursor = rest;
        }
    } else {
        return Ok(None);
    }

    // Step 2: reassemble HEADERS + 0..n CONTINUATION.
    let (block, _rest_after) = match reassemble_header_block(cursor)? {
        Some(p) => p,
        None => return Ok(None),
    };

    // Step 3: HPACK-decode the assembled block.
    decoder.decode_block(&block)
}

// ── Phase 3g.1 — per-stream HEADERS+CONTINUATION reassembly ──────────────
//
// The reassembler is keyed on stream id (RFC 7540 §5.1.1) so HEADERS on
// stream 1 and HEADERS on stream 3 multiplexed on the same connection
// remain isolated. CONTINUATION frames (§6.10) are valid ONLY for the
// stream that has an open (non-END_HEADERS) HEADERS frame; any other
// frame type or any frame on a different stream while a reassembly is
// open is a PROTOCOL_ERROR (§6.10) — the reassembler returns
// `InterleavedFrame` for that case.
//
// Defensive memory bounds:
//
//   - Per-stream block ≤ [`MAX_HEADER_BLOCK_SIZE`] (64 KiB).
//   - Aggregate in-flight bytes ≤ [`REASSEMBLER_TOTAL_IN_FLIGHT_MAX`]
//     (256 KiB).
//   - Concurrent in-flight reassemblies ≤
//     [`REASSEMBLER_MAX_CONCURRENT_STREAMS`] (64).
//
// On overflow the reassembler returns
// [`H2ParseError::ReassemblerOverflow`] for the OFFENDING frame's stream;
// the caller emits `l7_h2_unparseable_headers` for that stream and
// RST_STREAMs only that stream — other in-flight streams continue.

/// Defensive aggregate bound on bytes the per-stream reassembler will
/// hold for streams that have started HEADERS but not yet sent
/// END_HEADERS. Sized at 4× per-stream cap so a small fleet of in-flight
/// streams can still complete; an adversary opening many streams and
/// hanging each at 64 KiB-1 trips this bound first (cheaper than waiting
/// for per-stream).
pub const REASSEMBLER_TOTAL_IN_FLIGHT_MAX: usize = 256 * 1024;

/// Defensive bound on concurrent in-flight HEADERS+CONTINUATION
/// reassemblies. RFC 7540 §6.5.2 `SETTINGS_MAX_CONCURRENT_STREAMS` is
/// usually 100 — 64 is below that floor, which is fine because a stream
/// only counts against this bound while its HEADERS block is unfinished.
/// Production traffic finishes HEADERS in one frame; only adversarial
/// fragmentation chains accumulate.
pub const REASSEMBLER_MAX_CONCURRENT_STREAMS: usize = 64;

/// Per-stream HEADERS+CONTINUATION accumulator (Phase 3g.1).
///
/// One instance per h2 connection, fed every HEADERS / CONTINUATION
/// frame the proxy parses. Returns the completed (stream-id, block)
/// pair when END_HEADERS is observed; otherwise tracks the pending
/// reassembly internally.
///
/// Non-HEADERS / non-CONTINUATION frames pass through with `Ok(None)`.
/// A non-CONTINUATION frame from any stream while a reassembly is open
/// on another stream is rejected as
/// [`H2ParseError::InterleavedFrame`] (RFC 7540 §6.10).
pub struct H2StreamReassembler {
    /// Accumulator per stream id. Bounded by the size + count caps below.
    blocks: HashMap<u32, Vec<u8>>,
    /// Stream id currently mid-reassembly. RFC 7540 §6.10: only one
    /// stream at a time may have an open HEADERS-without-END_HEADERS
    /// because CONTINUATION must immediately follow on the same stream.
    active_stream: Option<u32>,
    /// Cumulative size of all in-flight blocks; defends against memory
    /// abuse via many simultaneously-open streams.
    total_in_flight: usize,
}

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

impl H2StreamReassembler {
    /// Construct a fresh reassembler with no pending streams.
    pub fn new() -> Self {
        Self {
            blocks: HashMap::new(),
            active_stream: None,
            total_in_flight: 0,
        }
    }

    /// Feed a single frame's `(header, payload)` pair.
    ///
    /// Returns:
    ///
    /// - `Ok(Some((stream_id, block)))` — the frame completed a
    ///   HEADERS+CONTINUATION block. The block has had padding +
    ///   priority stripped (frame layer) and is the concatenated header
    ///   block fragment ready for `HpackDecoder::decode_block`.
    /// - `Ok(None)` — the frame either was not HEADERS/CONTINUATION
    ///   (passed through; pure flow-control or DATA), or was HEADERS
    ///   without END_HEADERS (more bytes needed), or was a CONTINUATION
    ///   without END_HEADERS (more bytes needed).
    /// - `Err(InterleavedFrame)` — RFC 7540 §6.10 violation: a frame
    ///   appeared between HEADERS and the END_HEADERS-bearing
    ///   CONTINUATION on the same stream; or a CONTINUATION arrived
    ///   for a stream that does NOT have an open reassembly.
    /// - `Err(ReassemblerOverflow)` — one of the defensive bounds
    ///   tripped; caller RST_STREAMs the offending stream id.
    pub fn ingest(
        &mut self,
        frame: &FrameHeader,
        payload: &[u8],
    ) -> Result<Option<(u32, Vec<u8>)>, H2ParseError> {
        // Reject CONTINUATION on a stream that isn't the active one,
        // and reject any non-CONTINUATION on any stream while an
        // active reassembly is open. This is the §6.10 PROTOCOL_ERROR
        // contract.
        if let Some(active) = self.active_stream {
            if frame.frame_type == FRAME_TYPE_CONTINUATION {
                if frame.stream_id != active {
                    return Err(H2ParseError::InterleavedFrame {
                        frame_type: frame.frame_type,
                    });
                }
                // Append; check per-stream + aggregate bounds.
                let entry = self
                    .blocks
                    .get_mut(&active)
                    .expect("active_stream invariant: blocks contains active");
                if entry.len() + payload.len() > MAX_HEADER_BLOCK_SIZE {
                    return Err(H2ParseError::ReassemblerOverflow {
                        kind: ReassemblerOverflowKind::PerStreamBlock,
                    });
                }
                if self.total_in_flight + payload.len() > REASSEMBLER_TOTAL_IN_FLIGHT_MAX {
                    return Err(H2ParseError::ReassemblerOverflow {
                        kind: ReassemblerOverflowKind::TotalInFlight,
                    });
                }
                entry.extend_from_slice(payload);
                self.total_in_flight += payload.len();
                if frame.flags & FLAG_END_HEADERS != 0 {
                    let block = self
                        .blocks
                        .remove(&active)
                        .expect("active reassembly entry");
                    self.total_in_flight = self.total_in_flight.saturating_sub(block.len());
                    self.active_stream = None;
                    return Ok(Some((active, block)));
                }
                return Ok(None);
            }
            // Anything else mid-reassembly is §6.10 PROTOCOL_ERROR.
            return Err(H2ParseError::InterleavedFrame {
                frame_type: frame.frame_type,
            });
        }

        // No active reassembly. Only HEADERS opens one.
        if frame.frame_type == FRAME_TYPE_HEADERS {
            if self.blocks.len() >= REASSEMBLER_MAX_CONCURRENT_STREAMS {
                return Err(H2ParseError::ReassemblerOverflow {
                    kind: ReassemblerOverflowKind::ConcurrentStreams,
                });
            }
            let inner = strip_headers_padding_and_priority(payload, frame.flags)?;
            if inner.len() > MAX_HEADER_BLOCK_SIZE {
                return Err(H2ParseError::ReassemblerOverflow {
                    kind: ReassemblerOverflowKind::PerStreamBlock,
                });
            }
            if self.total_in_flight + inner.len() > REASSEMBLER_TOTAL_IN_FLIGHT_MAX {
                return Err(H2ParseError::ReassemblerOverflow {
                    kind: ReassemblerOverflowKind::TotalInFlight,
                });
            }
            if frame.flags & FLAG_END_HEADERS != 0 {
                // Single-frame HEADERS; complete immediately. No need to
                // touch blocks/active.
                return Ok(Some((frame.stream_id, inner.to_vec())));
            }
            self.blocks.insert(frame.stream_id, inner.to_vec());
            self.total_in_flight += inner.len();
            self.active_stream = Some(frame.stream_id);
            return Ok(None);
        }

        // CONTINUATION without active reassembly is §6.10 PROTOCOL_ERROR.
        if frame.frame_type == FRAME_TYPE_CONTINUATION {
            return Err(H2ParseError::InterleavedFrame {
                frame_type: frame.frame_type,
            });
        }

        // Any other frame type (DATA, SETTINGS, WINDOW_UPDATE, PING,
        // GOAWAY, PRIORITY, RST_STREAM, PUSH_PROMISE) passes through.
        Ok(None)
    }

    /// Number of streams currently mid-reassembly (HEADERS open without
    /// END_HEADERS). Test-only.
    #[cfg(test)]
    pub fn pending_count(&self) -> usize {
        self.blocks.len()
    }
}

/// Output of [`H2ConnectionDecoder::feed_frame`] when a HEADERS+
/// CONTINUATION block completes for a particular stream.
#[derive(Debug, Clone)]
pub struct H2HeadersDecoded {
    /// Stream id the HEADERS belonged to (RFC 7540 §4.1).
    pub stream_id: u32,
    /// Extracted `:authority`. `None` when the HEADERS block parsed
    /// cleanly but contained no `:authority` pseudo-header — caller
    /// emits `l7_h2_authority_missing` and RST_STREAMs.
    pub authority: Option<String>,
    /// True when HEADERS were transported via dynamic-table indexing.
    pub via_dynamic_table: bool,
    /// True when the `:authority` value used Huffman-coded literal.
    pub via_huffman: bool,
}

/// Per-connection h2c decoder stacking the per-stream
/// [`H2StreamReassembler`] over the per-connection [`HpackDecoder`].
///
/// The reassembler isolates HEADERS+CONTINUATION sequences per stream;
/// the HPACK decoder is per-connection because RFC 7541 §2.3.1 says the
/// dynamic table is a single per-connection compression context shared
/// across all streams.
pub struct H2ConnectionDecoder {
    hpack: HpackDecoder,
    reassembler: H2StreamReassembler,
}

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

impl H2ConnectionDecoder {
    /// Fresh decoder with empty dynamic table and no in-flight reassemblies.
    pub fn new() -> Self {
        Self {
            hpack: HpackDecoder::new(),
            reassembler: H2StreamReassembler::new(),
        }
    }

    /// Feed a single frame.
    ///
    /// Returns `Some(decoded)` when a HEADERS+CONTINUATION block
    /// completes; `None` when more bytes are needed or the frame was
    /// not HEADERS/CONTINUATION. Errors are stream-scoped — the caller
    /// decides whether to RST_STREAM the offender or close the
    /// whole connection.
    pub fn feed_frame(
        &mut self,
        frame: &FrameHeader,
        payload: &[u8],
    ) -> Result<Option<H2HeadersDecoded>, H2ParseError> {
        let (stream_id, block) = match self.reassembler.ingest(frame, payload)? {
            Some(p) => p,
            None => return Ok(None),
        };
        let decoded = self.hpack.decode_block(&block)?;
        let (authority, via_dynamic_table, via_huffman) = match decoded {
            Some(d) => {
                let dyn_indexed = matches!(d.provenance, AuthorityProvenance::DynamicIndexed);
                let huff = matches!(d.provenance, AuthorityProvenance::Huffman);
                (Some(d.value), dyn_indexed, huff)
            }
            None => (None, false, false),
        };
        Ok(Some(H2HeadersDecoded {
            stream_id,
            authority,
            via_dynamic_table,
            via_huffman,
        }))
    }
}

#[cfg(test)]
pub(crate) mod test_helpers {
    //! Hand-crafted h2 byte sequences for unit tests. Pure constructors —
    //! no I/O, no `h2`-crate dependency. The shapes mirror what real h2c
    //! clients (curl --http2-prior-knowledge, h2load, nghttp) emit on the
    //! wire.

    use super::frame::{FRAME_TYPE_CONTINUATION, FRAME_TYPE_HEADERS, FRAME_TYPE_SETTINGS};
    use super::hpack::huffman;

    /// Build a 9-byte h2 frame header.
    pub fn frame_header(length: u32, frame_type: u8, flags: u8, stream_id: u32) -> Vec<u8> {
        let mut out = Vec::with_capacity(9);
        out.push(((length >> 16) & 0xFF) as u8);
        out.push(((length >> 8) & 0xFF) as u8);
        out.push((length & 0xFF) as u8);
        out.push(frame_type);
        out.push(flags);
        out.extend_from_slice(&(stream_id & 0x7FFF_FFFF).to_be_bytes());
        out
    }

    /// Empty SETTINGS frame on stream 0 (no payload, no ACK).
    pub fn empty_settings_frame() -> Vec<u8> {
        frame_header(0, FRAME_TYPE_SETTINGS, 0, 0)
    }

    /// Build a HEADERS frame on stream 1 with `END_HEADERS | END_STREAM`,
    /// payload = the given header-block fragment.
    pub fn headers_frame(header_block: &[u8]) -> Vec<u8> {
        let length = header_block.len() as u32;
        let flags = super::frame::FLAG_END_HEADERS | 0x1; // END_STREAM
        let mut out = frame_header(length, FRAME_TYPE_HEADERS, flags, 1);
        out.extend_from_slice(header_block);
        out
    }

    /// Build a HEADERS frame missing END_HEADERS (i.e. CONTINUATION
    /// fragmentation). `flags` already excludes END_HEADERS.
    pub fn continuation_fragmented_headers(header_block: &[u8]) -> Vec<u8> {
        let length = header_block.len() as u32;
        let flags = 0x0;
        let mut out = frame_header(length, FRAME_TYPE_HEADERS, flags, 1);
        out.extend_from_slice(header_block);
        out
    }

    /// Build a CONTINUATION frame on stream 1.
    pub fn continuation_frame(header_block: &[u8], end_headers: bool) -> Vec<u8> {
        let length = header_block.len() as u32;
        let flags = if end_headers {
            super::frame::FLAG_END_HEADERS
        } else {
            0x0
        };
        let mut out = frame_header(length, FRAME_TYPE_CONTINUATION, flags, 1);
        out.extend_from_slice(header_block);
        out
    }

    /// HPACK-encode a literal-with-incremental-indexing entry whose name
    /// is an indexed reference (`name_index`) and whose value is a
    /// non-Huffman literal.
    pub fn hpack_literal_indexed_name(name_index: u8, value: &str) -> Vec<u8> {
        let mut out = Vec::new();
        out.push(0x40 | (name_index & 0x3F));
        encode_literal_string(&mut out, value);
        out
    }

    /// HPACK-encode a literal-with-incremental-indexing entry whose name
    /// AND value are both non-Huffman literals.
    pub fn hpack_literal_indexed_with_name(name: &str, value: &str) -> Vec<u8> {
        let mut out = Vec::new();
        out.push(0x40);
        encode_literal_string(&mut out, name);
        encode_literal_string(&mut out, value);
        out
    }

    /// HPACK-encode a literal-without-indexing entry whose name is an
    /// indexed reference. Uses the `0000xxxx` representation (RFC 7541
    /// §6.2.2); the four-bit prefix carries the name index.
    pub fn hpack_literal_no_indexing(name_index: u8, value: &str) -> Vec<u8> {
        let mut out = Vec::new();
        out.push(name_index & 0x0F);
        encode_literal_string(&mut out, value);
        out
    }

    /// HPACK-encode an indexed-header-field reference (8-bit form).
    pub fn hpack_indexed(index: u8) -> Vec<u8> {
        vec![0x80 | (index & 0x7F)]
    }

    /// HPACK-encode a literal-with-incremental-indexing entry whose name
    /// is indexed and whose value is *Huffman-coded*.
    pub fn hpack_literal_indexed_name_huffman(name_index: u8, value: &str) -> Vec<u8> {
        let mut out = Vec::new();
        out.push(0x40 | (name_index & 0x3F));
        let payload = huffman::encode(value);
        // Length octet: top bit = 1 (Huffman).
        if payload.len() < 0x7F {
            out.push(0x80 | payload.len() as u8);
        } else {
            out.push(0xFF);
            let mut v = payload.len() as u64 - 0x7F;
            while v >= 128 {
                out.push(((v & 0x7F) as u8) | 0x80);
                v >>= 7;
            }
            out.push(v as u8);
        }
        out.extend_from_slice(&payload);
        out
    }

    /// Encode an HPACK non-Huffman string literal (length prefix + raw
    /// octets) into `out`.
    pub fn encode_literal_string(out: &mut Vec<u8>, s: &str) {
        let len = s.len() as u64;
        if len < 0x7F {
            out.push(len as u8);
        } else {
            out.push(0x7F);
            let mut v = len - 0x7F;
            while v >= 128 {
                out.push(((v & 0x7F) as u8) | 0x80);
                v >>= 7;
            }
            out.push(v as u8);
        }
        out.extend_from_slice(s.as_bytes());
    }

    /// Build a Huffman-flagged literal whose payload is the raw bytes
    /// given (NOT actually Huffman-coded). Used to assert that broken
    /// Huffman streams reject.
    pub fn hpack_literal_huffman_indexed_name_with_raw(
        name_index: u8,
        raw_value: &[u8],
    ) -> Vec<u8> {
        let mut out = Vec::new();
        out.push(0x40 | (name_index & 0x3F));
        out.push(0x80 | (raw_value.len() as u8));
        out.extend_from_slice(raw_value);
        out
    }

    /// Build a complete h2c stream prefix: SETTINGS + HEADERS frame
    /// carrying the given header-block fragment. Does NOT include the
    /// 24-byte preface.
    pub fn settings_then_headers(header_block: &[u8]) -> Vec<u8> {
        let mut out = empty_settings_frame();
        out.extend_from_slice(&headers_frame(header_block));
        out
    }

    /// Build a HEADERS frame on `stream_id` with `END_HEADERS | END_STREAM`
    /// and the given header-block fragment. (Helper for Phase 3g.1
    /// per-stream reassembler tests.)
    pub fn headers_frame_on_stream(header_block: &[u8], stream_id: u32) -> Vec<u8> {
        let length = header_block.len() as u32;
        let flags = super::frame::FLAG_END_HEADERS | 0x1; // END_STREAM
        let mut out = frame_header(length, FRAME_TYPE_HEADERS, flags, stream_id);
        out.extend_from_slice(header_block);
        out
    }

    /// Build a HEADERS frame on `stream_id` WITHOUT END_HEADERS (i.e.
    /// CONTINUATION fragmentation expected).
    pub fn headers_frame_on_stream_no_end(header_block: &[u8], stream_id: u32) -> Vec<u8> {
        let length = header_block.len() as u32;
        let mut out = frame_header(length, FRAME_TYPE_HEADERS, 0x0, stream_id);
        out.extend_from_slice(header_block);
        out
    }

    /// Build a CONTINUATION frame on `stream_id`.
    pub fn continuation_frame_on_stream(
        header_block: &[u8],
        stream_id: u32,
        end_headers: bool,
    ) -> Vec<u8> {
        let length = header_block.len() as u32;
        let flags = if end_headers {
            super::frame::FLAG_END_HEADERS
        } else {
            0x0
        };
        let mut out = frame_header(length, FRAME_TYPE_CONTINUATION, flags, stream_id);
        out.extend_from_slice(header_block);
        out
    }

    /// Build a DATA frame on `stream_id` with arbitrary payload.
    pub fn data_frame_on_stream(payload: &[u8], stream_id: u32) -> Vec<u8> {
        let length = payload.len() as u32;
        let mut out = frame_header(length, super::frame::FRAME_TYPE_DATA, 0x0, stream_id);
        out.extend_from_slice(payload);
        out
    }
}

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

    // ── 1. Static-table indexed authority (no value carried) ────────────
    #[test]
    fn extracts_authority_from_static_table_index_1() {
        let block = hpack_indexed(1);
        let bytes = settings_then_headers(&block);
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(
            result, None,
            "indexed-only :authority has no value; parser returns None"
        );
    }

    // ── 2. Literal-with-indexing, name from static table index 1 ────────
    #[test]
    fn extracts_authority_from_literal_indexed() {
        let block = hpack_literal_indexed_name(1, "api.example.com");
        let bytes = settings_then_headers(&block);
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("api.example.com"));
    }

    // ── 3. Literal with name AND value as literals (no static index) ────
    #[test]
    fn extracts_authority_from_literal_with_incremental_indexing() {
        let block = hpack_literal_indexed_with_name(":authority", "api.example.com");
        let bytes = settings_then_headers(&block);
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("api.example.com"));
    }

    // ── 4. Oversized frame → reject ─────────────────────────────────────
    #[test]
    fn rejects_oversized_frame() {
        let mut bytes = empty_settings_frame();
        let bogus_header = frame_header(
            16_385,
            frame::FRAME_TYPE_HEADERS,
            frame::FLAG_END_HEADERS,
            1,
        );
        bytes.extend_from_slice(&bogus_header);
        let err = extract_h2_authority(&bytes).unwrap_err();
        assert!(matches!(
            err,
            H2ParseError::OversizedFrame {
                declared: 16_385,
                max: 16_384
            }
        ));
    }

    // ── 5. CONTINUATION fragmentation now reassembles ────────────────────
    // scope: a HEADERS payload is treated as the first chunk and we wait
    // for CONTINUATION; the test guards against regression to the older
    // "reject CONTINUATION-fragmented HEADERS" behaviour.
    #[test]
    fn reassembles_headers_with_one_continuation() {
        let block_full = hpack_literal_indexed_name(1, "api.example.com");
        // Split the block in half and stuff the second half in a CONTINUATION
        // with END_HEADERS.
        let mid = block_full.len() / 2;
        let (first_half, second_half) = block_full.split_at(mid);
        let mut bytes = empty_settings_frame();
        bytes.extend_from_slice(&continuation_fragmented_headers(first_half));
        bytes.extend_from_slice(&continuation_frame(second_half, true));
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("api.example.com"));
    }

    // ── 6. Non-HEADERS first frame after SETTINGS → reject ──────────────
    #[test]
    fn rejects_non_headers_first_frame() {
        let mut bytes = empty_settings_frame();
        bytes.extend_from_slice(&frame_header(0, frame::FRAME_TYPE_DATA, 0, 1));
        let err = extract_h2_authority(&bytes).unwrap_err();
        assert!(matches!(
            err,
            H2ParseError::UnexpectedFirstFrame { frame_type: 0 }
        ));
    }

    // ── 7. Mixed-case authority is lowercased ───────────────────────────
    #[test]
    fn extracts_lowercased_authority() {
        let block = hpack_literal_indexed_name(1, "API.Example.COM");
        let bytes = settings_then_headers(&block);
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("api.example.com"));
    }

    // ── 8. Port stripped from authority ─────────────────────────────────
    #[test]
    fn strips_port_from_authority() {
        let block = hpack_literal_indexed_name(1, "api.example.com:8443");
        let bytes = settings_then_headers(&block);
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("api.example.com"));
    }

    // ── 9. Buffer too short → Ok(None) ──────────────────────────────────
    #[test]
    fn returns_none_when_buffer_short() {
        let bytes = empty_settings_frame();
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result, None);

        let truncated = &bytes[..5];
        let result = extract_h2_authority(truncated).unwrap();
        assert_eq!(result, None);
    }

    // ── 10. h2c preface recognition ─────────────────────────────────────
    #[test]
    fn is_h2c_preface_recognises_canonical_preface() {
        let mut bytes = HTTP2_PREFACE.to_vec();
        bytes.extend_from_slice(&[0x00, 0x00, 0x00]);
        assert!(is_h2c_preface(&bytes));
        assert!(is_h2c_preface(HTTP2_PREFACE));
    }

    #[test]
    fn is_h2c_preface_rejects_malformed() {
        assert!(!is_h2c_preface(&HTTP2_PREFACE[..23]));
        let mut bad = HTTP2_PREFACE.to_vec();
        bad[5] = b'x';
        assert!(!is_h2c_preface(&bad));
        assert!(!is_h2c_preface(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n"));
        assert!(!is_h2c_preface(&[0x16, 0x03, 0x01]));
    }

    // ── 11. Padded HEADERS frame ────────────────────────────────────────
    #[test]
    fn extracts_authority_from_padded_headers_frame() {
        let block = hpack_literal_indexed_name(1, "api.example.com");
        let pad_len: u8 = 4;
        let mut padded_payload = Vec::new();
        padded_payload.push(pad_len);
        padded_payload.extend_from_slice(&block);
        padded_payload.extend(std::iter::repeat_n(0u8, pad_len as usize));
        let length = padded_payload.len() as u32;
        let flags = frame::FLAG_END_HEADERS | frame::FLAG_PADDED;
        let mut bytes = empty_settings_frame();
        bytes.extend_from_slice(&frame_header(length, frame::FRAME_TYPE_HEADERS, flags, 1));
        bytes.extend_from_slice(&padded_payload);
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("api.example.com"));
    }

    // ── 12. Padding length larger than payload → reject ────────────────
    #[test]
    fn rejects_padding_overflow() {
        let block = hpack_literal_indexed_name(1, "api.example.com");
        let mut padded_payload = Vec::new();
        padded_payload.push(255);
        padded_payload.extend_from_slice(&block);
        let length = padded_payload.len() as u32;
        let flags = frame::FLAG_END_HEADERS | frame::FLAG_PADDED;
        let mut bytes = empty_settings_frame();
        bytes.extend_from_slice(&frame_header(length, frame::FRAME_TYPE_HEADERS, flags, 1));
        bytes.extend_from_slice(&padded_payload);
        let err = extract_h2_authority(&bytes).unwrap_err();
        assert_eq!(err, H2ParseError::PaddingOverflow);
    }

    // ── 13. Huffman-coded literal NOW DECODES (Phase 3g) ───────────────
    // Previously this rejected as HpackUnsupported. Now it succeeds.
    #[test]
    fn decodes_huffman_coded_literal() {
        let block = hpack_literal_indexed_name_huffman(1, "api.example.com");
        let bytes = settings_then_headers(&block);
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("api.example.com"));
    }

    // ── 14. Dynamic-table reference NOW DECODES (Phase 3g) ─────────────
    // Requires a stateful decoder — the stateless one-shot path returns
    // an error since the dynamic table is empty. Use the stateful API.
    #[test]
    fn decodes_dynamic_table_reference_with_stateful_decoder() {
        let mut decoder = HpackDecoder::new();
        // Connection 1: incremental-index :authority = "api.example.com".
        let block1 = hpack_literal_indexed_name(1, "api.example.com");
        let bytes1 = settings_then_headers(&block1);
        let r1 = extract_h2_authority_with(&bytes1, &mut decoder)
            .unwrap()
            .unwrap();
        assert_eq!(r1.value, "api.example.com");
        // Connection 2 (same decoder): indexed-header-field at dynamic
        // index 62. Build a SETTINGS+HEADERS sequence.
        let block2 = vec![0x80 | 62];
        let bytes2 = settings_then_headers(&block2);
        let r2 = extract_h2_authority_with(&bytes2, &mut decoder)
            .unwrap()
            .unwrap();
        assert_eq!(r2.value, "api.example.com");
        assert_eq!(r2.provenance, AuthorityProvenance::DynamicIndexed);
    }

    // ── 15. PRIORITY-bearing HEADERS frame ──────────────────────────────
    #[test]
    fn extracts_authority_with_priority_flag() {
        let block = hpack_literal_indexed_name(1, "api.example.com");
        let priority_section = [0u8; 5];
        let mut payload = Vec::new();
        payload.extend_from_slice(&priority_section);
        payload.extend_from_slice(&block);
        let length = payload.len() as u32;
        let flags = frame::FLAG_END_HEADERS | frame::FLAG_PRIORITY;
        let mut bytes = empty_settings_frame();
        bytes.extend_from_slice(&frame_header(length, frame::FRAME_TYPE_HEADERS, flags, 1));
        bytes.extend_from_slice(&payload);
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("api.example.com"));
    }

    // ── 16. Authority with trailing dot is stripped ─────────────────────
    #[test]
    fn strips_trailing_dot_from_authority() {
        let block = hpack_literal_indexed_name(1, "api.example.com.");
        let bytes = settings_then_headers(&block);
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("api.example.com"));
    }

    // ── 17. IPv6 literal authority keeps brackets, drops port ───────────
    #[test]
    fn ipv6_literal_authority_keeps_brackets() {
        let block = hpack_literal_indexed_name(1, "[::1]:443");
        let bytes = settings_then_headers(&block);
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("[::1]"));
    }

    // ── 18. Literal-without-indexing path ───────────────────────────────
    #[test]
    fn extracts_authority_from_literal_without_indexing() {
        let block = hpack_literal_no_indexing(1, "api.example.com");
        let bytes = settings_then_headers(&block);
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("api.example.com"));
    }

    // ── 19. Broken Huffman bitstream → reject ──────────────────────────
    #[test]
    fn rejects_broken_huffman_literal() {
        // Raw bytes 0x00,0x00,0x00 inside a Huffman-flagged length —
        // decoder will fail to decode and return HuffmanInvalid (or
        // padding-violates-EOS rule).
        let block = hpack_literal_huffman_indexed_name_with_raw(1, &[0x00, 0x00, 0x00]);
        let bytes = settings_then_headers(&block);
        let err = extract_h2_authority(&bytes).unwrap_err();
        assert!(matches!(
            err,
            H2ParseError::HuffmanInvalid | H2ParseError::HuffmanOversized { .. }
        ));
    }

    // ── 20. Reassembly rejects interleaved DATA frame ──────────────────
    #[test]
    fn reassembly_rejects_interleaved_data_frame() {
        let block_full = hpack_literal_indexed_name(1, "api.example.com");
        let mid = block_full.len() / 2;
        let (first_half, second_half) = block_full.split_at(mid);
        let mut bytes = empty_settings_frame();
        bytes.extend_from_slice(&continuation_fragmented_headers(first_half));
        // Insert a DATA frame between HEADERS and the END_HEADERS continuation.
        bytes.extend_from_slice(&frame_header(0, frame::FRAME_TYPE_DATA, 0, 1));
        bytes.extend_from_slice(&continuation_frame(second_half, true));
        let err = extract_h2_authority(&bytes).unwrap_err();
        assert!(matches!(
            err,
            H2ParseError::InterleavedFrame { frame_type: 0 }
        ));
    }

    // ── 21. Three-fragment reassembly with priority + padding ──────────
    #[test]
    fn reassembles_three_fragments_with_priority_and_padding() {
        let block_full = hpack_literal_indexed_name(1, "api.example.com");
        let third = block_full.len() / 3;
        let (a, rest) = block_full.split_at(third);
        let (b, c) = rest.split_at(third);

        // First HEADERS frame: priority + padded, no END_HEADERS.
        let pad_len: u8 = 2;
        let priority = [0u8; 5];
        let mut headers_payload = Vec::new();
        headers_payload.push(pad_len);
        headers_payload.extend_from_slice(&priority);
        headers_payload.extend_from_slice(a);
        headers_payload.extend(std::iter::repeat_n(0u8, pad_len as usize));
        let headers_frame_bytes = {
            let length = headers_payload.len() as u32;
            let flags = frame::FLAG_PADDED | frame::FLAG_PRIORITY;
            let mut out = frame_header(length, frame::FRAME_TYPE_HEADERS, flags, 1);
            out.extend_from_slice(&headers_payload);
            out
        };

        let mut bytes = empty_settings_frame();
        bytes.extend_from_slice(&headers_frame_bytes);
        bytes.extend_from_slice(&continuation_frame(b, false));
        bytes.extend_from_slice(&continuation_frame(c, true));

        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("api.example.com"));
    }

    // ── 22. Reassembly oversized aggregate → reject ────────────────────
    #[test]
    fn reassembly_rejects_oversized_aggregate_block() {
        // Build > 64 KiB of header-block bytes split across CONTINUATION
        // frames. We use literal-with-indexing + a giant value that, on
        // its own, exceeds MAX_HEADER_BLOCK_SIZE.
        // Instead of a single oversized value (which would exceed
        // MAX_FRAME_SIZE = 16 KiB), chain enough CONTINUATION frames
        // each carrying ~16 KiB of arbitrary HPACK octets.
        let mut bytes = empty_settings_frame();
        // First HEADERS frame: 16383 bytes of payload, no END_HEADERS.
        // Use literal-with-indexing name=:authority value=<garbage>; the
        // decoder will likely error on garbage but reassembly checks the
        // size cap first.
        let chunk = vec![0xAAu8; 16_383];
        bytes.extend_from_slice(&continuation_fragmented_headers(&chunk));
        bytes.extend_from_slice(&continuation_frame(&chunk, false));
        bytes.extend_from_slice(&continuation_frame(&chunk, false));
        bytes.extend_from_slice(&continuation_frame(&chunk, false));
        bytes.extend_from_slice(&continuation_frame(&chunk, true));
        let err = extract_h2_authority(&bytes).unwrap_err();
        assert!(matches!(
            err,
            H2ParseError::HpackOversizedHeaderBlock { .. }
        ));
    }

    // ── 23. Single HEADERS with END_HEADERS works via reassembly path ──
    #[test]
    fn single_headers_with_end_headers_works_unchanged_via_reassembly_path() {
        // Same as test #2 but verifies reassembly is transparent.
        let block = hpack_literal_indexed_name(1, "api.example.com");
        let bytes = settings_then_headers(&block);
        let result = extract_h2_authority(&bytes).unwrap();
        assert_eq!(result.as_deref(), Some("api.example.com"));
    }

    // ─────────────────────────────────────────────────────────────────────
    // scope: H2StreamReassembler unit tests
    // ─────────────────────────────────────────────────────────────────────

    fn parse_first_frame(bytes: &[u8]) -> (frame::FrameHeader, Vec<u8>) {
        let (header, payload, _rest) = parse_one_frame(bytes).unwrap().unwrap();
        (header, payload.to_vec())
    }

    #[test]
    fn ingest_single_headers_with_end_headers_returns_block() {
        let mut r = H2StreamReassembler::new();
        let block = hpack_literal_indexed_name(1, "api.example.com");
        let frame_bytes = headers_frame(&block);
        let (h, p) = parse_first_frame(&frame_bytes);
        let out = r.ingest(&h, &p).unwrap();
        let (sid, returned) = out.expect("expected immediate completion");
        assert_eq!(sid, 1);
        assert_eq!(returned, block);
        assert_eq!(r.pending_count(), 0);
    }

    #[test]
    fn ingest_two_headers_on_different_streams_returns_two_blocks() {
        let mut r = H2StreamReassembler::new();
        let block_a = hpack_literal_indexed_name(1, "api.example.com");
        let block_b = hpack_literal_indexed_name(1, "api.other.com");
        let f_a = headers_frame_on_stream(&block_a, 1);
        let f_b = headers_frame_on_stream(&block_b, 3);
        let (ha, pa) = parse_first_frame(&f_a);
        let (hb, pb) = parse_first_frame(&f_b);
        let out_a = r.ingest(&ha, &pa).unwrap().unwrap();
        let out_b = r.ingest(&hb, &pb).unwrap().unwrap();
        assert_eq!(out_a.0, 1);
        assert_eq!(out_b.0, 3);
        assert_eq!(out_a.1, block_a);
        assert_eq!(out_b.1, block_b);
    }

    #[test]
    fn ingest_continuation_on_correct_active_stream_reassembles() {
        let mut r = H2StreamReassembler::new();
        let block_full = hpack_literal_indexed_name(1, "api.example.com");
        let mid = block_full.len() / 2;
        let (a, b) = block_full.split_at(mid);
        let f1 = headers_frame_on_stream_no_end(a, 1);
        let f2 = continuation_frame_on_stream(b, 1, true);
        let (h1, p1) = parse_first_frame(&f1);
        let (h2, p2) = parse_first_frame(&f2);
        assert!(r.ingest(&h1, &p1).unwrap().is_none());
        assert_eq!(r.pending_count(), 1);
        let (sid, block) = r.ingest(&h2, &p2).unwrap().unwrap();
        assert_eq!(sid, 1);
        assert_eq!(block, block_full);
        assert_eq!(r.pending_count(), 0);
    }

    #[test]
    fn ingest_continuation_on_wrong_stream_returns_interleaved() {
        let mut r = H2StreamReassembler::new();
        let block_full = hpack_literal_indexed_name(1, "api.example.com");
        let mid = block_full.len() / 2;
        let (a, b) = block_full.split_at(mid);
        let f1 = headers_frame_on_stream_no_end(a, 1);
        let f2 = continuation_frame_on_stream(b, 3, true);
        let (h1, p1) = parse_first_frame(&f1);
        let (h2, p2) = parse_first_frame(&f2);
        r.ingest(&h1, &p1).unwrap();
        let err = r.ingest(&h2, &p2).unwrap_err();
        assert!(matches!(err, H2ParseError::InterleavedFrame { .. }));
    }

    #[test]
    fn ingest_data_frame_passes_through_with_no_block_returned() {
        let mut r = H2StreamReassembler::new();
        let f = data_frame_on_stream(b"hello", 1);
        let (h, p) = parse_first_frame(&f);
        assert!(r.ingest(&h, &p).unwrap().is_none());
        assert_eq!(r.pending_count(), 0);
    }

    #[test]
    fn ingest_oversized_per_stream_block_returns_overflow() {
        let mut r = H2StreamReassembler::new();
        // Open a HEADERS that's already at MAX_HEADER_BLOCK_SIZE - 1 then
        // append a CONTINUATION that pushes over the per-stream cap.
        // Frames must obey MAX_FRAME_SIZE = 16 KiB so we chain many
        // CONTINUATIONs of ~16 KiB until we trip the per-stream cap.
        let chunk = vec![0u8; 16_383];
        let f1 = headers_frame_on_stream_no_end(&chunk, 1);
        let (h1, p1) = parse_first_frame(&f1);
        r.ingest(&h1, &p1).unwrap();
        // After 1 HEADERS + 3 CONTINUATIONs we're at 4*16383 = 65532 ≤
        // 65536 — borderline. Add a 4th CONTINUATION of 16383 → 81915 >
        // 65536 → PerStreamBlock.
        for _ in 0..3 {
            let f = continuation_frame_on_stream(&chunk, 1, false);
            let (h, p) = parse_first_frame(&f);
            r.ingest(&h, &p).unwrap();
        }
        let f = continuation_frame_on_stream(&chunk, 1, false);
        let (h, p) = parse_first_frame(&f);
        let err = r.ingest(&h, &p).unwrap_err();
        assert!(matches!(
            err,
            H2ParseError::ReassemblerOverflow {
                kind: ReassemblerOverflowKind::PerStreamBlock
            }
        ));
    }

    #[test]
    fn ingest_oversized_aggregate_returns_overflow() {
        // Open 5 streams each at 60 KiB → 300 KiB > 256 KiB cap.
        // Use ascending odd stream ids (RFC 7540 §5.1.1: client streams
        // are odd). Each stream just opens HEADERS without END_HEADERS.
        // The first 4 streams accumulate fine; the 5th (or its
        // continuation) trips the aggregate cap.
        //
        // Strategy: build streams that have just-under-MAX_HEADER_BLOCK_SIZE
        // pending. We can't have multiple streams "active" at once
        // because §6.10 forbids interleaving — so this test exercises
        // the aggregate-cap math by opening one stream at a time and
        // closing it, then reusing the bytes. That doesn't accumulate.
        //
        // Instead: each open HEADERS-with-END_HEADERS returns immediately
        // and DOESN'T linger. To exercise the aggregate cap we need
        // to accumulate via the active reassembly. So use a single
        // stream that pushes past 256 KiB — but per-stream cap is
        // 64 KiB, which trips first. Therefore the aggregate cap is
        // primarily a defence against fragmented HEADERS pending across
        // many active reassemblies — but only ONE reassembly may be
        // active per §6.10.
        //
        // The aggregate cap is exercised by the reassembler keeping
        // pending blocks in the map even after the active stream
        // changes (which §6.10 forbids — so this is purely defence in
        // depth). To trigger it deterministically we directly inject
        // entries via the test-only API (no such API exists).
        //
        // Simplest deterministic test: feed one stream up to the
        // per-stream cap and verify the cap trips with kind =
        // PerStreamBlock; then construct a smaller cap scenario via a
        // second reassembler whose total_in_flight is artificially
        // pushed by feeding many small CONTINUATIONs.
        //
        // Since per-stream cap < aggregate cap, the per-stream cap
        // always trips first for a single stream. The aggregate cap is
        // therefore unreachable in normal protocol; we can still assert
        // it triggers if total_in_flight would exceed via direct path:
        // we build an *accept* path that fills up to 250 KiB across
        // multiple completed streams (each is removed on END_HEADERS
        // so total_in_flight returns to 0). To force it nonzero we
        // need pending streams. RFC §6.10 says only one open at once
        // — but the reassembler enforces that via InterleavedFrame.
        //
        // Conclusion: aggregate cap is checked but only triggers in
        // anomalous cases (e.g. a future PUSH_PROMISE path that doesn't
        // close a stream cleanly). For Phase 3g.1 unit-test coverage we
        // assert that the bound is checked at ingest (any fresh
        // single-stream block past the bound trips PerStreamBlock first
        // because per-stream is the smaller of the two). The aggregate
        // path is therefore tested by the synthetic case below where
        // we manually stuff the reassembler's internal state.
        let mut r = H2StreamReassembler::new();
        // Manually pre-fill an entry to simulate a pending block from a
        // prior frame (this can occur if a future code path leaves
        // residual reassembly state).
        r.blocks.insert(1, vec![0u8; 60 * 1024]);
        r.total_in_flight = 60 * 1024;
        r.active_stream = Some(1);
        // Append another 200 KiB worth via a CONTINUATION that nominally
        // would only add per-frame bytes. Use four CONTINUATIONs each
        // 16 KiB on stream 1; the per-stream cap will trip on the 1st
        // because 60+16 = 76 > 64.
        let chunk = vec![0u8; 16_383];
        let f = continuation_frame_on_stream(&chunk, 1, false);
        let (h, p) = parse_first_frame(&f);
        let err = r.ingest(&h, &p).unwrap_err();
        // Per-stream cap trips first by construction; that is the
        // reachable bound for a single-stream chain. Verify bound is
        // checked.
        assert!(matches!(
            err,
            H2ParseError::ReassemblerOverflow {
                kind: ReassemblerOverflowKind::PerStreamBlock
            }
        ));

        // Now verify the aggregate-cap path with synthetic state: set
        // total_in_flight just under cap, attempt a fresh HEADERS that
        // would push over. (Per-stream cap doesn't trip because the
        // new stream has 0 bytes pending.)
        let mut r2 = H2StreamReassembler::new();
        r2.total_in_flight = REASSEMBLER_TOTAL_IN_FLIGHT_MAX - 100;
        let block = vec![0u8; 200];
        let f = headers_frame_on_stream_no_end(&block, 5);
        let (h, p) = parse_first_frame(&f);
        let err = r2.ingest(&h, &p).unwrap_err();
        assert!(matches!(
            err,
            H2ParseError::ReassemblerOverflow {
                kind: ReassemblerOverflowKind::TotalInFlight
            }
        ));
    }

    #[test]
    fn ingest_too_many_concurrent_streams_returns_overflow() {
        let mut r = H2StreamReassembler::new();
        // Synthetic fill: jam the blocks map to the cap directly,
        // then attempt to open a fresh HEADERS. (Real protocol can't
        // sustain >1 open HEADERS reassembly at once per §6.10, so
        // this path is purely defence in depth.)
        for i in 1..=REASSEMBLER_MAX_CONCURRENT_STREAMS as u32 {
            r.blocks.insert(i * 2 + 1, vec![]);
        }
        // active_stream stays None so the new-HEADERS path runs.
        let block = hpack_literal_indexed_name(1, "api.example.com");
        let f = headers_frame_on_stream(&block, 999);
        let (h, p) = parse_first_frame(&f);
        let err = r.ingest(&h, &p).unwrap_err();
        assert!(matches!(
            err,
            H2ParseError::ReassemblerOverflow {
                kind: ReassemblerOverflowKind::ConcurrentStreams
            }
        ));
    }

    // ─────────────────────────────────────────────────────────────────────
    // scope: H2ConnectionDecoder unit tests
    // ─────────────────────────────────────────────────────────────────────

    #[test]
    fn feed_returns_decoded_authority_for_complete_headers_on_stream_1() {
        let mut d = H2ConnectionDecoder::new();
        let block = hpack_literal_indexed_name(1, "api.example.com");
        let f = headers_frame(&block);
        let (h, p) = parse_first_frame(&f);
        let out = d.feed_frame(&h, &p).unwrap().unwrap();
        assert_eq!(out.stream_id, 1);
        assert_eq!(out.authority.as_deref(), Some("api.example.com"));
        assert!(!out.via_dynamic_table);
        assert!(!out.via_huffman);
    }

    #[test]
    fn feed_returns_two_decoded_blocks_for_two_streams_interleaved_via_continuation() {
        let mut d = H2ConnectionDecoder::new();
        let block_a = hpack_literal_indexed_name(1, "api.example.com");
        let block_b = hpack_literal_indexed_name(1, "api.other.com");
        // Stream 1 single-frame HEADERS.
        let fa = headers_frame_on_stream(&block_a, 1);
        let (ha, pa) = parse_first_frame(&fa);
        let out_a = d.feed_frame(&ha, &pa).unwrap().unwrap();
        assert_eq!(out_a.stream_id, 1);
        assert_eq!(out_a.authority.as_deref(), Some("api.example.com"));
        // Stream 3 fragmented across HEADERS + CONTINUATION.
        let mid = block_b.len() / 2;
        let (b1, b2) = block_b.split_at(mid);
        let f1 = headers_frame_on_stream_no_end(b1, 3);
        let f2 = continuation_frame_on_stream(b2, 3, true);
        let (h1, p1) = parse_first_frame(&f1);
        let (h2, p2) = parse_first_frame(&f2);
        assert!(d.feed_frame(&h1, &p1).unwrap().is_none());
        let out_b = d.feed_frame(&h2, &p2).unwrap().unwrap();
        assert_eq!(out_b.stream_id, 3);
        assert_eq!(out_b.authority.as_deref(), Some("api.other.com"));
    }

    #[test]
    fn feed_returns_provenance_when_huffman_used() {
        let mut d = H2ConnectionDecoder::new();
        let block = hpack_literal_indexed_name_huffman(1, "api.example.com");
        let f = headers_frame(&block);
        let (h, p) = parse_first_frame(&f);
        let out = d.feed_frame(&h, &p).unwrap().unwrap();
        assert!(out.via_huffman);
    }

    #[test]
    fn feed_returns_provenance_when_dynamic_table_used() {
        let mut d = H2ConnectionDecoder::new();
        // Block 1: literal-with-incremental-indexing :authority =
        // "api.example.com" — populates dynamic index 62.
        let block1 = hpack_literal_indexed_name(1, "api.example.com");
        let f1 = headers_frame_on_stream(&block1, 1);
        let (h1, p1) = parse_first_frame(&f1);
        let _ = d.feed_frame(&h1, &p1).unwrap();
        // Block 2: indexed-header at dynamic index 62 — provenance
        // should reflect dynamic-table reuse.
        let block2 = vec![0x80 | 62];
        let f2 = headers_frame_on_stream(&block2, 3);
        let (h2, p2) = parse_first_frame(&f2);
        let out = d.feed_frame(&h2, &p2).unwrap().unwrap();
        assert!(out.via_dynamic_table);
        assert_eq!(out.authority.as_deref(), Some("api.example.com"));
    }

    #[test]
    fn feed_returns_none_for_data_frame() {
        let mut d = H2ConnectionDecoder::new();
        let f = data_frame_on_stream(b"payload", 1);
        let (h, p) = parse_first_frame(&f);
        assert!(d.feed_frame(&h, &p).unwrap().is_none());
    }

    #[test]
    fn feed_propagates_hpack_decoder_state_across_streams() {
        // Same as feed_returns_provenance_when_dynamic_table_used but
        // explicitly asserts on the decoded value matching what the
        // first stream pushed into the dynamic table.
        let mut d = H2ConnectionDecoder::new();
        let block1 = hpack_literal_indexed_name(1, "shared.example.com");
        let f1 = headers_frame_on_stream(&block1, 1);
        let (h1, p1) = parse_first_frame(&f1);
        let r1 = d.feed_frame(&h1, &p1).unwrap().unwrap();
        assert_eq!(r1.authority.as_deref(), Some("shared.example.com"));
        let block2 = vec![0x80 | 62];
        let f2 = headers_frame_on_stream(&block2, 5);
        let (h2, p2) = parse_first_frame(&f2);
        let r2 = d.feed_frame(&h2, &p2).unwrap().unwrap();
        assert_eq!(r2.authority.as_deref(), Some("shared.example.com"));
        assert_eq!(r2.stream_id, 5);
    }

    #[test]
    fn feed_propagates_reassembler_overflow_per_stream() {
        let mut d = H2ConnectionDecoder::new();
        let chunk = vec![0u8; 16_383];
        let f1 = headers_frame_on_stream_no_end(&chunk, 1);
        let (h1, p1) = parse_first_frame(&f1);
        d.feed_frame(&h1, &p1).unwrap();
        for _ in 0..3 {
            let f = continuation_frame_on_stream(&chunk, 1, false);
            let (h, p) = parse_first_frame(&f);
            d.feed_frame(&h, &p).unwrap();
        }
        let f = continuation_frame_on_stream(&chunk, 1, false);
        let (h, p) = parse_first_frame(&f);
        let err = d.feed_frame(&h, &p).unwrap_err();
        assert!(matches!(
            err,
            H2ParseError::ReassemblerOverflow {
                kind: ReassemblerOverflowKind::PerStreamBlock
            }
        ));
    }

    #[test]
    fn feed_handles_settings_frame_size_update_to_dynamic_table() {
        // SETTINGS frames pass through the reassembler (Ok(None)) and
        // never reach the HPACK decoder; the dynamic-table SIZE update
        // is an HPACK §6.3 representation that lives INSIDE a HEADERS
        // block, not a SETTINGS frame. So a SETTINGS frame should be a
        // no-op for the decoder, and a HEADERS block carrying a §6.3
        // size update should successfully update the dynamic table.
        let mut d = H2ConnectionDecoder::new();
        // Empty SETTINGS: pass-through.
        let settings = empty_settings_frame();
        let (sh, sp) = parse_first_frame(&settings);
        assert!(d.feed_frame(&sh, &sp).unwrap().is_none());
        // HEADERS carrying a §6.3 size update (set to 0) followed by
        // an indexed header (no :authority) — should succeed without
        // surfacing an authority.
        let block: Vec<u8> = vec![
            0x20,     // 001 00000 — size update, new max = 0
            0x80 | 2, // indexed :method GET
        ];
        let f = headers_frame_on_stream(&block, 1);
        let (h, p) = parse_first_frame(&f);
        let out = d.feed_frame(&h, &p).unwrap().unwrap();
        assert_eq!(out.stream_id, 1);
        assert!(out.authority.is_none());
    }

    #[test]
    fn feed_returns_none_authority_when_headers_block_lacks_pseudo_header() {
        let mut d = H2ConnectionDecoder::new();
        // Indexed-header at static index 2 (`:method GET`) — no :authority.
        let block = vec![0x80 | 2];
        let f = headers_frame_on_stream(&block, 1);
        let (h, p) = parse_first_frame(&f);
        let out = d.feed_frame(&h, &p).unwrap().unwrap();
        assert_eq!(out.stream_id, 1);
        assert!(out.authority.is_none());
    }

    #[test]
    fn feed_returns_block_for_continuation_after_headers_without_end_headers() {
        // Same as feed_returns_two_decoded_blocks but on stream 1 with
        // a single fragmented stream — verifies the CONTINUATION-only
        // completion path returns Some(decoded).
        let mut d = H2ConnectionDecoder::new();
        let block_full = hpack_literal_indexed_name(1, "api.example.com");
        let mid = block_full.len() / 2;
        let (a, b) = block_full.split_at(mid);
        let f1 = headers_frame_on_stream_no_end(a, 1);
        let f2 = continuation_frame_on_stream(b, 1, true);
        let (h1, p1) = parse_first_frame(&f1);
        let (h2, p2) = parse_first_frame(&f2);
        assert!(d.feed_frame(&h1, &p1).unwrap().is_none());
        let out = d.feed_frame(&h2, &p2).unwrap().unwrap();
        assert_eq!(out.stream_id, 1);
        assert_eq!(out.authority.as_deref(), Some("api.example.com"));
    }
}