h2/codec/
framed_read.rs

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

h2/codec/
framed_read.rs
