h2/codec/
framed_read.rs

1use crate::frame::{self, Frame, Kind, Reason};
2use crate::frame::{
3    DEFAULT_MAX_FRAME_SIZE, DEFAULT_SETTINGS_HEADER_TABLE_SIZE, MAX_MAX_FRAME_SIZE,
4};
5use crate::proto::Error;
6
7use crate::hpack;
8
9use futures_core::Stream;
10
11use bytes::{Buf, BytesMut};
12
13use std::io;
14
15use std::pin::Pin;
16use std::task::{Context, Poll};
17use tokio::io::AsyncRead;
18use tokio_util::codec::FramedRead as InnerFramedRead;
19use tokio_util::codec::{LengthDelimitedCodec, LengthDelimitedCodecError};
20
21// 16 MB "sane default" taken from golang http2
22const DEFAULT_SETTINGS_MAX_HEADER_LIST_SIZE: usize = 16 << 20;
23
24#[derive(Debug)]
25pub struct FramedRead<T> {
26    inner: InnerFramedRead<T, LengthDelimitedCodec>,
27
28    decoder: FrameDecoder,
29}
30
31#[derive(Debug)]
32struct FrameDecoder {
33    // hpack decoder state
34    hpack: hpack::Decoder,
35
36    max_header_list_size: usize,
37
38    max_continuation_frames: usize,
39
40    partial: Option<Partial>,
41}
42
43/// Partially loaded headers frame
44#[derive(Debug)]
45struct Partial {
46    /// Empty frame
47    frame: Continuable,
48
49    /// Partial header payload
50    buf: BytesMut,
51
52    continuation_frames_count: usize,
53}
54
55#[derive(Debug)]
56enum Continuable {
57    Headers(frame::Headers),
58    PushPromise(frame::PushPromise),
59}
60
61impl<T> FramedRead<T> {
62    pub fn new(inner: InnerFramedRead<T, LengthDelimitedCodec>) -> FramedRead<T> {
63        let decoder = FrameDecoder::new(inner.decoder().max_frame_length());
64        FramedRead { inner, decoder }
65    }
66
67    pub fn get_ref(&self) -> &T {
68        self.inner.get_ref()
69    }
70
71    pub fn get_mut(&mut self) -> &mut T {
72        self.inner.get_mut()
73    }
74
75    /// Returns the current max frame size setting
76    #[inline]
77    pub fn max_frame_size(&self) -> usize {
78        self.inner.decoder().max_frame_length()
79    }
80
81    /// Updates the max frame size setting.
82    ///
83    /// Must be within 16,384 and 16,777,215.
84    #[inline]
85    pub fn set_max_frame_size(&mut self, val: usize) {
86        assert!(DEFAULT_MAX_FRAME_SIZE as usize <= val && val <= MAX_MAX_FRAME_SIZE as usize);
87        self.inner.decoder_mut().set_max_frame_length(val);
88        // Update max CONTINUATION frames too, since its based on this
89        self.decoder.set_max_frame_size(val);
90    }
91
92    /// Update the max header list size setting.
93    #[inline]
94    pub fn set_max_header_list_size(&mut self, val: usize) {
95        self.decoder
96            .set_max_header_list_size(val, self.max_frame_size());
97    }
98
99    /// Update the header table size setting.
100    #[inline]
101    pub fn set_header_table_size(&mut self, val: usize) {
102        self.decoder.set_header_table_size(val);
103    }
104}
105
106fn calc_max_continuation_frames(header_max: usize, frame_max: usize) -> usize {
107    // At least this many frames needed to use max header list size
108    let min_frames_for_list = (header_max / frame_max).max(1);
109    // Some padding for imperfectly packed frames
110    // 25% without floats
111    let padding = min_frames_for_list >> 2;
112    min_frames_for_list.saturating_add(padding).max(5)
113}
114
115impl FrameDecoder {
116    fn new(max_frame_size: usize) -> Self {
117        let max_header_list_size = DEFAULT_SETTINGS_MAX_HEADER_LIST_SIZE;
118        FrameDecoder {
119            hpack: hpack::Decoder::new(DEFAULT_SETTINGS_HEADER_TABLE_SIZE),
120            max_header_list_size,
121            max_continuation_frames: calc_max_continuation_frames(
122                max_header_list_size,
123                max_frame_size,
124            ),
125            partial: None,
126        }
127    }
128
129    fn set_max_frame_size(&mut self, val: usize) {
130        self.max_continuation_frames = calc_max_continuation_frames(self.max_header_list_size, val);
131    }
132
133    fn set_max_header_list_size(&mut self, val: usize, max_frame_size: usize) {
134        self.max_header_list_size = val;
135        // Update max CONTINUATION frames too, since its based on this
136        self.max_continuation_frames = calc_max_continuation_frames(val, max_frame_size);
137    }
138
139    fn set_header_table_size(&mut self, val: usize) {
140        self.hpack.queue_size_update(val);
141    }
142
143    fn decode(&mut self, bytes: BytesMut) -> Result<Option<Frame>, Error> {
144        decode_frame(self, bytes)
145    }
146}
147
148/// Decodes a frame.
149///
150/// This function is intentionally de-generified and outlined because it is very large.
151fn decode_frame(decoder: &mut FrameDecoder, mut bytes: BytesMut) -> Result<Option<Frame>, Error> {
152    let span = tracing::trace_span!("FramedRead::decode_frame", offset = bytes.len());
153    let _e = span.enter();
154
155    tracing::trace!("decoding frame from {}B", bytes.len());
156
157    // Parse the head
158    let head = frame::Head::parse(&bytes);
159
160    if decoder.partial.is_some() && head.kind() != Kind::Continuation {
161        proto_err!(conn: "expected CONTINUATION, got {:?}", head.kind());
162        return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
163    }
164
165    let kind = head.kind();
166
167    tracing::trace!(frame.kind = ?kind);
168
169    macro_rules! header_block {
170        ($frame:ident, $head:ident, $bytes:ident) => ({
171            // Drop the frame header
172            $bytes.advance(frame::HEADER_LEN);
173
174            // Parse the header frame w/o parsing the payload
175            let (mut frame, mut payload) = match frame::$frame::load($head, $bytes) {
176                Ok(res) => res,
177                Err(frame::Error::InvalidDependencyId) => {
178                    proto_err!(stream: "invalid HEADERS dependency ID");
179                    // A stream cannot depend on itself. An endpoint MUST
180                    // treat this as a stream error (Section 5.4.2) of type
181                    // `PROTOCOL_ERROR`.
182                    return Err(Error::library_reset($head.stream_id(), Reason::PROTOCOL_ERROR));
183                },
184                Err(e) => {
185                    proto_err!(conn: "failed to load frame; err={:?}", e);
186                    return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
187                }
188            };
189
190            let is_end_headers = frame.is_end_headers();
191
192            // Load the HPACK encoded headers
193            match frame.load_hpack(&mut payload, decoder.max_header_list_size, &mut decoder.hpack) {
194                Ok(_) => {},
195                Err(frame::Error::Hpack(hpack::DecoderError::NeedMore(_))) if !is_end_headers => {},
196                Err(frame::Error::MalformedMessage) => {
197                    let id = $head.stream_id();
198                    proto_err!(stream: "malformed header block; stream={:?}", id);
199                    return Err(Error::library_reset(id, Reason::PROTOCOL_ERROR));
200                },
201                Err(frame::Error::HeaderListWayTooLarge) => {
202                    proto_err!(conn: "decoded header list size over abuse limit");
203                    return Err(Error::library_go_away_data(
204                        Reason::ENHANCE_YOUR_CALM,
205                        "header_list_way_too_large",
206                    ));
207                },
208                Err(e) => {
209                    proto_err!(conn: "failed HPACK decoding; err={:?}", e);
210                    return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
211                }
212            }
213
214            if is_end_headers {
215                frame.into()
216            } else {
217                tracing::trace!("loaded partial header block");
218                // Defer returning the frame
219                decoder.partial = Some(Partial {
220                    frame: Continuable::$frame(frame),
221                    buf: payload,
222                    continuation_frames_count: 0,
223                });
224
225                return Ok(None);
226            }
227        });
228    }
229
230    let frame = match kind {
231        Kind::Settings => {
232            let res = frame::Settings::load(head, &bytes[frame::HEADER_LEN..]);
233
234            res.map_err(|e| {
235                proto_err!(conn: "failed to load SETTINGS frame; err={:?}", e);
236                Error::library_go_away(Reason::PROTOCOL_ERROR)
237            })?
238            .into()
239        }
240        Kind::Ping => {
241            let res = frame::Ping::load(head, &bytes[frame::HEADER_LEN..]);
242
243            res.map_err(|e| {
244                proto_err!(conn: "failed to load PING frame; err={:?}", e);
245                Error::library_go_away(Reason::PROTOCOL_ERROR)
246            })?
247            .into()
248        }
249        Kind::WindowUpdate => {
250            let res = frame::WindowUpdate::load(head, &bytes[frame::HEADER_LEN..]);
251
252            res.map_err(|e| {
253                proto_err!(conn: "failed to load WINDOW_UPDATE frame; err={:?}", e);
254                Error::library_go_away(Reason::PROTOCOL_ERROR)
255            })?
256            .into()
257        }
258        Kind::Data => {
259            bytes.advance(frame::HEADER_LEN);
260            let res = frame::Data::load(head, bytes.freeze());
261
262            // TODO: Should this always be connection level? Probably not...
263            res.map_err(|e| {
264                proto_err!(conn: "failed to load DATA frame; err={:?}", e);
265                Error::library_go_away(Reason::PROTOCOL_ERROR)
266            })?
267            .into()
268        }
269        Kind::Headers => header_block!(Headers, head, bytes),
270        Kind::Reset => {
271            let res = frame::Reset::load(head, &bytes[frame::HEADER_LEN..]);
272            res.map_err(|e| {
273                proto_err!(conn: "failed to load RESET frame; err={:?}", e);
274                Error::library_go_away(Reason::PROTOCOL_ERROR)
275            })?
276            .into()
277        }
278        Kind::GoAway => {
279            let res = frame::GoAway::load(head, &bytes[frame::HEADER_LEN..]);
280            res.map_err(|e| {
281                proto_err!(conn: "failed to load GO_AWAY frame; err={:?}", e);
282                Error::library_go_away(Reason::PROTOCOL_ERROR)
283            })?
284            .into()
285        }
286        Kind::PushPromise => header_block!(PushPromise, head, bytes),
287        Kind::Priority => {
288            if head.stream_id() == 0 {
289                // Invalid stream identifier
290                proto_err!(conn: "invalid stream ID 0");
291                return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
292            }
293
294            match frame::Priority::load(head, &bytes[frame::HEADER_LEN..]) {
295                Ok(frame) => frame.into(),
296                Err(frame::Error::InvalidDependencyId) => {
297                    // A stream cannot depend on itself. An endpoint MUST
298                    // treat this as a stream error (Section 5.4.2) of type
299                    // `PROTOCOL_ERROR`.
300                    let id = head.stream_id();
301                    proto_err!(stream: "PRIORITY invalid dependency ID; stream={:?}", id);
302                    return Err(Error::library_reset(id, Reason::PROTOCOL_ERROR));
303                }
304                Err(e) => {
305                    proto_err!(conn: "failed to load PRIORITY frame; err={:?};", e);
306                    return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
307                }
308            }
309        }
310        Kind::Continuation => {
311            let is_end_headers = (head.flag() & 0x4) == 0x4;
312
313            let mut partial = match decoder.partial.take() {
314                Some(partial) => partial,
315                None => {
316                    proto_err!(conn: "received unexpected CONTINUATION frame");
317                    return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
318                }
319            };
320
321            // The stream identifiers must match
322            if partial.frame.stream_id() != head.stream_id() {
323                proto_err!(conn: "CONTINUATION frame stream ID does not match previous frame stream ID");
324                return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
325            }
326
327            // Check for CONTINUATION flood
328            if is_end_headers {
329                partial.continuation_frames_count = 0;
330            } else {
331                let cnt = partial.continuation_frames_count + 1;
332                if cnt > decoder.max_continuation_frames {
333                    tracing::debug!(
334                        "too_many_continuations, max = {}",
335                        decoder.max_continuation_frames
336                    );
337                    return Err(Error::library_go_away_data(
338                        Reason::ENHANCE_YOUR_CALM,
339                        "too_many_continuations",
340                    ));
341                } else {
342                    partial.continuation_frames_count = cnt;
343                }
344            }
345
346            // Extend the buf
347            if partial.buf.is_empty() {
348                partial.buf = bytes.split_off(frame::HEADER_LEN);
349            } else {
350                if partial.frame.is_over_size() {
351                    // If there was left over bytes previously, they may be
352                    // needed to continue decoding, even though we will
353                    // be ignoring this frame. This is done to keep the HPACK
354                    // decoder state up-to-date.
355                    //
356                    // Still, we need to be careful, because if a malicious
357                    // attacker were to try to send a gigantic string, such
358                    // that it fits over multiple header blocks, we could
359                    // grow memory uncontrollably again, and that'd be a shame.
360                    //
361                    // Instead, we use a simple heuristic to determine if
362                    // we should continue to ignore decoding, or to tell
363                    // the attacker to go away.
364                    if partial.buf.len() + bytes.len() > decoder.max_header_list_size {
365                        proto_err!(conn: "CONTINUATION frame header block size over ignorable limit");
366                        return Err(Error::library_go_away(Reason::COMPRESSION_ERROR));
367                    }
368                }
369                partial.buf.extend_from_slice(&bytes[frame::HEADER_LEN..]);
370            }
371
372            match partial.frame.load_hpack(
373                &mut partial.buf,
374                decoder.max_header_list_size,
375                &mut decoder.hpack,
376            ) {
377                Ok(_) => {}
378                Err(frame::Error::Hpack(hpack::DecoderError::NeedMore(_))) if !is_end_headers => {}
379                Err(frame::Error::MalformedMessage) => {
380                    let id = head.stream_id();
381                    proto_err!(stream: "malformed CONTINUATION frame; stream={:?}", id);
382                    return Err(Error::library_reset(id, Reason::PROTOCOL_ERROR));
383                }
384                Err(frame::Error::HeaderListWayTooLarge) => {
385                    proto_err!(conn: "decoded CONTINUATION header list size over abuse limit");
386                    return Err(Error::library_go_away_data(
387                        Reason::ENHANCE_YOUR_CALM,
388                        "header_list_way_too_large",
389                    ));
390                }
391                Err(e) => {
392                    proto_err!(conn: "failed HPACK decoding; err={:?}", e);
393                    return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
394                }
395            }
396
397            if is_end_headers {
398                partial.frame.into()
399            } else {
400                decoder.partial = Some(partial);
401                return Ok(None);
402            }
403        }
404        Kind::Unknown => {
405            // Unknown frames are ignored
406            return Ok(None);
407        }
408    };
409
410    Ok(Some(frame))
411}
412
413impl<T> Stream for FramedRead<T>
414where
415    T: AsyncRead + Unpin,
416{
417    type Item = Result<Frame, Error>;
418
419    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
420        let span = tracing::trace_span!("FramedRead::poll_next");
421        let _e = span.enter();
422        loop {
423            tracing::trace!("poll");
424            let bytes = match ready!(Pin::new(&mut self.inner).poll_next(cx)) {
425                Some(Ok(bytes)) => bytes,
426                Some(Err(e)) => return Poll::Ready(Some(Err(map_err(e)))),
427                None => return Poll::Ready(None),
428            };
429
430            tracing::trace!(read.bytes = bytes.len());
431            if let Some(frame) = self.decoder.decode(bytes)? {
432                tracing::debug!(?frame, "received");
433                return Poll::Ready(Some(Ok(frame)));
434            }
435        }
436    }
437}
438
439fn map_err(err: io::Error) -> Error {
440    if let io::ErrorKind::InvalidData = err.kind() {
441        if let Some(custom) = err.get_ref() {
442            if custom.is::<LengthDelimitedCodecError>() {
443                return Error::library_go_away(Reason::FRAME_SIZE_ERROR);
444            }
445        }
446    }
447    err.into()
448}
449
450// ===== impl Continuable =====
451
452impl Continuable {
453    fn stream_id(&self) -> frame::StreamId {
454        match *self {
455            Continuable::Headers(ref h) => h.stream_id(),
456            Continuable::PushPromise(ref p) => p.stream_id(),
457        }
458    }
459
460    fn is_over_size(&self) -> bool {
461        match *self {
462            Continuable::Headers(ref h) => h.is_over_size(),
463            Continuable::PushPromise(ref p) => p.is_over_size(),
464        }
465    }
466
467    fn load_hpack(
468        &mut self,
469        src: &mut BytesMut,
470        max_header_list_size: usize,
471        decoder: &mut hpack::Decoder,
472    ) -> Result<(), frame::Error> {
473        match *self {
474            Continuable::Headers(ref mut h) => h.load_hpack(src, max_header_list_size, decoder),
475            Continuable::PushPromise(ref mut p) => p.load_hpack(src, max_header_list_size, decoder),
476        }
477    }
478}
479
480impl<T> From<Continuable> for Frame<T> {
481    fn from(cont: Continuable) -> Self {
482        match cont {
483            Continuable::Headers(mut headers) => {
484                headers.set_end_headers();
485                headers.into()
486            }
487            Continuable::PushPromise(mut push) => {
488                push.set_end_headers();
489                push.into()
490            }
491        }
492    }
493}
h2/codec/framed_read.rs

h2/codec/
framed_read.rs
