simoom_qpack/
decoder.rs

1//! QPACK decoder (RFC 9204 sections 3, 4.3, 4.4 and 4.5).
2
3use std::collections::VecDeque;
4
5use crate::prim::{get_int, get_span, put_int, span_bytes, Short, STATIC_TABLE};
6use crate::{Error, Field};
7
8/// Result of decoding a field section.
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub enum Decoded {
11    /// The decoded field lines.
12    Fields(Vec<Field>),
13    /// The section references entries not yet received; it is kept and returned by
14    /// [`Decoder::on_encoder_stream`] once they arrive.
15    Blocked,
16}
17
18/// A QPACK decoder for one HTTP/3 connection.
19#[derive(Debug)]
20pub struct Decoder {
21    max_cap: u64,
22    max_blocked: u64,
23    max_section: u64,
24    cap: u64,
25    size: u64,
26    entries: VecDeque<Field>,
27    inserted: u64,
28    /// Insert count the encoder knows we have (Section Acks and Insert Count Increments sent).
29    acked: u64,
30    blocked: Vec<(u64, u64, Vec<u8>)>,
31    enc_buf: Vec<u8>,
32    out: Vec<u8>,
33}
34
35impl Decoder {
36    /// `max_table_capacity` and `max_blocked_streams` are the values announced in our SETTINGS
37    /// (SETTINGS_QPACK_MAX_TABLE_CAPACITY, SETTINGS_QPACK_BLOCKED_STREAMS);
38    /// `max_field_section_size` bounds each decoded section (RFC 9114 4.2.2).
39    pub fn new(max_table_capacity: u64, max_blocked_streams: u64, max_field_section_size: u64) -> Self {
40        Decoder {
41            max_cap: max_table_capacity.min(crate::MAX_INT),
42            max_blocked: max_blocked_streams,
43            max_section: max_field_section_size,
44            cap: 0,
45            size: 0,
46            entries: VecDeque::new(),
47            inserted: 0,
48            acked: 0,
49            blocked: Vec::new(),
50            enc_buf: Vec::new(),
51            out: Vec::new(),
52        }
53    }
54
55    /// Number of entries inserted into the dynamic table so far.
56    pub fn insert_count(&self) -> u64 {
57        self.inserted
58    }
59
60    /// Current dynamic table size in bytes (entries counted as name + value + 32).
61    pub fn table_size(&self) -> u64 {
62        self.size
63    }
64
65    /// Number of streams currently blocked.
66    pub fn blocked_streams(&self) -> usize {
67        let mut n = 0;
68        for (i, b) in self.blocked.iter().enumerate() {
69            if !self.blocked[..i].iter().any(|x| x.0 == b.0) {
70                n += 1;
71            }
72        }
73        n
74    }
75
76    /// Decodes a field section received on request stream `stream_id`.
77    pub fn decode(&mut self, stream_id: u64, block: &[u8]) -> Result<Decoded, Error> {
78        if block.len() as u64 > self.max_section.saturating_add(64) {
79            return Err(Error::FieldSectionTooLarge);
80        }
81        match self.decode_block(block) {
82            Ok((fields, ric)) => {
83                self.ack_section(stream_id, ric);
84                Ok(Decoded::Fields(fields))
85            }
86            Err(Error::Blocked) => {
87                if !self.blocked.iter().any(|b| b.0 == stream_id)
88                    && self.blocked_streams() as u64 >= self.max_blocked
89                {
90                    return Err(Error::DecompressionFailed);
91                }
92                let ric = self.required_insert_count(block)?;
93                self.blocked.push((stream_id, ric, block.to_vec()));
94                Ok(Decoded::Blocked)
95            }
96            Err(e) => Err(e),
97        }
98    }
99
100    fn ack_section(&mut self, stream_id: u64, ric: u64) {
101        if ric > 0 {
102            put_int(&mut self.out, 0x80, 7, stream_id);
103            self.acked = self.acked.max(ric);
104        }
105    }
106
107    /// Processes bytes from the peer's encoder stream. Returns the streams that became
108    /// unblocked, in the order they blocked, with their decode results.
109    #[allow(clippy::type_complexity)]
110    pub fn on_encoder_stream(&mut self, data: &[u8]) -> Result<Vec<(u64, Result<Vec<Field>, Error>)>, Error> {
111        let mut buf = std::mem::take(&mut self.enc_buf);
112        buf.extend_from_slice(data);
113        let mut pos = 0;
114        let r = loop {
115            let start = pos;
116            match self.instruction(&buf, &mut pos) {
117                Ok(()) => {}
118                Err(Short::Incomplete) => {
119                    pos = start;
120                    break Ok(());
121                }
122                Err(Short::Invalid) => break Err(Error::EncoderStream),
123            }
124        };
125        r?;
126        buf.drain(..pos);
127        // a partial instruction can hold at most one entry, Huffman-coded at <= 30 bits a byte
128        if buf.len() as u64 > self.max_cap.saturating_mul(4).saturating_add(64) {
129            return Err(Error::EncoderStream);
130        }
131        self.enc_buf = buf;
132        let mut done = Vec::new();
133        if self.blocked.iter().any(|b| b.1 <= self.inserted) {
134            let blocked = std::mem::take(&mut self.blocked);
135            for (sid, ric, block) in blocked {
136                if ric > self.inserted {
137                    self.blocked.push((sid, ric, block));
138                    continue;
139                }
140                let r = self.decode_block(&block).map(|(f, ric)| {
141                    self.ack_section(sid, ric);
142                    f
143                });
144                done.push((sid, r));
145            }
146        }
147        Ok(done)
148    }
149
150    /// The stream was reset or abandoned before its field sections were fully processed: drops
151    /// its blocked sections and queues a Stream Cancellation. Call it only for streams whose
152    /// sections were not all decoded.
153    pub fn cancel_stream(&mut self, stream_id: u64) {
154        self.blocked.retain(|b| b.0 != stream_id);
155        if self.max_cap > 0 {
156            put_int(&mut self.out, 0x40, 6, stream_id);
157        }
158    }
159
160    /// Appends pending decoder-stream instructions to `out`: Section Acknowledgments, Stream
161    /// Cancellations, then an Insert Count Increment for inserts not yet acknowledged.
162    pub fn take_decoder_stream(&mut self, out: &mut Vec<u8>) {
163        out.append(&mut self.out);
164        if self.inserted > self.acked {
165            put_int(out, 0x00, 6, self.inserted - self.acked);
166            self.acked = self.inserted;
167        }
168    }
169
170    fn dropped(&self) -> u64 {
171        self.inserted - self.entries.len() as u64
172    }
173
174    fn entry(&self, abs: u64) -> Option<&Field> {
175        let d = self.dropped();
176        if abs < d || abs >= self.inserted {
177            return None;
178        }
179        self.entries.get((abs - d) as usize)
180    }
181
182    fn insert(&mut self, f: Field) -> Result<(), Short> {
183        let size = f.size();
184        if size > self.cap {
185            return Err(Short::Invalid);
186        }
187        while self.size + size > self.cap {
188            let e = self.entries.pop_front().ok_or(Short::Invalid)?;
189            self.size -= e.size();
190        }
191        self.size += size;
192        self.entries.push_back(f);
193        self.inserted += 1;
194        Ok(())
195    }
196
197    /// Parses and applies one encoder-stream instruction (RFC 9204 4.3). Nothing changes when
198    /// the instruction is incomplete.
199    fn instruction(&mut self, buf: &[u8], pos: &mut usize) -> Result<(), Short> {
200        let b = *buf.get(*pos).ok_or(Short::Incomplete)?;
201        let limit = self.max_cap.saturating_mul(4).saturating_add(64);
202        if b & 0x80 != 0 {
203            // Insert with Name Reference
204            let idx = get_int(buf, pos, 6)?;
205            let v = get_span(buf, pos, 7)?;
206            if (v.end - v.start) as u64 > limit {
207                return Err(Short::Invalid);
208            }
209            let name = if b & 0x40 != 0 {
210                STATIC_TABLE.get(idx as usize).ok_or(Short::Invalid)?.0.as_bytes().to_vec()
211            } else {
212                let abs = self.inserted.checked_sub(idx + 1).ok_or(Short::Invalid)?;
213                self.entry(abs).ok_or(Short::Invalid)?.name.clone()
214            };
215            let value = span_bytes(buf, v)?;
216            self.insert(Field { name, value })
217        } else if b & 0x40 != 0 {
218            // Insert with Literal Name
219            let n = get_span(buf, pos, 5)?;
220            let v = get_span(buf, pos, 7)?;
221            if (v.end - n.start) as u64 > limit {
222                return Err(Short::Invalid);
223            }
224            let name = span_bytes(buf, n)?;
225            let value = span_bytes(buf, v)?;
226            self.insert(Field { name, value })
227        } else if b & 0x20 != 0 {
228            // Set Dynamic Table Capacity
229            let cap = get_int(buf, pos, 5)?;
230            if cap > self.max_cap {
231                return Err(Short::Invalid);
232            }
233            self.cap = cap;
234            while self.size > self.cap {
235                let e = self.entries.pop_front().ok_or(Short::Invalid)?;
236                self.size -= e.size();
237            }
238            Ok(())
239        } else {
240            // Duplicate
241            let idx = get_int(buf, pos, 5)?;
242            let abs = self.inserted.checked_sub(idx + 1).ok_or(Short::Invalid)?;
243            let f = self.entry(abs).ok_or(Short::Invalid)?.clone();
244            self.insert(f)
245        }
246    }
247
248    /// Decodes the Required Insert Count of a section (RFC 9204 4.5.1.1).
249    fn required_insert_count(&self, block: &[u8]) -> Result<u64, Error> {
250        let mut pos = 0;
251        let enc = get_int(block, &mut pos, 8).map_err(|_| Error::DecompressionFailed)?;
252        if enc == 0 {
253            return Ok(0);
254        }
255        let max_entries = self.max_cap / 32;
256        let full_range = 2 * max_entries;
257        if enc > full_range {
258            return Err(Error::DecompressionFailed);
259        }
260        let max_value = self.inserted + max_entries;
261        let max_wrapped = (max_value / full_range) * full_range;
262        let mut ric = max_wrapped + enc - 1;
263        if ric > max_value {
264            if ric <= full_range {
265                return Err(Error::DecompressionFailed);
266            }
267            ric -= full_range;
268        }
269        if ric == 0 {
270            return Err(Error::DecompressionFailed);
271        }
272        Ok(ric)
273    }
274
275    /// Decodes a whole section. Returns `Error::Blocked` when it needs entries not yet received.
276    fn decode_block(&self, block: &[u8]) -> Result<(Vec<Field>, u64), Error> {
277        let ric = self.required_insert_count(block)?;
278        if ric > self.inserted {
279            return Err(Error::Blocked);
280        }
281        let fail = |_| Error::DecompressionFailed;
282        let mut pos = 0;
283        get_int(block, &mut pos, 8).map_err(fail)?;
284        let sign = *block.get(pos).ok_or(Error::DecompressionFailed)? & 0x80 != 0;
285        let delta = get_int(block, &mut pos, 7).map_err(fail)?;
286        let base = if sign {
287            ric.checked_sub(delta + 1).ok_or(Error::DecompressionFailed)?
288        } else {
289            ric.checked_add(delta).ok_or(Error::DecompressionFailed)?
290        };
291        let mut fields = Vec::new();
292        let mut total = 0u64;
293        let mut max_ref: Option<u64> = None;
294        // resolves a dynamic absolute index, which must lie below the Required Insert Count
295        let mut dyn_ref = |abs: Option<u64>| -> Result<&Field, Error> {
296            let abs = abs.filter(|&a| a < ric).ok_or(Error::DecompressionFailed)?;
297            max_ref = Some(max_ref.map_or(abs, |m: u64| m.max(abs)));
298            self.entry(abs).ok_or(Error::DecompressionFailed)
299        };
300        let static_ref = |idx: u64| -> Result<(&'static [u8], &'static [u8]), Error> {
301            let (n, v) = STATIC_TABLE.get(idx as usize).ok_or(Error::DecompressionFailed)?;
302            Ok((n.as_bytes(), v.as_bytes()))
303        };
304        while pos < block.len() {
305            let b = block[pos];
306            let field = if b & 0x80 != 0 {
307                // Indexed Field Line
308                let idx = get_int(block, &mut pos, 6).map_err(fail)?;
309                if b & 0x40 != 0 {
310                    let (n, v) = static_ref(idx)?;
311                    Field::new(n, v)
312                } else {
313                    dyn_ref(base.checked_sub(idx + 1))?.clone()
314                }
315            } else if b & 0x40 != 0 {
316                // Literal Field Line with Name Reference
317                let idx = get_int(block, &mut pos, 4).map_err(fail)?;
318                let name = if b & 0x10 != 0 {
319                    static_ref(idx)?.0.to_vec()
320                } else {
321                    dyn_ref(base.checked_sub(idx + 1))?.name.clone()
322                };
323                let v = get_span(block, &mut pos, 7).map_err(fail)?;
324                Field { name, value: span_bytes(block, v).map_err(fail)? }
325            } else if b & 0x20 != 0 {
326                // Literal Field Line with Literal Name
327                let n = get_span(block, &mut pos, 3).map_err(fail)?;
328                let v = get_span(block, &mut pos, 7).map_err(fail)?;
329                Field { name: span_bytes(block, n).map_err(fail)?, value: span_bytes(block, v).map_err(fail)? }
330            } else if b & 0x10 != 0 {
331                // Indexed Field Line with Post-Base Index
332                let idx = get_int(block, &mut pos, 4).map_err(fail)?;
333                dyn_ref(base.checked_add(idx))?.clone()
334            } else {
335                // Literal Field Line with Post-Base Name Reference
336                let idx = get_int(block, &mut pos, 3).map_err(fail)?;
337                let name = dyn_ref(base.checked_add(idx))?.name.clone();
338                let v = get_span(block, &mut pos, 7).map_err(fail)?;
339                Field { name, value: span_bytes(block, v).map_err(fail)? }
340            };
341            total += field.size();
342            if total > self.max_section {
343                return Err(Error::FieldSectionTooLarge);
344            }
345            fields.push(field);
346        }
347        // the Required Insert Count must be exactly what the references need (RFC 9204 2.2.3)
348        if ric != max_ref.map_or(0, |m| m + 1) {
349            return Err(Error::DecompressionFailed);
350        }
351        Ok((fields, ric))
352    }
353}
simoom_qpack/decoder.rs

simoom_qpack/
decoder.rs
