diff options
Diffstat (limited to 'third_party/rust/neqo-http3/src/frames')
-rw-r--r-- | third_party/rust/neqo-http3/src/frames/hframe.rs | 227 | ||||
-rw-r--r-- | third_party/rust/neqo-http3/src/frames/mod.rs | 21 | ||||
-rw-r--r-- | third_party/rust/neqo-http3/src/frames/reader.rs | 280 | ||||
-rw-r--r-- | third_party/rust/neqo-http3/src/frames/tests/hframe.rs | 116 | ||||
-rw-r--r-- | third_party/rust/neqo-http3/src/frames/tests/mod.rs | 84 | ||||
-rw-r--r-- | third_party/rust/neqo-http3/src/frames/tests/reader.rs | 518 | ||||
-rw-r--r-- | third_party/rust/neqo-http3/src/frames/tests/wtframe.rs | 17 | ||||
-rw-r--r-- | third_party/rust/neqo-http3/src/frames/wtframe.rs | 62 |
8 files changed, 1325 insertions, 0 deletions
diff --git a/third_party/rust/neqo-http3/src/frames/hframe.rs b/third_party/rust/neqo-http3/src/frames/hframe.rs new file mode 100644 index 0000000000..83e69ba894 --- /dev/null +++ b/third_party/rust/neqo-http3/src/frames/hframe.rs @@ -0,0 +1,227 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{fmt::Debug, io::Write}; + +use neqo_common::{Decoder, Encoder}; +use neqo_crypto::random; +use neqo_transport::StreamId; + +use crate::{frames::reader::FrameDecoder, settings::HSettings, Error, Priority, Res}; + +pub(crate) type HFrameType = u64; + +pub const H3_FRAME_TYPE_DATA: HFrameType = 0x0; +pub const H3_FRAME_TYPE_HEADERS: HFrameType = 0x1; +pub const H3_FRAME_TYPE_CANCEL_PUSH: HFrameType = 0x3; +pub const H3_FRAME_TYPE_SETTINGS: HFrameType = 0x4; +pub const H3_FRAME_TYPE_PUSH_PROMISE: HFrameType = 0x5; +pub const H3_FRAME_TYPE_GOAWAY: HFrameType = 0x7; +pub const H3_FRAME_TYPE_MAX_PUSH_ID: HFrameType = 0xd; +pub const H3_FRAME_TYPE_PRIORITY_UPDATE_REQUEST: HFrameType = 0xf0700; +pub const H3_FRAME_TYPE_PRIORITY_UPDATE_PUSH: HFrameType = 0xf0701; + +pub const H3_RESERVED_FRAME_TYPES: &[HFrameType] = &[0x2, 0x6, 0x8, 0x9]; + +// data for DATA frame is not read into HFrame::Data. +#[derive(PartialEq, Eq, Debug)] +pub enum HFrame { + Data { + len: u64, // length of the data + }, + Headers { + header_block: Vec<u8>, + }, + CancelPush { + push_id: u64, + }, + Settings { + settings: HSettings, + }, + PushPromise { + push_id: u64, + header_block: Vec<u8>, + }, + Goaway { + stream_id: StreamId, + }, + MaxPushId { + push_id: u64, + }, + Grease, + PriorityUpdateRequest { + element_id: u64, + priority: Priority, + }, + PriorityUpdatePush { + element_id: u64, + priority: Priority, + }, +} + +impl HFrame { + fn get_type(&self) -> HFrameType { + match self { + Self::Data { .. } => H3_FRAME_TYPE_DATA, + Self::Headers { .. } => H3_FRAME_TYPE_HEADERS, + Self::CancelPush { .. } => H3_FRAME_TYPE_CANCEL_PUSH, + Self::Settings { .. } => H3_FRAME_TYPE_SETTINGS, + Self::PushPromise { .. } => H3_FRAME_TYPE_PUSH_PROMISE, + Self::Goaway { .. } => H3_FRAME_TYPE_GOAWAY, + Self::MaxPushId { .. } => H3_FRAME_TYPE_MAX_PUSH_ID, + Self::PriorityUpdateRequest { .. } => H3_FRAME_TYPE_PRIORITY_UPDATE_REQUEST, + Self::PriorityUpdatePush { .. } => H3_FRAME_TYPE_PRIORITY_UPDATE_PUSH, + Self::Grease => { + let r = random(7); + Decoder::from(&r).decode_uint(7).unwrap() * 0x1f + 0x21 + } + } + } + + pub fn encode(&self, enc: &mut Encoder) { + enc.encode_varint(self.get_type()); + + match self { + Self::Data { len } => { + // DATA frame only encode the length here. + enc.encode_varint(*len); + } + Self::Headers { header_block } => { + enc.encode_vvec(header_block); + } + Self::CancelPush { push_id } => { + enc.encode_vvec_with(|enc_inner| { + enc_inner.encode_varint(*push_id); + }); + } + Self::Settings { settings } => { + settings.encode_frame_contents(enc); + } + Self::PushPromise { + push_id, + header_block, + } => { + enc.encode_varint((header_block.len() + (Encoder::varint_len(*push_id))) as u64); + enc.encode_varint(*push_id); + enc.encode(header_block); + } + Self::Goaway { stream_id } => { + enc.encode_vvec_with(|enc_inner| { + enc_inner.encode_varint(stream_id.as_u64()); + }); + } + Self::MaxPushId { push_id } => { + enc.encode_vvec_with(|enc_inner| { + enc_inner.encode_varint(*push_id); + }); + } + Self::Grease => { + // Encode some number of random bytes. + let r = random(8); + enc.encode_vvec(&r[1..usize::from(1 + (r[0] & 0x7))]); + } + Self::PriorityUpdateRequest { + element_id, + priority, + } + | Self::PriorityUpdatePush { + element_id, + priority, + } => { + let mut update_frame = Encoder::new(); + update_frame.encode_varint(*element_id); + + let mut priority_enc: Vec<u8> = Vec::new(); + write!(priority_enc, "{priority}").unwrap(); + + update_frame.encode(&priority_enc); + enc.encode_varint(update_frame.len() as u64); + enc.encode(update_frame.as_ref()); + } + } + } +} + +impl FrameDecoder<HFrame> for HFrame { + fn frame_type_allowed(frame_type: u64) -> Res<()> { + if H3_RESERVED_FRAME_TYPES.contains(&frame_type) { + return Err(Error::HttpFrameUnexpected); + } + Ok(()) + } + + fn decode(frame_type: u64, frame_len: u64, data: Option<&[u8]>) -> Res<Option<HFrame>> { + if frame_type == H3_FRAME_TYPE_DATA { + Ok(Some(HFrame::Data { len: frame_len })) + } else if let Some(payload) = data { + let mut dec = Decoder::from(payload); + Ok(match frame_type { + H3_FRAME_TYPE_DATA => unreachable!("DATA frame has been handled already."), + H3_FRAME_TYPE_HEADERS => Some(HFrame::Headers { + header_block: dec.decode_remainder().to_vec(), + }), + H3_FRAME_TYPE_CANCEL_PUSH => Some(HFrame::CancelPush { + push_id: dec.decode_varint().ok_or(Error::HttpFrame)?, + }), + H3_FRAME_TYPE_SETTINGS => { + let mut settings = HSettings::default(); + settings.decode_frame_contents(&mut dec).map_err(|e| { + if e == Error::HttpSettings { + e + } else { + Error::HttpFrame + } + })?; + Some(HFrame::Settings { settings }) + } + H3_FRAME_TYPE_PUSH_PROMISE => Some(HFrame::PushPromise { + push_id: dec.decode_varint().ok_or(Error::HttpFrame)?, + header_block: dec.decode_remainder().to_vec(), + }), + H3_FRAME_TYPE_GOAWAY => Some(HFrame::Goaway { + stream_id: StreamId::new(dec.decode_varint().ok_or(Error::HttpFrame)?), + }), + H3_FRAME_TYPE_MAX_PUSH_ID => Some(HFrame::MaxPushId { + push_id: dec.decode_varint().ok_or(Error::HttpFrame)?, + }), + H3_FRAME_TYPE_PRIORITY_UPDATE_REQUEST | H3_FRAME_TYPE_PRIORITY_UPDATE_PUSH => { + let element_id = dec.decode_varint().ok_or(Error::HttpFrame)?; + let priority = dec.decode_remainder(); + let priority = Priority::from_bytes(priority)?; + if frame_type == H3_FRAME_TYPE_PRIORITY_UPDATE_REQUEST { + Some(HFrame::PriorityUpdateRequest { + element_id, + priority, + }) + } else { + Some(HFrame::PriorityUpdatePush { + element_id, + priority, + }) + } + } + _ => None, + }) + } else { + Ok(None) + } + } + + fn is_known_type(frame_type: u64) -> bool { + matches!( + frame_type, + H3_FRAME_TYPE_DATA + | H3_FRAME_TYPE_HEADERS + | H3_FRAME_TYPE_CANCEL_PUSH + | H3_FRAME_TYPE_SETTINGS + | H3_FRAME_TYPE_PUSH_PROMISE + | H3_FRAME_TYPE_GOAWAY + | H3_FRAME_TYPE_MAX_PUSH_ID + | H3_FRAME_TYPE_PRIORITY_UPDATE_REQUEST + | H3_FRAME_TYPE_PRIORITY_UPDATE_PUSH + ) + } +} diff --git a/third_party/rust/neqo-http3/src/frames/mod.rs b/third_party/rust/neqo-http3/src/frames/mod.rs new file mode 100644 index 0000000000..8b615fad01 --- /dev/null +++ b/third_party/rust/neqo-http3/src/frames/mod.rs @@ -0,0 +1,21 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +pub(crate) mod hframe; +pub(crate) mod reader; +pub(crate) mod wtframe; + +#[allow(unused_imports)] +pub(crate) use hframe::{ + HFrame, H3_FRAME_TYPE_HEADERS, H3_FRAME_TYPE_SETTINGS, H3_RESERVED_FRAME_TYPES, +}; +pub(crate) use reader::{ + FrameReader, StreamReaderConnectionWrapper, StreamReaderRecvStreamWrapper, +}; +pub(crate) use wtframe::WebTransportFrame; + +#[cfg(test)] +mod tests; diff --git a/third_party/rust/neqo-http3/src/frames/reader.rs b/third_party/rust/neqo-http3/src/frames/reader.rs new file mode 100644 index 0000000000..5017c666a4 --- /dev/null +++ b/third_party/rust/neqo-http3/src/frames/reader.rs @@ -0,0 +1,280 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +#![allow(clippy::module_name_repetitions)] + +use std::{convert::TryFrom, fmt::Debug}; + +use neqo_common::{ + hex_with_len, qtrace, Decoder, IncrementalDecoderBuffer, IncrementalDecoderIgnore, + IncrementalDecoderUint, +}; +use neqo_transport::{Connection, StreamId}; + +use crate::{Error, RecvStream, Res}; + +const MAX_READ_SIZE: usize = 4096; + +pub(crate) trait FrameDecoder<T> { + fn is_known_type(frame_type: u64) -> bool; + /// # Errors + /// + /// Returns `HttpFrameUnexpected` if frames is not alowed, i.e. is a `H3_RESERVED_FRAME_TYPES`. + fn frame_type_allowed(_frame_type: u64) -> Res<()> { + Ok(()) + } + + /// # Errors + /// + /// If a frame cannot be properly decoded. + fn decode(frame_type: u64, frame_len: u64, data: Option<&[u8]>) -> Res<Option<T>>; +} + +pub(crate) trait StreamReader { + /// # Errors + /// + /// An error may happen while reading a stream, e.g. early close, protocol error, etc. + /// Return an error if the stream was closed on the transport layer, but that information is not + /// yet consumed on the http/3 layer. + fn read_data(&mut self, buf: &mut [u8]) -> Res<(usize, bool)>; +} + +pub(crate) struct StreamReaderConnectionWrapper<'a> { + conn: &'a mut Connection, + stream_id: StreamId, +} + +impl<'a> StreamReaderConnectionWrapper<'a> { + pub fn new(conn: &'a mut Connection, stream_id: StreamId) -> Self { + Self { conn, stream_id } + } +} + +impl<'a> StreamReader for StreamReaderConnectionWrapper<'a> { + /// # Errors + /// + /// An error may happen while reading a stream, e.g. early close, protocol error, etc. + fn read_data(&mut self, buf: &mut [u8]) -> Res<(usize, bool)> { + let res = self.conn.stream_recv(self.stream_id, buf)?; + Ok(res) + } +} + +pub(crate) struct StreamReaderRecvStreamWrapper<'a> { + recv_stream: &'a mut Box<dyn RecvStream>, + conn: &'a mut Connection, +} + +impl<'a> StreamReaderRecvStreamWrapper<'a> { + pub fn new(conn: &'a mut Connection, recv_stream: &'a mut Box<dyn RecvStream>) -> Self { + Self { recv_stream, conn } + } +} + +impl<'a> StreamReader for StreamReaderRecvStreamWrapper<'a> { + /// # Errors + /// + /// An error may happen while reading a stream, e.g. early close, protocol error, etc. + fn read_data(&mut self, buf: &mut [u8]) -> Res<(usize, bool)> { + self.recv_stream.read_data(self.conn, buf) + } +} + +#[derive(Clone, Debug)] +enum FrameReaderState { + GetType { decoder: IncrementalDecoderUint }, + GetLength { decoder: IncrementalDecoderUint }, + GetData { decoder: IncrementalDecoderBuffer }, + UnknownFrameDischargeData { decoder: IncrementalDecoderIgnore }, +} + +#[allow(clippy::module_name_repetitions)] +#[derive(Debug)] +pub(crate) struct FrameReader { + state: FrameReaderState, + frame_type: u64, + frame_len: u64, +} + +impl Default for FrameReader { + fn default() -> Self { + Self::new() + } +} + +impl FrameReader { + #[must_use] + pub fn new() -> Self { + Self { + state: FrameReaderState::GetType { + decoder: IncrementalDecoderUint::default(), + }, + frame_type: 0, + frame_len: 0, + } + } + + #[must_use] + pub fn new_with_type(frame_type: u64) -> Self { + Self { + state: FrameReaderState::GetLength { + decoder: IncrementalDecoderUint::default(), + }, + frame_type, + frame_len: 0, + } + } + + fn reset(&mut self) { + self.state = FrameReaderState::GetType { + decoder: IncrementalDecoderUint::default(), + }; + } + + fn min_remaining(&self) -> usize { + match &self.state { + FrameReaderState::GetType { decoder } | FrameReaderState::GetLength { decoder } => { + decoder.min_remaining() + } + FrameReaderState::GetData { decoder } => decoder.min_remaining(), + FrameReaderState::UnknownFrameDischargeData { decoder } => decoder.min_remaining(), + } + } + + fn decoding_in_progress(&self) -> bool { + if let FrameReaderState::GetType { decoder } = &self.state { + decoder.decoding_in_progress() + } else { + true + } + } + + /// returns true if quic stream was closed. + /// + /// # Errors + /// + /// May return `HttpFrame` if a frame cannot be decoded. + /// and `TransportStreamDoesNotExist` if `stream_recv` fails. + pub fn receive<T: FrameDecoder<T>>( + &mut self, + stream_reader: &mut dyn StreamReader, + ) -> Res<(Option<T>, bool)> { + loop { + let to_read = std::cmp::min(self.min_remaining(), MAX_READ_SIZE); + let mut buf = vec![0; to_read]; + let (output, read, fin) = match stream_reader + .read_data(&mut buf) + .map_err(|e| Error::map_stream_recv_errors(&e))? + { + (0, f) => (None, false, f), + (amount, f) => { + qtrace!("FrameReader::receive: reading {} byte, fin={}", amount, f); + (self.consume::<T>(Decoder::from(&buf[..amount]))?, true, f) + } + }; + + if output.is_some() { + break Ok((output, fin)); + } + + if fin { + if self.decoding_in_progress() { + break Err(Error::HttpFrame); + } + break Ok((None, fin)); + } + + if !read { + // There was no new data, exit the loop. + break Ok((None, false)); + } + } + } + + /// # Errors + /// + /// May return `HttpFrame` if a frame cannot be decoded. + fn consume<T: FrameDecoder<T>>(&mut self, mut input: Decoder) -> Res<Option<T>> { + match &mut self.state { + FrameReaderState::GetType { decoder } => { + if let Some(v) = decoder.consume(&mut input) { + qtrace!("FrameReader::receive: read frame type {}", v); + self.frame_type_decoded::<T>(v)?; + } + } + FrameReaderState::GetLength { decoder } => { + if let Some(len) = decoder.consume(&mut input) { + qtrace!( + "FrameReader::receive: frame type {} length {}", + self.frame_type, + len + ); + return self.frame_length_decoded::<T>(len); + } + } + FrameReaderState::GetData { decoder } => { + if let Some(data) = decoder.consume(&mut input) { + qtrace!( + "received frame {}: {}", + self.frame_type, + hex_with_len(&data[..]) + ); + return self.frame_data_decoded::<T>(&data); + } + } + FrameReaderState::UnknownFrameDischargeData { decoder } => { + if decoder.consume(&mut input) { + self.reset(); + } + } + } + Ok(None) + } +} + +impl FrameReader { + fn frame_type_decoded<T: FrameDecoder<T>>(&mut self, frame_type: u64) -> Res<()> { + T::frame_type_allowed(frame_type)?; + self.frame_type = frame_type; + self.state = FrameReaderState::GetLength { + decoder: IncrementalDecoderUint::default(), + }; + Ok(()) + } + + fn frame_length_decoded<T: FrameDecoder<T>>(&mut self, len: u64) -> Res<Option<T>> { + self.frame_len = len; + if let Some(f) = T::decode( + self.frame_type, + self.frame_len, + if len > 0 { None } else { Some(&[]) }, + )? { + self.reset(); + return Ok(Some(f)); + } else if T::is_known_type(self.frame_type) { + self.state = FrameReaderState::GetData { + decoder: IncrementalDecoderBuffer::new( + usize::try_from(len).or(Err(Error::HttpFrame))?, + ), + }; + } else if self.frame_len == 0 { + self.reset(); + } else { + self.state = FrameReaderState::UnknownFrameDischargeData { + decoder: IncrementalDecoderIgnore::new( + usize::try_from(len).or(Err(Error::HttpFrame))?, + ), + }; + } + Ok(None) + } + + fn frame_data_decoded<T: FrameDecoder<T>>(&mut self, data: &[u8]) -> Res<Option<T>> { + let res = T::decode(self.frame_type, self.frame_len, Some(data))?; + self.reset(); + Ok(res) + } +} diff --git a/third_party/rust/neqo-http3/src/frames/tests/hframe.rs b/third_party/rust/neqo-http3/src/frames/tests/hframe.rs new file mode 100644 index 0000000000..3da7e7fc36 --- /dev/null +++ b/third_party/rust/neqo-http3/src/frames/tests/hframe.rs @@ -0,0 +1,116 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use neqo_common::{Decoder, Encoder}; +use neqo_transport::StreamId; +use test_fixture::fixture_init; + +use super::enc_dec_hframe; +use crate::{ + frames::HFrame, + settings::{HSetting, HSettingType, HSettings}, + Priority, +}; + +#[test] +fn test_data_frame() { + let f = HFrame::Data { len: 3 }; + enc_dec_hframe(&f, "0003010203", 3); +} + +#[test] +fn test_headers_frame() { + let f = HFrame::Headers { + header_block: vec![0x01, 0x02, 0x03], + }; + enc_dec_hframe(&f, "0103010203", 0); +} + +#[test] +fn test_cancel_push_frame4() { + let f = HFrame::CancelPush { push_id: 5 }; + enc_dec_hframe(&f, "030105", 0); +} + +#[test] +fn test_settings_frame4() { + let f = HFrame::Settings { + settings: HSettings::new(&[HSetting::new(HSettingType::MaxHeaderListSize, 4)]), + }; + enc_dec_hframe(&f, "04020604", 0); +} + +#[test] +fn test_push_promise_frame4() { + let f = HFrame::PushPromise { + push_id: 4, + header_block: vec![0x61, 0x62, 0x63, 0x64], + }; + enc_dec_hframe(&f, "05050461626364", 0); +} + +#[test] +fn test_goaway_frame4() { + let f = HFrame::Goaway { + stream_id: StreamId::new(5), + }; + enc_dec_hframe(&f, "070105", 0); +} + +#[test] +fn grease() { + fn make_grease() -> u64 { + let mut enc = Encoder::default(); + HFrame::Grease.encode(&mut enc); + let mut dec = Decoder::from(&enc); + let ft = dec.decode_varint().unwrap(); + assert_eq!((ft - 0x21) % 0x1f, 0); + let body = dec.decode_vvec().unwrap(); + assert!(body.len() <= 7); + ft + } + + fixture_init(); + let t1 = make_grease(); + let t2 = make_grease(); + assert_ne!(t1, t2); +} + +#[test] +fn test_priority_update_request_default() { + let f = HFrame::PriorityUpdateRequest { + element_id: 6, + priority: Priority::default(), + }; + enc_dec_hframe(&f, "800f07000106", 0); +} + +#[test] +fn test_priority_update_request_incremental_default() { + let f = HFrame::PriorityUpdateRequest { + element_id: 7, + priority: Priority::new(6, false), + }; + enc_dec_hframe(&f, "800f07000407753d36", 0); // "u=6" +} + +#[test] +fn test_priority_update_request_urgency_default() { + let f = HFrame::PriorityUpdateRequest { + element_id: 8, + priority: Priority::new(3, true), + }; + enc_dec_hframe(&f, "800f0700020869", 0); // "i" +} + +#[test] +fn test_priority_update_push_default() { + let f = HFrame::PriorityUpdatePush { + element_id: 10, + priority: Priority::default(), + }; + enc_dec_hframe(&f, "800f0701010a", 0); +} diff --git a/third_party/rust/neqo-http3/src/frames/tests/mod.rs b/third_party/rust/neqo-http3/src/frames/tests/mod.rs new file mode 100644 index 0000000000..33eea5497a --- /dev/null +++ b/third_party/rust/neqo-http3/src/frames/tests/mod.rs @@ -0,0 +1,84 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::mem; + +use neqo_common::Encoder; +use neqo_crypto::AuthenticationStatus; +use neqo_transport::StreamType; +use test_fixture::{default_client, default_server, now}; + +use crate::frames::{ + reader::FrameDecoder, FrameReader, HFrame, StreamReaderConnectionWrapper, WebTransportFrame, +}; + +#[allow(clippy::many_single_char_names)] +pub(crate) fn enc_dec<T: FrameDecoder<T>>(d: &Encoder, st: &str, remaining: usize) -> T { + // For data, headers and push_promise we do not read all bytes from the buffer + let d2 = Encoder::from_hex(st); + assert_eq!(d.as_ref(), &d2.as_ref()[..d.as_ref().len()]); + + let mut conn_c = default_client(); + let mut conn_s = default_server(); + let out = conn_c.process(None, now()); + let out = conn_s.process(out.as_dgram_ref(), now()); + let out = conn_c.process(out.as_dgram_ref(), now()); + mem::drop(conn_s.process(out.as_dgram_ref(), now())); + conn_c.authenticated(AuthenticationStatus::Ok, now()); + let out = conn_c.process(None, now()); + mem::drop(conn_s.process(out.as_dgram_ref(), now())); + + // create a stream + let stream_id = conn_s.stream_create(StreamType::BiDi).unwrap(); + + let mut fr: FrameReader = FrameReader::new(); + + // conver string into u8 vector + let buf = Encoder::from_hex(st); + conn_s.stream_send(stream_id, buf.as_ref()).unwrap(); + let out = conn_s.process(None, now()); + mem::drop(conn_c.process(out.as_dgram_ref(), now())); + + let (frame, fin) = fr + .receive::<T>(&mut StreamReaderConnectionWrapper::new( + &mut conn_c, + stream_id, + )) + .unwrap(); + assert!(!fin); + assert!(frame.is_some()); + + // Check remaining data. + let mut buf = [0_u8; 100]; + let (amount, _) = conn_c.stream_recv(stream_id, &mut buf).unwrap(); + assert_eq!(amount, remaining); + + frame.unwrap() +} + +pub fn enc_dec_hframe(f: &HFrame, st: &str, remaining: usize) { + let mut d = Encoder::default(); + + f.encode(&mut d); + + let frame = enc_dec::<HFrame>(&d, st, remaining); + + assert_eq!(*f, frame); +} + +pub fn enc_dec_wtframe(f: &WebTransportFrame, st: &str, remaining: usize) { + let mut d = Encoder::default(); + + f.encode(&mut d); + + let frame = enc_dec::<WebTransportFrame>(&d, st, remaining); + + assert_eq!(*f, frame); +} + +mod hframe; +mod reader; +mod wtframe; diff --git a/third_party/rust/neqo-http3/src/frames/tests/reader.rs b/third_party/rust/neqo-http3/src/frames/tests/reader.rs new file mode 100644 index 0000000000..fed1477ba4 --- /dev/null +++ b/third_party/rust/neqo-http3/src/frames/tests/reader.rs @@ -0,0 +1,518 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::{fmt::Debug, mem}; + +use neqo_common::Encoder; +use neqo_transport::{Connection, StreamId, StreamType}; +use test_fixture::{connect, now}; + +use crate::{ + frames::{ + reader::FrameDecoder, FrameReader, HFrame, StreamReaderConnectionWrapper, WebTransportFrame, + }, + settings::{HSetting, HSettingType, HSettings}, + Error, +}; + +struct FrameReaderTest { + pub fr: FrameReader, + pub conn_c: Connection, + pub conn_s: Connection, + pub stream_id: StreamId, +} + +impl FrameReaderTest { + pub fn new() -> Self { + let (conn_c, mut conn_s) = connect(); + let stream_id = conn_s.stream_create(StreamType::BiDi).unwrap(); + Self { + fr: FrameReader::new(), + conn_c, + conn_s, + stream_id, + } + } + + fn process<T: FrameDecoder<T>>(&mut self, v: &[u8]) -> Option<T> { + self.conn_s.stream_send(self.stream_id, v).unwrap(); + let out = self.conn_s.process(None, now()); + mem::drop(self.conn_c.process(out.as_dgram_ref(), now())); + let (frame, fin) = self + .fr + .receive::<T>(&mut StreamReaderConnectionWrapper::new( + &mut self.conn_c, + self.stream_id, + )) + .unwrap(); + assert!(!fin); + frame + } +} + +// Test receiving byte by byte for a SETTINGS frame. +#[test] +fn test_frame_reading_with_stream_settings1() { + let mut fr = FrameReaderTest::new(); + + // Send and read settings frame 040406040804 + assert!(fr.process::<HFrame>(&[0x4]).is_none()); + assert!(fr.process::<HFrame>(&[0x4]).is_none()); + assert!(fr.process::<HFrame>(&[0x6]).is_none()); + assert!(fr.process::<HFrame>(&[0x4]).is_none()); + assert!(fr.process::<HFrame>(&[0x8]).is_none()); + let frame = fr.process(&[0x4]); + + assert!(frame.is_some()); + if let HFrame::Settings { settings } = frame.unwrap() { + assert!(settings.len() == 1); + assert!(settings[0] == HSetting::new(HSettingType::MaxHeaderListSize, 4)); + } else { + panic!("wrong frame type"); + } +} + +// Test receiving byte by byte for a SETTINGS frame with larger varints +#[test] +fn test_frame_reading_with_stream_settings2() { + let mut fr = FrameReaderTest::new(); + + // Read settings frame 400406064004084100 + for i in &[0x40, 0x04, 0x06, 0x06, 0x40, 0x04, 0x08, 0x41] { + assert!(fr.process::<HFrame>(&[*i]).is_none()); + } + let frame = fr.process(&[0x0]); + + assert!(frame.is_some()); + if let HFrame::Settings { settings } = frame.unwrap() { + assert!(settings.len() == 1); + assert!(settings[0] == HSetting::new(HSettingType::MaxHeaderListSize, 4)); + } else { + panic!("wrong frame type"); + } +} + +// Test receiving byte by byte for a PUSH_PROMISE frame. +#[test] +fn test_frame_reading_with_stream_push_promise() { + let mut fr = FrameReaderTest::new(); + + // Read push-promise frame 05054101010203 + for i in &[0x05, 0x05, 0x41, 0x01, 0x01, 0x02] { + assert!(fr.process::<HFrame>(&[*i]).is_none()); + } + let frame = fr.process(&[0x3]); + + assert!(frame.is_some()); + if let HFrame::PushPromise { + push_id, + header_block, + } = frame.unwrap() + { + assert_eq!(push_id, 257); + assert_eq!(header_block, &[0x1, 0x2, 0x3]); + } else { + panic!("wrong frame type"); + } +} + +// Test DATA +#[test] +fn test_frame_reading_with_stream_data() { + let mut fr = FrameReaderTest::new(); + + // Read data frame 0003010203 + let frame = fr.process(&[0x0, 0x3, 0x1, 0x2, 0x3]).unwrap(); + assert!(matches!(frame, HFrame::Data { len } if len == 3)); + + // payloead is still on the stream. + // assert that we have 3 bytes in the stream + let mut buf = [0_u8; 100]; + let (amount, _) = fr.conn_c.stream_recv(fr.stream_id, &mut buf).unwrap(); + assert_eq!(amount, 3); +} + +// Test an unknown frame +#[test] +fn test_unknown_frame() { + // Construct an unknown frame. + const UNKNOWN_FRAME_LEN: usize = 832; + + let mut fr = FrameReaderTest::new(); + + let mut enc = Encoder::with_capacity(UNKNOWN_FRAME_LEN + 4); + enc.encode_varint(1028_u64); // Arbitrary type. + enc.encode_varint(UNKNOWN_FRAME_LEN as u64); + let mut buf: Vec<_> = enc.into(); + buf.resize(UNKNOWN_FRAME_LEN + buf.len(), 0); + assert!(fr.process::<HFrame>(&buf).is_none()); + + // now receive a CANCEL_PUSH fram to see that frame reader is ok. + let frame = fr.process(&[0x03, 0x01, 0x05]); + assert!(frame.is_some()); + if let HFrame::CancelPush { push_id } = frame.unwrap() { + assert!(push_id == 5); + } else { + panic!("wrong frame type"); + } +} + +// Test receiving byte by byte for a WT_FRAME_CLOSE_SESSION frame. +#[test] +fn test_frame_reading_with_stream_wt_close_session() { + let mut fr = FrameReaderTest::new(); + + // Read CloseSession frame 6843090000000548656c6c6f + for i in &[ + 0x68, 0x43, 0x09, 0x00, 0x00, 0x00, 0x05, 0x48, 0x65, 0x6c, 0x6c, + ] { + assert!(fr.process::<WebTransportFrame>(&[*i]).is_none()); + } + let frame = fr.process::<WebTransportFrame>(&[0x6f]); + + assert!(frame.is_some()); + let WebTransportFrame::CloseSession { error, message } = frame.unwrap(); + assert_eq!(error, 5); + assert_eq!(message, "Hello".to_string()); +} + +// Test an unknown frame for WebTransportFrames. +#[test] +fn test_unknown_wt_frame() { + // Construct an unknown frame. + const UNKNOWN_FRAME_LEN: usize = 832; + + let mut fr = FrameReaderTest::new(); + + let mut enc = Encoder::with_capacity(UNKNOWN_FRAME_LEN + 4); + enc.encode_varint(1028_u64); // Arbitrary type. + enc.encode_varint(UNKNOWN_FRAME_LEN as u64); + let mut buf: Vec<_> = enc.into(); + buf.resize(UNKNOWN_FRAME_LEN + buf.len(), 0); + assert!(fr.process::<WebTransportFrame>(&buf).is_none()); + + // now receive a WT_FRAME_CLOSE_SESSION fram to see that frame reader is ok. + let frame = fr.process(&[ + 0x68, 0x43, 0x09, 0x00, 0x00, 0x00, 0x05, 0x48, 0x65, 0x6c, 0x6c, 0x6f, + ]); + assert!(frame.is_some()); + let WebTransportFrame::CloseSession { error, message } = frame.unwrap(); + assert_eq!(error, 5); + assert_eq!(message, "Hello".to_string()); +} + +enum FrameReadingTestSend { + OnlyData, + DataWithFin, + DataThenFin, +} + +enum FrameReadingTestExpect { + Error, + Incomplete, + FrameComplete, + FrameAndStreamComplete, + StreamDoneWithoutFrame, +} + +fn test_reading_frame<T: FrameDecoder<T> + PartialEq + Debug>( + buf: &[u8], + test_to_send: &FrameReadingTestSend, + expected_result: &FrameReadingTestExpect, +) { + let mut fr = FrameReaderTest::new(); + + fr.conn_s.stream_send(fr.stream_id, buf).unwrap(); + if let FrameReadingTestSend::DataWithFin = test_to_send { + fr.conn_s.stream_close_send(fr.stream_id).unwrap(); + } + + let out = fr.conn_s.process(None, now()); + mem::drop(fr.conn_c.process(out.as_dgram_ref(), now())); + + if let FrameReadingTestSend::DataThenFin = test_to_send { + fr.conn_s.stream_close_send(fr.stream_id).unwrap(); + let out = fr.conn_s.process(None, now()); + mem::drop(fr.conn_c.process(out.as_dgram_ref(), now())); + } + + let rv = fr.fr.receive::<T>(&mut StreamReaderConnectionWrapper::new( + &mut fr.conn_c, + fr.stream_id, + )); + + match expected_result { + FrameReadingTestExpect::Error => assert_eq!(Err(Error::HttpFrame), rv), + FrameReadingTestExpect::Incomplete => { + assert_eq!(Ok((None, false)), rv); + } + FrameReadingTestExpect::FrameComplete => { + let (f, fin) = rv.unwrap(); + assert!(!fin); + assert!(f.is_some()); + } + FrameReadingTestExpect::FrameAndStreamComplete => { + let (f, fin) = rv.unwrap(); + assert!(fin); + assert!(f.is_some()); + } + FrameReadingTestExpect::StreamDoneWithoutFrame => { + let (f, fin) = rv.unwrap(); + assert!(fin); + assert!(f.is_none()); + } + }; +} + +#[test] +fn test_complete_and_incomplete_unknown_frame() { + // Construct an unknown frame. + const UNKNOWN_FRAME_LEN: usize = 832; + let mut enc = Encoder::with_capacity(UNKNOWN_FRAME_LEN + 4); + enc.encode_varint(1028_u64); // Arbitrary type. + enc.encode_varint(UNKNOWN_FRAME_LEN as u64); + let mut buf: Vec<_> = enc.into(); + buf.resize(UNKNOWN_FRAME_LEN + buf.len(), 0); + + let len = std::cmp::min(buf.len() - 1, 10); + for i in 1..len { + test_reading_frame::<HFrame>( + &buf[..i], + &FrameReadingTestSend::OnlyData, + &FrameReadingTestExpect::Incomplete, + ); + test_reading_frame::<HFrame>( + &buf[..i], + &FrameReadingTestSend::DataWithFin, + &FrameReadingTestExpect::Error, + ); + test_reading_frame::<HFrame>( + &buf[..i], + &FrameReadingTestSend::DataThenFin, + &FrameReadingTestExpect::Error, + ); + } + test_reading_frame::<HFrame>( + &buf, + &FrameReadingTestSend::OnlyData, + &FrameReadingTestExpect::Incomplete, + ); + test_reading_frame::<HFrame>( + &buf, + &FrameReadingTestSend::DataWithFin, + &FrameReadingTestExpect::StreamDoneWithoutFrame, + ); + test_reading_frame::<HFrame>( + &buf, + &FrameReadingTestSend::DataThenFin, + &FrameReadingTestExpect::StreamDoneWithoutFrame, + ); +} + +// if we read more than done_state bytes FrameReader will be in done state. +fn test_complete_and_incomplete_frame<T: FrameDecoder<T> + PartialEq + Debug>( + buf: &[u8], + done_state: usize, +) { + use std::cmp::Ordering; + // Let's consume partial frames. It is enough to test partal frames + // up to 10 byte. 10 byte is greater than frame type and frame + // length and bit of data. + let len = std::cmp::min(buf.len() - 1, 10); + for i in 1..len { + test_reading_frame::<T>( + &buf[..i], + &FrameReadingTestSend::OnlyData, + if i >= done_state { + &FrameReadingTestExpect::FrameComplete + } else { + &FrameReadingTestExpect::Incomplete + }, + ); + test_reading_frame::<T>( + &buf[..i], + &FrameReadingTestSend::DataWithFin, + match i.cmp(&done_state) { + Ordering::Greater => &FrameReadingTestExpect::FrameComplete, + Ordering::Equal => &FrameReadingTestExpect::FrameAndStreamComplete, + Ordering::Less => &FrameReadingTestExpect::Error, + }, + ); + test_reading_frame::<T>( + &buf[..i], + &FrameReadingTestSend::DataThenFin, + match i.cmp(&done_state) { + Ordering::Greater => &FrameReadingTestExpect::FrameComplete, + Ordering::Equal => &FrameReadingTestExpect::FrameAndStreamComplete, + Ordering::Less => &FrameReadingTestExpect::Error, + }, + ); + } + test_reading_frame::<T>( + buf, + &FrameReadingTestSend::OnlyData, + &FrameReadingTestExpect::FrameComplete, + ); + test_reading_frame::<T>( + buf, + &FrameReadingTestSend::DataWithFin, + if buf.len() == done_state { + &FrameReadingTestExpect::FrameAndStreamComplete + } else { + &FrameReadingTestExpect::FrameComplete + }, + ); + test_reading_frame::<T>( + buf, + &FrameReadingTestSend::DataThenFin, + if buf.len() == done_state { + &FrameReadingTestExpect::FrameAndStreamComplete + } else { + &FrameReadingTestExpect::FrameComplete + }, + ); +} + +#[test] +fn test_complete_and_incomplete_frames() { + const FRAME_LEN: usize = 10; + const HEADER_BLOCK: &[u8] = &[0x01, 0x02, 0x03, 0x04]; + + // H3_FRAME_TYPE_DATA len=0 + let f = HFrame::Data { len: 0 }; + let mut enc = Encoder::with_capacity(2); + f.encode(&mut enc); + let buf: Vec<_> = enc.into(); + test_complete_and_incomplete_frame::<HFrame>(&buf, 2); + + // H3_FRAME_TYPE_DATA len=FRAME_LEN + let f = HFrame::Data { + len: FRAME_LEN as u64, + }; + let mut enc = Encoder::with_capacity(2); + f.encode(&mut enc); + let mut buf: Vec<_> = enc.into(); + buf.resize(FRAME_LEN + buf.len(), 0); + test_complete_and_incomplete_frame::<HFrame>(&buf, 2); + + // H3_FRAME_TYPE_HEADERS empty header block + let f = HFrame::Headers { + header_block: Vec::new(), + }; + let mut enc = Encoder::default(); + f.encode(&mut enc); + let buf: Vec<_> = enc.into(); + test_complete_and_incomplete_frame::<HFrame>(&buf, 2); + + // H3_FRAME_TYPE_HEADERS + let f = HFrame::Headers { + header_block: HEADER_BLOCK.to_vec(), + }; + let mut enc = Encoder::default(); + f.encode(&mut enc); + let buf: Vec<_> = enc.into(); + test_complete_and_incomplete_frame::<HFrame>(&buf, buf.len()); + + // H3_FRAME_TYPE_CANCEL_PUSH + let f = HFrame::CancelPush { push_id: 5 }; + let mut enc = Encoder::default(); + f.encode(&mut enc); + let buf: Vec<_> = enc.into(); + test_complete_and_incomplete_frame::<HFrame>(&buf, buf.len()); + + // H3_FRAME_TYPE_SETTINGS + let f = HFrame::Settings { + settings: HSettings::new(&[HSetting::new(HSettingType::MaxHeaderListSize, 4)]), + }; + let mut enc = Encoder::default(); + f.encode(&mut enc); + let buf: Vec<_> = enc.into(); + test_complete_and_incomplete_frame::<HFrame>(&buf, buf.len()); + + // H3_FRAME_TYPE_PUSH_PROMISE + let f = HFrame::PushPromise { + push_id: 4, + header_block: HEADER_BLOCK.to_vec(), + }; + let mut enc = Encoder::default(); + f.encode(&mut enc); + let buf: Vec<_> = enc.into(); + test_complete_and_incomplete_frame::<HFrame>(&buf, buf.len()); + + // H3_FRAME_TYPE_GOAWAY + let f = HFrame::Goaway { + stream_id: StreamId::new(5), + }; + let mut enc = Encoder::default(); + f.encode(&mut enc); + let buf: Vec<_> = enc.into(); + test_complete_and_incomplete_frame::<HFrame>(&buf, buf.len()); + + // H3_FRAME_TYPE_MAX_PUSH_ID + let f = HFrame::MaxPushId { push_id: 5 }; + let mut enc = Encoder::default(); + f.encode(&mut enc); + let buf: Vec<_> = enc.into(); + test_complete_and_incomplete_frame::<HFrame>(&buf, buf.len()); +} + +#[test] +fn test_complete_and_incomplete_wt_frames() { + // H3_FRAME_TYPE_MAX_PUSH_ID + let f = WebTransportFrame::CloseSession { + error: 5, + message: "Hello".to_string(), + }; + let mut enc = Encoder::default(); + f.encode(&mut enc); + let buf: Vec<_> = enc.into(); + test_complete_and_incomplete_frame::<WebTransportFrame>(&buf, buf.len()); +} + +// Test closing a stream before any frame is sent should not cause an error. +#[test] +fn test_frame_reading_when_stream_is_closed_before_sending_data() { + let mut fr = FrameReaderTest::new(); + + fr.conn_s.stream_send(fr.stream_id, &[0x00]).unwrap(); + let out = fr.conn_s.process(None, now()); + mem::drop(fr.conn_c.process(out.as_dgram_ref(), now())); + + assert_eq!(Ok(()), fr.conn_c.stream_close_send(fr.stream_id)); + let out = fr.conn_c.process(None, now()); + mem::drop(fr.conn_s.process(out.as_dgram_ref(), now())); + assert_eq!( + Ok((None, true)), + fr.fr + .receive::<HFrame>(&mut StreamReaderConnectionWrapper::new( + &mut fr.conn_s, + fr.stream_id + )) + ); +} + +// Test closing a stream before any frame is sent should not cause an error. +// This is the same as the previous just for WebTransportFrame. +#[test] +fn test_wt_frame_reading_when_stream_is_closed_before_sending_data() { + let mut fr = FrameReaderTest::new(); + + fr.conn_s.stream_send(fr.stream_id, &[0x00]).unwrap(); + let out = fr.conn_s.process(None, now()); + mem::drop(fr.conn_c.process(out.as_dgram_ref(), now())); + + assert_eq!(Ok(()), fr.conn_c.stream_close_send(fr.stream_id)); + let out = fr.conn_c.process(None, now()); + mem::drop(fr.conn_s.process(out.as_dgram_ref(), now())); + assert_eq!( + Ok((None, true)), + fr.fr + .receive::<WebTransportFrame>(&mut StreamReaderConnectionWrapper::new( + &mut fr.conn_s, + fr.stream_id + )) + ); +} diff --git a/third_party/rust/neqo-http3/src/frames/tests/wtframe.rs b/third_party/rust/neqo-http3/src/frames/tests/wtframe.rs new file mode 100644 index 0000000000..b6470a89cf --- /dev/null +++ b/third_party/rust/neqo-http3/src/frames/tests/wtframe.rs @@ -0,0 +1,17 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use super::enc_dec_wtframe; +use crate::frames::WebTransportFrame; + +#[test] +fn test_wt_close_session() { + let f = WebTransportFrame::CloseSession { + error: 5, + message: "Hello".to_string(), + }; + enc_dec_wtframe(&f, "6843090000000548656c6c6f", 0); +} diff --git a/third_party/rust/neqo-http3/src/frames/wtframe.rs b/third_party/rust/neqo-http3/src/frames/wtframe.rs new file mode 100644 index 0000000000..deb7a026a0 --- /dev/null +++ b/third_party/rust/neqo-http3/src/frames/wtframe.rs @@ -0,0 +1,62 @@ +// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or +// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license +// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your +// option. This file may not be copied, modified, or distributed +// except according to those terms. + +use std::convert::TryFrom; + +use neqo_common::{Decoder, Encoder}; + +use crate::{frames::reader::FrameDecoder, Error, Res}; + +pub(crate) type WebTransportFrameType = u64; + +const WT_FRAME_CLOSE_SESSION: WebTransportFrameType = 0x2843; +const WT_FRAME_CLOSE_MAX_MESSAGE_SIZE: u64 = 1024; + +#[derive(PartialEq, Eq, Debug)] +pub enum WebTransportFrame { + CloseSession { error: u32, message: String }, +} + +impl WebTransportFrame { + pub fn encode(&self, enc: &mut Encoder) { + enc.encode_varint(WT_FRAME_CLOSE_SESSION); + let WebTransportFrame::CloseSession { error, message } = &self; + enc.encode_varint(4 + message.len() as u64); + enc.encode_uint(4, *error); + enc.encode(message.as_bytes()); + } +} + +impl FrameDecoder<WebTransportFrame> for WebTransportFrame { + fn decode( + frame_type: u64, + frame_len: u64, + data: Option<&[u8]>, + ) -> Res<Option<WebTransportFrame>> { + if let Some(payload) = data { + let mut dec = Decoder::from(payload); + if frame_type == WT_FRAME_CLOSE_SESSION { + if frame_len > WT_FRAME_CLOSE_MAX_MESSAGE_SIZE + 4 { + return Err(Error::HttpMessageError); + } + let error = + u32::try_from(dec.decode_uint(4).ok_or(Error::HttpMessageError)?).unwrap(); + let Ok(message) = String::from_utf8(dec.decode_remainder().to_vec()) else { + return Err(Error::HttpMessageError); + }; + Ok(Some(WebTransportFrame::CloseSession { error, message })) + } else { + Ok(None) + } + } else { + Ok(None) + } + } + + fn is_known_type(frame_type: u64) -> bool { + frame_type == WT_FRAME_CLOSE_SESSION + } +} |