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: usize = (header_max / frame_max).max(1);
113 // Some padding for imperfectly packed frames
114 // 25% without floats
115 let padding: usize = 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: &(dyn Error + Send + Sync)) = err.get_ref() {
414 if custom.is::<LengthDelimitedCodecError>() {
415 return Error::library_go_away(reason: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: Headers) => {
456 headers.set_end_headers();
457 headers.into()
458 }
459 Continuable::PushPromise(mut push: PushPromise) => {
460 push.set_end_headers();
461 push.into()
462 }
463 }
464 }
465}
466