fastapi_http/
http2.rs

1//! HTTP/2 (RFC 7540) framing + HPACK (RFC 7541).
2//!
3//! Scope (bd-2c9t):
4//! - Implement enough of HTTP/2 to accept cleartext prior-knowledge connections (h2c)
5//! - Provide a correct HPACK decoder (including Huffman) for request headers
6//!
7//! This module intentionally avoids Tokio/Hyper and uses only asupersync for async I/O.
8
9use asupersync::io::{AsyncRead, AsyncWrite, ReadBuf};
10use asupersync::net::TcpStream;
11use std::collections::VecDeque;
12use std::future::poll_fn;
13use std::io;
14use std::pin::Pin;
15use std::sync::OnceLock;
16use std::task::Poll;
17
18/// HTTP/2 connection preface for prior-knowledge cleartext.
19pub const PREFACE: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
20
21/// HTTP/2 frame type (RFC 7540).
22#[derive(Debug, Clone, Copy, PartialEq, Eq)]
23#[repr(u8)]
24pub enum FrameType {
25    Data = 0x0,
26    Headers = 0x1,
27    Priority = 0x2,
28    RstStream = 0x3,
29    Settings = 0x4,
30    PushPromise = 0x5,
31    Ping = 0x6,
32    Goaway = 0x7,
33    WindowUpdate = 0x8,
34    Continuation = 0x9,
35    Unknown = 0xFF,
36}
37
38impl FrameType {
39    #[must_use]
40    pub fn from_u8(v: u8) -> Self {
41        match v {
42            0x0 => Self::Data,
43            0x1 => Self::Headers,
44            0x2 => Self::Priority,
45            0x3 => Self::RstStream,
46            0x4 => Self::Settings,
47            0x5 => Self::PushPromise,
48            0x6 => Self::Ping,
49            0x7 => Self::Goaway,
50            0x8 => Self::WindowUpdate,
51            0x9 => Self::Continuation,
52            _ => Self::Unknown,
53        }
54    }
55}
56
57/// A parsed HTTP/2 frame header.
58#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59pub struct FrameHeader {
60    pub length: u32, // 24-bit
61    pub frame_type: u8,
62    pub flags: u8,
63    pub stream_id: u32, // 31-bit
64}
65
66impl FrameHeader {
67    pub const LEN: usize = 9;
68
69    #[must_use]
70    pub fn frame_type(&self) -> FrameType {
71        FrameType::from_u8(self.frame_type)
72    }
73
74    #[must_use]
75    pub fn is_stream_zero(&self) -> bool {
76        self.stream_id == 0
77    }
78}
79
80/// A full HTTP/2 frame.
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub struct Frame {
83    pub header: FrameHeader,
84    pub payload: Vec<u8>,
85}
86
87#[derive(Debug)]
88pub enum Http2Error {
89    Io(io::Error),
90    Protocol(&'static str),
91    Hpack(HpackError),
92}
93
94impl std::fmt::Display for Http2Error {
95    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
96        match self {
97            Self::Io(e) => write!(f, "http2 I/O error: {e}"),
98            Self::Protocol(m) => write!(f, "http2 protocol error: {m}"),
99            Self::Hpack(e) => write!(f, "hpack error: {e}"),
100        }
101    }
102}
103
104impl std::error::Error for Http2Error {
105    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
106        match self {
107            Self::Io(e) => Some(e),
108            Self::Hpack(e) => Some(e),
109            Self::Protocol(_) => None,
110        }
111    }
112}
113
114impl From<io::Error> for Http2Error {
115    fn from(e: io::Error) -> Self {
116        Self::Io(e)
117    }
118}
119
120impl From<HpackError> for Http2Error {
121    fn from(e: HpackError) -> Self {
122        Self::Hpack(e)
123    }
124}
125
126/// A simple framed HTTP/2 I/O wrapper.
127#[derive(Debug)]
128pub struct FramedH2 {
129    stream: TcpStream,
130    rx: Vec<u8>,
131}
132
133impl FramedH2 {
134    #[must_use]
135    pub fn new(stream: TcpStream, buffered: Vec<u8>) -> Self {
136        Self {
137            stream,
138            rx: buffered,
139        }
140    }
141
142    /// Read the next HTTP/2 frame.
143    pub async fn read_frame(&mut self, max_frame_size: u32) -> Result<Frame, Http2Error> {
144        let header_bytes = self.read_exact(FrameHeader::LEN).await?;
145        let length = ((u32::from(header_bytes[0])) << 16)
146            | ((u32::from(header_bytes[1])) << 8)
147            | u32::from(header_bytes[2]);
148        let frame_type = header_bytes[3];
149        let flags = header_bytes[4];
150        let stream_id = u32::from_be_bytes([
151            header_bytes[5],
152            header_bytes[6],
153            header_bytes[7],
154            header_bytes[8],
155        ]) & 0x7FFF_FFFF;
156
157        if length > max_frame_size {
158            return Err(Http2Error::Protocol("frame length exceeds max_frame_size"));
159        }
160
161        let payload = self.read_exact(length as usize).await?;
162        Ok(Frame {
163            header: FrameHeader {
164                length,
165                frame_type,
166                flags,
167                stream_id,
168            },
169            payload,
170        })
171    }
172
173    /// Write an HTTP/2 frame.
174    pub async fn write_frame(
175        &mut self,
176        frame_type: FrameType,
177        flags: u8,
178        stream_id: u32,
179        payload: &[u8],
180    ) -> Result<(), Http2Error> {
181        if stream_id & 0x8000_0000 != 0 {
182            return Err(Http2Error::Protocol("reserved bit set in stream_id"));
183        }
184        let len = u32::try_from(payload.len())
185            .map_err(|_| Http2Error::Protocol("payload length too large"))?;
186        if len > 0x00FF_FFFF {
187            return Err(Http2Error::Protocol("payload length exceeds 24-bit limit"));
188        }
189
190        let mut out = Vec::with_capacity(FrameHeader::LEN + payload.len());
191        out.push(((len >> 16) & 0xff) as u8);
192        out.push(((len >> 8) & 0xff) as u8);
193        out.push((len & 0xff) as u8);
194        out.push(frame_type as u8);
195        out.push(flags);
196        out.extend_from_slice(&(stream_id & 0x7FFF_FFFF).to_be_bytes());
197        out.extend_from_slice(payload);
198
199        write_all(&mut self.stream, &out).await?;
200        flush(&mut self.stream).await?;
201        Ok(())
202    }
203
204    async fn read_exact(&mut self, n: usize) -> io::Result<Vec<u8>> {
205        while self.rx.len() < n {
206            let mut tmp = vec![0u8; 8192];
207            let read = read_once(&mut self.stream, &mut tmp).await?;
208            if read == 0 {
209                return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "EOF"));
210            }
211            self.rx.extend_from_slice(&tmp[..read]);
212        }
213        Ok(self.rx.drain(..n).collect())
214    }
215}
216
217async fn read_once(stream: &mut TcpStream, buffer: &mut [u8]) -> io::Result<usize> {
218    poll_fn(|cx| {
219        let mut read_buf = ReadBuf::new(buffer);
220        match Pin::new(&mut *stream).poll_read(cx, &mut read_buf) {
221            Poll::Ready(Ok(())) => Poll::Ready(Ok(read_buf.filled().len())),
222            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
223            Poll::Pending => Poll::Pending,
224        }
225    })
226    .await
227}
228
229async fn write_all(stream: &mut TcpStream, mut buf: &[u8]) -> io::Result<()> {
230    while !buf.is_empty() {
231        let n = poll_fn(|cx| Pin::new(&mut *stream).poll_write(cx, buf)).await?;
232        if n == 0 {
233            return Err(io::Error::new(io::ErrorKind::WriteZero, "write zero"));
234        }
235        buf = &buf[n..];
236    }
237    Ok(())
238}
239
240async fn flush(stream: &mut TcpStream) -> io::Result<()> {
241    poll_fn(|cx| Pin::new(&mut *stream).poll_flush(cx)).await
242}
243
244// =============================================================================
245// HPACK decoder (RFC 7541)
246// =============================================================================
247
248#[derive(Debug, Clone, PartialEq, Eq)]
249pub enum HpackError {
250    InvalidInteger,
251    InvalidString,
252    InvalidIndex,
253    InvalidHuffman,
254    DynamicTableSizeUpdateOutOfRange,
255    HeaderListTooLarge,
256}
257
258impl std::fmt::Display for HpackError {
259    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
260        write!(f, "{self:?}")
261    }
262}
263
264impl std::error::Error for HpackError {}
265
266#[derive(Debug, Clone, PartialEq, Eq)]
267struct HeaderField {
268    name: Vec<u8>,
269    value: Vec<u8>,
270    size: usize,
271}
272
273impl HeaderField {
274    fn new(name: Vec<u8>, value: Vec<u8>) -> Self {
275        let size = 32 + name.len() + value.len();
276        Self { name, value, size }
277    }
278}
279
280/// HPACK decoder with a dynamic table.
281#[derive(Debug)]
282pub struct HpackDecoder {
283    dynamic: VecDeque<HeaderField>,
284    dynamic_size: usize,
285    dynamic_max_size: usize,
286    max_header_list_size: usize,
287}
288
289/// Decoded HPACK headers (name, value) as raw bytes.
290pub type HeaderList = Vec<(Vec<u8>, Vec<u8>)>;
291
292impl Default for HpackDecoder {
293    fn default() -> Self {
294        Self::new()
295    }
296}
297
298impl HpackDecoder {
299    #[must_use]
300    pub fn new() -> Self {
301        Self {
302            dynamic: VecDeque::new(),
303            dynamic_size: 0,
304            dynamic_max_size: 4096,
305            max_header_list_size: 64 * 1024,
306        }
307    }
308
309    pub fn set_dynamic_table_max_size(&mut self, n: usize) {
310        self.dynamic_max_size = n;
311        self.evict_to_max();
312    }
313
314    pub fn set_max_header_list_size(&mut self, n: usize) {
315        self.max_header_list_size = n;
316    }
317
318    pub fn decode(&mut self, block: &[u8]) -> Result<HeaderList, HpackError> {
319        let mut out: HeaderList = Vec::new();
320        let mut i = 0usize;
321
322        while i < block.len() {
323            let b = block[i];
324
325            if (b & 0x80) != 0 {
326                // Indexed Header Field Representation (1xxxxxxx)
327                let (index, used) = decode_integer(&block[i..], 7)?;
328                i += used;
329                let (name, value) = self.get_indexed(index)?;
330                out.push((name, value));
331                continue;
332            }
333
334            if (b & 0xC0) == 0x40 {
335                // Literal Header Field with Incremental Indexing (01xxxxxx)
336                let (name, value, used) = self.decode_literal(&block[i..], 6)?;
337                i += used;
338                self.insert_dynamic(name.clone(), value.clone());
339                out.push((name, value));
340                continue;
341            }
342
343            if (b & 0xE0) == 0x20 {
344                // Dynamic Table Size Update (001xxxxx)
345                let (new_size, used) = decode_integer(&block[i..], 5)?;
346                i += used;
347                if new_size > self.dynamic_max_size {
348                    return Err(HpackError::DynamicTableSizeUpdateOutOfRange);
349                }
350                self.set_dynamic_table_max_size(new_size);
351                continue;
352            }
353
354            // Literal Header Field without Indexing / Never Indexed.
355            // 0000xxxx and 0001xxxx share the same literal payload shape.
356            let (name, value, used) = self.decode_literal(&block[i..], 4)?;
357            i += used;
358            out.push((name, value));
359        }
360
361        let total_list_bytes: usize = out.iter().map(|(n, v)| n.len() + v.len() + 32).sum();
362        if total_list_bytes > self.max_header_list_size {
363            return Err(HpackError::HeaderListTooLarge);
364        }
365
366        Ok(out)
367    }
368
369    fn decode_literal(
370        &mut self,
371        buf: &[u8],
372        name_prefix_bits: u8,
373    ) -> Result<(Vec<u8>, Vec<u8>, usize), HpackError> {
374        // Name: either indexed (prefix integer) or literal string (0 + string)
375        let first = buf[0];
376        let name_index_prefix_mask = (1u8 << name_prefix_bits) - 1;
377        let name_index = usize::from(first & name_index_prefix_mask);
378
379        let mut used = 0usize;
380        let name = if name_index == 0 {
381            used += 1;
382            let (name_bytes, n_used) = decode_string(&buf[used..])?;
383            used += n_used;
384            name_bytes
385        } else {
386            let (index, n_used) = decode_integer(buf, name_prefix_bits)?;
387            used += n_used;
388            let (name, _value) = self.get_indexed(index)?;
389            name
390        };
391
392        let (value, v_used) = decode_string(&buf[used..])?;
393        used += v_used;
394        Ok((name, value, used))
395    }
396
397    fn get_indexed(&self, index: usize) -> Result<(Vec<u8>, Vec<u8>), HpackError> {
398        if index == 0 {
399            return Err(HpackError::InvalidIndex);
400        }
401        let static_len = STATIC_TABLE.len();
402        if index <= static_len {
403            let (n, v) = STATIC_TABLE[index - 1];
404            return Ok((n.to_vec(), v.to_vec()));
405        }
406        let dyn_index = index - static_len - 1;
407        let field = self
408            .dynamic
409            .get(dyn_index)
410            .ok_or(HpackError::InvalidIndex)?;
411        Ok((field.name.clone(), field.value.clone()))
412    }
413
414    fn insert_dynamic(&mut self, name: Vec<u8>, value: Vec<u8>) {
415        let field = HeaderField::new(name, value);
416        if field.size > self.dynamic_max_size {
417            self.dynamic.clear();
418            self.dynamic_size = 0;
419            return;
420        }
421        self.dynamic.push_front(field);
422        self.dynamic_size = self.dynamic.iter().map(|f| f.size).sum();
423        self.evict_to_max();
424    }
425
426    fn evict_to_max(&mut self) {
427        while self.dynamic_size > self.dynamic_max_size {
428            let Some(back) = self.dynamic.pop_back() else {
429                self.dynamic_size = 0;
430                break;
431            };
432            self.dynamic_size = self.dynamic_size.saturating_sub(back.size);
433        }
434    }
435}
436
437fn decode_integer(buf: &[u8], prefix_bits: u8) -> Result<(usize, usize), HpackError> {
438    if buf.is_empty() || prefix_bits == 0 || prefix_bits > 8 {
439        return Err(HpackError::InvalidInteger);
440    }
441    let prefix_max = (1usize << prefix_bits) - 1;
442    let mut value = usize::from(buf[0] & (prefix_max as u8));
443    if value < prefix_max {
444        return Ok((value, 1));
445    }
446    let mut m = 0usize;
447    let mut idx = 1usize;
448    loop {
449        let b = *buf.get(idx).ok_or(HpackError::InvalidInteger)?;
450        idx += 1;
451        value = value
452            .checked_add((usize::from(b & 0x7f)) << m)
453            .ok_or(HpackError::InvalidInteger)?;
454        if (b & 0x80) == 0 {
455            break;
456        }
457        m = m.checked_add(7).ok_or(HpackError::InvalidInteger)?;
458        if m > 63 {
459            return Err(HpackError::InvalidInteger);
460        }
461    }
462    Ok((value, idx))
463}
464
465fn decode_string(buf: &[u8]) -> Result<(Vec<u8>, usize), HpackError> {
466    if buf.is_empty() {
467        return Err(HpackError::InvalidString);
468    }
469    let huffman = (buf[0] & 0x80) != 0;
470    let (len, used) = decode_integer(buf, 7)?;
471    let start = used;
472    let end = start.checked_add(len).ok_or(HpackError::InvalidString)?;
473    let s = buf.get(start..end).ok_or(HpackError::InvalidString)?;
474    if huffman {
475        let decoded = huffman_decode(s)?;
476        Ok((decoded, end))
477    } else {
478        Ok((s.to_vec(), end))
479    }
480}
481
482// =============================================================================
483// HPACK encoder (minimal, for HTTP/2 responses)
484// =============================================================================
485
486fn encode_integer(out: &mut Vec<u8>, first: u8, prefix_bits: u8, mut value: usize) {
487    let prefix_max = (1usize << prefix_bits) - 1;
488    if value < prefix_max {
489        out.push(first | (value as u8));
490        return;
491    }
492
493    out.push(first | (prefix_max as u8));
494    value -= prefix_max;
495    while value >= 128 {
496        out.push(((value & 0x7f) as u8) | 0x80);
497        value >>= 7;
498    }
499    out.push(value as u8);
500}
501
502fn encode_string(out: &mut Vec<u8>, bytes: &[u8]) {
503    // Huffman bit = 0 (no huffman); length uses a 7-bit prefixed integer.
504    encode_integer(out, 0x00, 7, bytes.len());
505    out.extend_from_slice(bytes);
506}
507
508/// Encode a literal header field without indexing (RFC 7541).
509///
510/// This is intentionally minimal:
511/// - never uses huffman
512/// - never indexes into dynamic table
513/// - always encodes the name as a literal (name-index = 0)
514pub fn hpack_encode_literal_without_indexing(out: &mut Vec<u8>, name: &[u8], value: &[u8]) {
515    // Literal Header Field without Indexing:
516    // 0000xxxx where xxxx is the name index (4-bit prefix integer).
517    // We always use name-index = 0 (literal name follows).
518    encode_integer(out, 0x00, 4, 0);
519    encode_string(out, name);
520    encode_string(out, value);
521}
522
523// Static table: RFC 7541 Appendix A.
524// Kept as bytes to avoid UTF-8 assumptions (header fields are bytes in HTTP/2).
525const STATIC_TABLE: [(&[u8], &[u8]); 61] = [
526    (b":authority", b""),
527    (b":method", b"GET"),
528    (b":method", b"POST"),
529    (b":path", b"/"),
530    (b":path", b"/index.html"),
531    (b":scheme", b"http"),
532    (b":scheme", b"https"),
533    (b":status", b"200"),
534    (b":status", b"204"),
535    (b":status", b"206"),
536    (b":status", b"304"),
537    (b":status", b"400"),
538    (b":status", b"404"),
539    (b":status", b"500"),
540    (b"accept-charset", b""),
541    (b"accept-encoding", b"gzip, deflate"),
542    (b"accept-language", b""),
543    (b"accept-ranges", b""),
544    (b"accept", b""),
545    (b"access-control-allow-origin", b""),
546    (b"age", b""),
547    (b"allow", b""),
548    (b"authorization", b""),
549    (b"cache-control", b""),
550    (b"content-disposition", b""),
551    (b"content-encoding", b""),
552    (b"content-language", b""),
553    (b"content-length", b""),
554    (b"content-location", b""),
555    (b"content-range", b""),
556    (b"content-type", b""),
557    (b"cookie", b""),
558    (b"date", b""),
559    (b"etag", b""),
560    (b"expect", b""),
561    (b"expires", b""),
562    (b"from", b""),
563    (b"host", b""),
564    (b"if-match", b""),
565    (b"if-modified-since", b""),
566    (b"if-none-match", b""),
567    (b"if-range", b""),
568    (b"if-unmodified-since", b""),
569    (b"last-modified", b""),
570    (b"link", b""),
571    (b"location", b""),
572    (b"max-forwards", b""),
573    (b"proxy-authenticate", b""),
574    (b"proxy-authorization", b""),
575    (b"range", b""),
576    (b"referer", b""),
577    (b"refresh", b""),
578    (b"retry-after", b""),
579    (b"server", b""),
580    (b"set-cookie", b""),
581    (b"strict-transport-security", b""),
582    (b"transfer-encoding", b""),
583    (b"user-agent", b""),
584    (b"vary", b""),
585    (b"via", b""),
586    (b"www-authenticate", b""),
587];
588
589#[derive(Debug, Clone, Copy)]
590struct HuffmanNode {
591    left: Option<usize>,
592    right: Option<usize>,
593    sym: Option<u16>,
594}
595
596fn huffman_tree() -> &'static Vec<HuffmanNode> {
597    static TREE: OnceLock<Vec<HuffmanNode>> = OnceLock::new();
598    TREE.get_or_init(|| {
599        let mut nodes = vec![HuffmanNode {
600            left: None,
601            right: None,
602            sym: None,
603        }];
604
605        for (sym, (&code, &bits)) in HUFFMAN_CODES.iter().zip(HUFFMAN_BITS.iter()).enumerate() {
606            let mut cur = 0usize;
607            for bit_index in (0..bits).rev() {
608                let bit = (code >> bit_index) & 1;
609                let next_idx = if bit == 0 {
610                    nodes[cur].left
611                } else {
612                    nodes[cur].right
613                };
614
615                cur = if let Some(idx) = next_idx {
616                    idx
617                } else {
618                    let idx = nodes.len();
619                    nodes.push(HuffmanNode {
620                        left: None,
621                        right: None,
622                        sym: None,
623                    });
624                    if bit == 0 {
625                        nodes[cur].left = Some(idx);
626                    } else {
627                        nodes[cur].right = Some(idx);
628                    }
629                    idx
630                };
631            }
632            nodes[cur].sym = Some(u16::try_from(sym).unwrap_or(256));
633        }
634
635        nodes
636    })
637}
638
639fn eos_prefix_nodes() -> &'static Vec<bool> {
640    static NODES: OnceLock<Vec<bool>> = OnceLock::new();
641    NODES.get_or_init(|| {
642        let tree = huffman_tree();
643        let mut is_prefix = vec![false; tree.len()];
644        let eos_code = HUFFMAN_CODES[256];
645        let eos_bits = HUFFMAN_BITS[256];
646
647        let mut cur = 0usize;
648        is_prefix[cur] = true;
649        for bit_index in (0..eos_bits).rev() {
650            let bit = (eos_code >> bit_index) & 1;
651            cur = if bit == 0 {
652                tree[cur].left.expect("eos left")
653            } else {
654                tree[cur].right.expect("eos right")
655            };
656            if cur >= is_prefix.len() {
657                break;
658            }
659            is_prefix[cur] = true;
660        }
661        is_prefix
662    })
663}
664
665fn huffman_decode(bytes: &[u8]) -> Result<Vec<u8>, HpackError> {
666    let tree = huffman_tree();
667    let eos_prefix = eos_prefix_nodes();
668
669    let mut out = Vec::with_capacity(bytes.len());
670    let mut cur = 0usize;
671
672    for &byte in bytes {
673        for bit_shift in (0..8).rev() {
674            let bit = (byte >> bit_shift) & 1;
675            cur = if bit == 0 {
676                tree[cur].left.ok_or(HpackError::InvalidHuffman)?
677            } else {
678                tree[cur].right.ok_or(HpackError::InvalidHuffman)?
679            };
680            if let Some(sym) = tree[cur].sym {
681                if sym == 256 {
682                    // EOS must not appear in the decoded stream.
683                    return Err(HpackError::InvalidHuffman);
684                }
685                out.push(u8::try_from(sym).map_err(|_| HpackError::InvalidHuffman)?);
686                cur = 0;
687            }
688        }
689    }
690
691    // Validate padding: the terminal state must be a prefix of EOS.
692    if cur != 0 && !eos_prefix.get(cur).copied().unwrap_or(false) {
693        return Err(HpackError::InvalidHuffman);
694    }
695
696    Ok(out)
697}
698
699// Huffman table: RFC 7541 Appendix B.
700// code (MSB-first) + bit-length for symbols 0..=256 (EOS).
701//
702// This is large but stable; keep it as constants so the decoder stays dependency-free.
703#[rustfmt::skip]
704#[allow(clippy::unreadable_literal)]
705const HUFFMAN_CODES: [u32; 257] = [
706    0x1ff8,0x7fffd8,0xfffffe2,0xfffffe3,0xfffffe4,0xfffffe5,0xfffffe6,0xfffffe7,
707    0xfffffe8,0xffffea,0x3ffffffc,0xfffffe9,0xfffffea,0x3ffffffd,0xfffffeb,0xfffffec,
708    0xfffffed,0xfffffee,0xfffffef,0xffffff0,0xffffff1,0xffffff2,0x3ffffffe,0xffffff3,
709    0xffffff4,0xffffff5,0xffffff6,0xffffff7,0xffffff8,0xffffff9,0xffffffa,0xffffffb,
710    0x14,0x3f8,0x3f9,0xffa,0x1ff9,0x15,0xf8,0x7fa,0x3fa,0x3fb,0xf9,0x7fb,0xfa,
711    0x16,0x17,0x18,0x0,0x1,0x2,0x19,0x1a,0x1b,0x1c,0x1d,0x1e,0x1f,0x5c,0xfb,
712    0x7ffc,0x20,0xffb,0x3fc,0x1ffa,0x21,0x5d,0x5e,0x5f,0x60,0x61,0x62,0x63,
713    0x64,0x65,0x66,0x67,0x68,0x69,0x6a,0x6b,0x6c,0x6d,0x6e,0x6f,0x70,0x71,
714    0x72,0xfc,0x73,0xfd,0x1ffb,0x7fff0,0x1ffc,0x3ffc,0x22,0x7ffd,0x3,0x23,0x4,
715    0x24,0x5,0x25,0x26,0x27,0x6,0x74,0x75,0x28,0x29,0x2a,0x7,0x2b,0x76,0x2c,
716    0x8,0x9,0x2d,0x77,0x78,0x79,0x7a,0x7b,0x7ffe,0x7fc,0x3ffd,0x1ffd,0xffffffc,
717    0xfffe6,0x3fffd2,0xfffe7,0xfffe8,0x3fffd3,0x3fffd4,0x3fffd5,0x7fffd9,0x3fffd6,
718    0x7fffda,0x7fffdb,0x7fffdc,0x7fffdd,0x7fffde,0xffffeb,0x7fffdf,0xffffec,0xffffed,
719    0x3fffd7,0x7fffe0,0xffffee,0x7fffe1,0x7fffe2,0x7fffe3,0x7fffe4,0x1fffdc,0x3fffd8,
720    0x7fffe5,0x3fffd9,0x7fffe6,0x7fffe7,0xffffef,0x3fffda,0x1fffdd,0xfffe9,0x3fffdb,
721    0x3fffdc,0x7fffe8,0x7fffe9,0x1fffde,0x7fffea,0x3fffdd,0x3fffde,0xfffff0,0x1fffdf,
722    0x3fffdf,0x7fffeb,0x7fffec,0x1fffe0,0x1fffe1,0x3fffe0,0x1fffe2,0x7fffed,0x3fffe1,
723    0x7fffee,0x7fffef,0xfffea,0x3fffe2,0x3fffe3,0x3fffe4,0x7ffff0,0x3fffe5,0x3fffe6,
724    0x7ffff1,0x3ffffe0,0x3ffffe1,0xfffeb,0x7fff1,0x3fffe7,0x7ffff2,0x3fffe8,0x1ffffec,
725    0x3ffffe2,0x3ffffe3,0x3ffffe4,0x7ffffde,0x7ffffdf,0x3ffffe5,0xfffff1,0x1ffffed,
726    0x7fff2,0x1fffe3,0x3ffffe6,0x7ffffe0,0x7ffffe1,0x3ffffe7,0x7ffffe2,0xfffff2,
727    0x1fffe4,0x1fffe5,0x3ffffe8,0x3ffffe9,0xffffffd,0x7ffffe3,0x7ffffe4,0x7ffffe5,
728    0xfffec,0xfffff3,0xfffed,0x1fffe6,0x3fffe9,0x1fffe7,0x1fffe8,0x7ffff3,0x3fffea,
729    0x3fffeb,0x1ffffee,0x1ffffef,0xfffff4,0xfffff5,0x3ffffea,0x7ffff4,0x3ffffeb,
730    0x7ffffe6,0x3ffffec,0x3ffffed,0x7ffffe7,0x7ffffe8,0x7ffffe9,0x7ffffea,0x7ffffeb,
731    0xffffffe,0x7ffffec,0x7ffffed,0x7ffffee,0x7ffffef,0x7fffff0,0x3ffffee,0x3fffffff,
732];
733
734#[rustfmt::skip]
735const HUFFMAN_BITS: [u8; 257] = [
736    13,23,28,28,28,28,28,28,28,24,30,28,28,30,28,28,
737    28,28,28,28,28,28,30,28,28,28,28,28,28,28,28,28,
738    6,10,10,12,13,6,8,11,10,10,8,11,8,6,6,6,
739    5,5,5,6,6,6,6,6,6,6,7,8,15,6,12,10,
740    13,6,7,7,7,7,7,7,7,7,7,7,7,7,7,7,
741    7,7,7,7,7,7,7,7,8,7,8,13,19,13,14,6,
742    15,5,6,5,6,5,6,6,6,5,7,7,6,6,6,5,
743    6,7,6,5,5,6,7,7,7,7,7,15,11,14,13,28,
744    20,22,20,20,22,22,22,23,22,23,23,23,23,23,24,23,
745    24,24,22,23,24,23,23,23,23,21,22,23,22,23,23,24,
746    22,21,20,22,22,23,23,21,23,22,22,24,21,22,23,23,
747    21,21,22,21,23,22,23,23,20,22,22,22,23,22,22,23,
748    26,26,20,19,22,23,22,25,26,26,26,27,27,26,24,25,
749    19,21,26,27,27,26,27,24,21,21,26,26,28,27,27,27,
750    20,24,20,21,22,21,21,23,22,22,25,25,24,24,26,23,
751    26,27,26,26,27,27,27,27,27,28,27,27,27,27,27,26,
752    30,
753];
754
755// =============================================================================
756// HTTP/2 Flow Control (RFC 7540 §6.9)
757// =============================================================================
758
759/// RFC 7540 default initial window size (65,535 bytes).
760pub const DEFAULT_INITIAL_WINDOW_SIZE: u32 = 65_535;
761
762/// When the consumed (unreported) bytes exceed this fraction of the initial
763/// window size, a WINDOW_UPDATE is sent. Using half the window keeps the
764/// sender from stalling while avoiding excessive WINDOW_UPDATE traffic.
765const WINDOW_UPDATE_THRESHOLD_DIVISOR: u32 = 2;
766
767/// Tracks HTTP/2 flow-control receive windows at connection and stream level.
768///
769/// The receiver (server) decrements windows when DATA frames arrive and sends
770/// WINDOW_UPDATE frames when enough data has been consumed to keep the sender
771/// from stalling.
772#[derive(Debug)]
773pub struct H2FlowControl {
774    /// Connection-level receive window remaining.
775    conn_window: i64,
776    /// How many bytes have been consumed at connection level since the last
777    /// WINDOW_UPDATE was sent.
778    conn_consumed: u32,
779    /// The initial window size advertised (or default). Used to compute the
780    /// threshold at which a WINDOW_UPDATE should be emitted.
781    initial_window_size: u32,
782    /// Connection-level send window remaining (peer's receive window).
783    send_conn_window: i64,
784    /// Per-stream initial send window (from peer's SETTINGS_INITIAL_WINDOW_SIZE).
785    peer_initial_window_size: u32,
786}
787
788impl H2FlowControl {
789    /// Create a new flow-control tracker with the RFC default initial window.
790    #[must_use]
791    pub fn new() -> Self {
792        Self {
793            conn_window: i64::from(DEFAULT_INITIAL_WINDOW_SIZE),
794            conn_consumed: 0,
795            initial_window_size: DEFAULT_INITIAL_WINDOW_SIZE,
796            send_conn_window: i64::from(DEFAULT_INITIAL_WINDOW_SIZE),
797            peer_initial_window_size: DEFAULT_INITIAL_WINDOW_SIZE,
798        }
799    }
800
801    /// Update the initial window size (from SETTINGS_INITIAL_WINDOW_SIZE).
802    /// This only affects future streams; the connection-level window is
803    /// independent of this setting per RFC 7540 §6.9.2.
804    pub fn set_initial_window_size(&mut self, size: u32) {
805        self.initial_window_size = size;
806    }
807
808    /// Record that `n` bytes of DATA payload were received on the connection.
809    /// Returns the connection-level WINDOW_UPDATE increment to send, or 0 if
810    /// no update is needed yet.
811    pub fn data_received_connection(&mut self, n: u32) -> u32 {
812        self.conn_window -= i64::from(n);
813        self.conn_consumed += n;
814
815        let threshold = self.initial_window_size / WINDOW_UPDATE_THRESHOLD_DIVISOR;
816        if self.conn_consumed >= threshold {
817            let increment = self.conn_consumed;
818            self.conn_window += i64::from(increment);
819            self.conn_consumed = 0;
820            increment
821        } else {
822            0
823        }
824    }
825
826    /// Compute a stream-level WINDOW_UPDATE increment after `total_received`
827    /// bytes have been received for a stream. Returns the increment to send,
828    /// or 0 if no update is needed.
829    ///
830    /// For simplicity, the server sends a stream-level WINDOW_UPDATE when the
831    /// consumed bytes exceed the threshold. Since streams are short-lived
832    /// request bodies, we track this per-call rather than storing per-stream
833    /// state.
834    pub fn stream_window_update(&self, total_received: u32) -> u32 {
835        let threshold = self.initial_window_size / WINDOW_UPDATE_THRESHOLD_DIVISOR;
836        if total_received >= threshold {
837            total_received
838        } else {
839            0
840        }
841    }
842
843    /// The initial window size for new streams.
844    #[must_use]
845    pub fn initial_window_size(&self) -> u32 {
846        self.initial_window_size
847    }
848
849    // --- Send-side flow control ---
850
851    /// Set the peer's initial window size (from peer's SETTINGS_INITIAL_WINDOW_SIZE).
852    /// This determines the send window for new streams.
853    pub fn set_peer_initial_window_size(&mut self, size: u32) {
854        self.peer_initial_window_size = size;
855    }
856
857    /// The peer's initial window size for new streams (send window).
858    #[must_use]
859    pub fn peer_initial_window_size(&self) -> u32 {
860        self.peer_initial_window_size
861    }
862
863    /// The maximum number of bytes we can send on the connection right now.
864    #[must_use]
865    pub fn send_conn_window(&self) -> i64 {
866        self.send_conn_window
867    }
868
869    /// Record a WINDOW_UPDATE received from the peer for the connection
870    /// (stream_id == 0). Increases the connection-level send window.
871    pub fn peer_window_update_connection(&mut self, increment: u32) {
872        self.send_conn_window += i64::from(increment);
873    }
874
875    /// Consume `n` bytes of the connection-level send window.
876    pub fn consume_send_conn_window(&mut self, n: u32) {
877        self.send_conn_window -= i64::from(n);
878    }
879}
880
881impl Default for H2FlowControl {
882    fn default() -> Self {
883        Self::new()
884    }
885}
886
887#[cfg(test)]
888mod tests {
889    use super::*;
890
891    #[test]
892    fn hpack_rfc_vector_first_request() {
893        // RFC 7541 C.2.1 "First Request" header block.
894        let block: [u8; 17] = [
895            0x82, 0x86, 0x84, 0x41, 0x8c, 0xf1, 0xe3, 0xc2, 0xe5, 0xf2, 0x3a, 0x6b, 0xa0, 0xab,
896            0x90, 0xf4, 0xff,
897        ];
898        let mut dec = HpackDecoder::new();
899        let headers = dec.decode(&block).unwrap();
900
901        assert!(headers.contains(&(b":method".to_vec(), b"GET".to_vec())));
902        assert!(headers.contains(&(b":scheme".to_vec(), b"http".to_vec())));
903        assert!(headers.contains(&(b":path".to_vec(), b"/".to_vec())));
904        assert!(headers.contains(&(b":authority".to_vec(), b"www.example.com".to_vec())));
905    }
906
907    #[test]
908    fn hpack_rejects_eos_symbol() {
909        // A single EOS symbol (all-ones 30-bit code) is invalid. Construct a buffer
910        // that decodes to EOS by providing exactly the EOS code bytes.
911        //
912        // EOS code = 0x3fffffff (30 bits). Provide it as 4 bytes with huffman flag.
913        let buf: [u8; 5] = [0x80 | 4, 0xff, 0xff, 0xff, 0xff];
914        let res = decode_string(&buf);
915        assert!(matches!(res, Err(HpackError::InvalidHuffman)));
916    }
917
918    #[test]
919    fn preface_constant_matches_rfc() {
920        assert_eq!(PREFACE, b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n");
921        assert_eq!(PREFACE.len(), 24);
922    }
923
924    #[test]
925    fn flow_control_no_update_below_threshold() {
926        let mut fc = H2FlowControl::new();
927        // Receive 1000 bytes — well below half of 65535.
928        let increment = fc.data_received_connection(1000);
929        assert_eq!(increment, 0);
930    }
931
932    #[test]
933    fn flow_control_emits_update_at_threshold() {
934        let mut fc = H2FlowControl::new();
935        // Threshold is 65535 / 2 = 32767. Receive exactly that amount.
936        let increment = fc.data_received_connection(32_767);
937        assert_eq!(increment, 32_767);
938    }
939
940    #[test]
941    fn flow_control_accumulates_across_calls() {
942        let mut fc = H2FlowControl::new();
943        // Two small receives that together exceed threshold.
944        assert_eq!(fc.data_received_connection(20_000), 0);
945        assert_eq!(fc.data_received_connection(20_000), 40_000);
946    }
947
948    #[test]
949    fn flow_control_resets_consumed_after_update() {
950        let mut fc = H2FlowControl::new();
951        assert_eq!(fc.data_received_connection(40_000), 40_000);
952        // After reset, small amounts should not trigger again.
953        assert_eq!(fc.data_received_connection(1_000), 0);
954    }
955
956    #[test]
957    fn flow_control_stream_below_threshold() {
958        let fc = H2FlowControl::new();
959        assert_eq!(fc.stream_window_update(1_000), 0);
960    }
961
962    #[test]
963    fn flow_control_stream_at_threshold() {
964        let fc = H2FlowControl::new();
965        assert_eq!(fc.stream_window_update(32_767), 32_767);
966    }
967
968    #[test]
969    fn flow_control_custom_initial_window() {
970        let mut fc = H2FlowControl::new();
971        fc.set_initial_window_size(100_000);
972        // Threshold is now 50_000.
973        assert_eq!(fc.data_received_connection(49_999), 0);
974        assert_eq!(fc.data_received_connection(1), 50_000);
975    }
976}
fastapi_http/http2.rs

fastapi_http/
http2.rs
