From 835678384a861e0241e4c0583be56ac4c3cec05f Mon Sep 17 00:00:00 2001 From: SC123667 <205642379+SC123667@users.noreply.github.com> Date: Tue, 11 Aug 2026 17:41:41 +0800 Subject: [PATCH] Add QPACK dynamic table support --- src/h3/connection.rs | 555 +++++++++++++- src/h3/qpack/dynamic_table.rs | 333 +++++++++ src/h3/qpack/prefix_int.rs | 23 +- src/h3/qpack/qpack.rs | 1294 ++++++++++++++++++++++++++++++--- src/h3/stream.rs | 13 + 5 files changed, 2069 insertions(+), 149 deletions(-) create mode 100644 src/h3/qpack/dynamic_table.rs diff --git a/src/h3/connection.rs b/src/h3/connection.rs index 7962b37b6..ec9f2bf06 100644 --- a/src/h3/connection.rs +++ b/src/h3/connection.rs @@ -15,7 +15,6 @@ use std::collections::hash_map; use std::collections::VecDeque; use std::convert::TryFrom; -use std::mem::MaybeUninit; use std::sync::Arc; use bytes::Bytes; @@ -85,6 +84,20 @@ pub struct Http3Connection { /// The streams for peer QPACK. peer_qpack_streams: QpackStreams, + /// Field sections waiting for encoder-stream insertions. + blocked_field_sections: StreamIdHashMap, + + /// Decoder instructions waiting for QPACK decoder-stream flow-control + /// credit. + pending_qpack_decoder_instructions: VecDeque, + + /// Encoder instructions waiting for QPACK encoder-stream flow-control + /// credit. + pending_qpack_encoder_instructions: VecDeque, + + /// HTTP/3 events made ready by QPACK encoder-stream processing. + pending_qpack_events: VecDeque<(u64, Http3Event)>, + /// The next stream ID to be used for a request(bididirectional) stream. next_request_stream_id: u64, /// The next stream ID to be used for a unidirectional stream. @@ -147,7 +160,9 @@ impl Http3Connection { }, qpack_encoder: qpack::QpackEncoder::new(), - qpack_decoder: qpack::QpackDecoder::new(), + qpack_decoder: qpack::QpackDecoder::with_max_capacity( + config.qpack_max_table_capacity.unwrap_or(0), + ), local_qpack_streams: QpackStreams { encoder_stream_id: None, @@ -159,6 +174,11 @@ impl Http3Connection { decoder_stream_id: None, }, + blocked_field_sections: Default::default(), + pending_qpack_decoder_instructions: VecDeque::new(), + pending_qpack_encoder_instructions: VecDeque::new(), + pending_qpack_events: VecDeque::new(), + next_request_stream_id: 0, next_uni_stream_id: initial_uni_stream_id, @@ -325,6 +345,7 @@ impl Http3Connection { let _ = conn.stream_shutdown(stream_id, crate::Shutdown::Write, 0); } + self.cancel_qpack_stream(conn, stream_id)?; self.stream_destroy(stream_id); Ok(()) } @@ -355,6 +376,23 @@ impl Http3Connection { /// Encode HTTP/3 header fields into a field section with QPACK. fn encode_header_fields(&mut self, headers: &[T]) -> Result { + self.encode_header_fields_inner(None, headers) + .map(|(field_section, _)| field_section) + } + + fn encode_header_fields_for_stream( + &mut self, + stream_id: u64, + headers: &[T], + ) -> Result<(Bytes, Vec)> { + self.encode_header_fields_inner(Some(stream_id), headers) + } + + fn encode_header_fields_inner( + &mut self, + stream_id: Option, + headers: &[T], + ) -> Result<(Bytes, Vec)> { // RFC9114: The default value of max_field_section_size is unlimited. let max_field_section_size = self .peer_settings @@ -374,15 +412,67 @@ impl Http3Connection { } let mut header_block = BytesMut::zeroed(headers_size); - match self.qpack_encoder.encode(headers, header_block.as_mut()) { - Ok(v) => { - header_block.truncate(v); - Ok(header_block.freeze()) + let encoded = match stream_id { + Some(stream_id) => { + self.qpack_encoder + .encode_for_stream(stream_id, headers, header_block.as_mut()) + } + None => self + .qpack_encoder + .encode(headers, header_block.as_mut()) + .map(|field_len| (field_len, Vec::new())), + }; + match encoded { + Ok((field_len, instructions)) => { + header_block.truncate(field_len); + Ok((header_block.freeze(), instructions)) } Err(_) => Err(Http3Error::InternalError), } } + fn queue_qpack_encoder_instructions( + &mut self, + conn: &mut Connection, + instructions: Vec, + ) -> Result<()> { + if !instructions.is_empty() { + self.pending_qpack_encoder_instructions + .push_back(Bytes::from(instructions)); + } + self.flush_qpack_encoder_instructions(conn) + } + + fn flush_qpack_encoder_instructions(&mut self, conn: &mut Connection) -> Result<()> { + let Some(stream_id) = self.local_qpack_streams.encoder_stream_id else { + return if self.pending_qpack_encoder_instructions.is_empty() { + Ok(()) + } else { + Err(Http3Error::InternalError) + }; + }; + + while let Some(instruction) = self.pending_qpack_encoder_instructions.pop_front() { + let capacity = conn.stream_capacity(stream_id)?; + if capacity == 0 { + self.pending_qpack_encoder_instructions + .push_front(instruction); + let _ = conn.stream_want_write(stream_id, true); + return Ok(()); + } + + let write_len = capacity.min(instruction.len()); + let written = conn.stream_write(stream_id, instruction.slice(..write_len), false)?; + if written < instruction.len() { + self.pending_qpack_encoder_instructions + .push_front(instruction.slice(written..)); + let _ = conn.stream_want_write(stream_id, true); + return Ok(()); + } + } + Ok(()) + } + /// Write HTTP/3 header block to quic stream buffer. fn send_header_block( &mut self, @@ -493,7 +583,18 @@ impl Http3Connection { stream.mark_priority_initialized(); } - let header_block = self.encode_header_fields(headers)?; + // A previous call may already have encoded this field section and + // queued its dynamic-table references before request-stream flow + // control blocked the HEADERS frame. Reuse that section so a retry + // cannot create an additional QPACK acknowledgment obligation for a + // field section that will never be sent. + if let Some((header_block, cached_fin)) = stream.take_header_block() { + return self.send_header_block(conn, stream_id, header_block, cached_fin || fin); + } + + let (header_block, encoder_instructions) = + self.encode_header_fields_for_stream(stream_id, headers)?; + self.queue_qpack_encoder_instructions(conn, encoder_instructions)?; self.send_header_block(conn, stream_id, header_block, fin) } @@ -969,10 +1070,8 @@ impl Http3Connection { } }; - // Try to open QPACK encoder/decoder streams, but ignore errors if it fails - // since we don't support QPACK dynamic table yet. - self.open_qpack_encoder_stream(conn).ok(); - self.open_qpack_decoder_stream(conn).ok(); + self.open_qpack_encoder_stream(conn)?; + self.open_qpack_decoder_stream(conn)?; Ok(()) } @@ -1113,17 +1212,75 @@ impl Http3Connection { stream_id: u64, field_section: Vec, ) -> Result<(u64, Http3Event)> { + let Some(headers) = self.decode_qpack_field_section( + conn, + stream_id, + field_section, + BlockedFieldSectionKind::Headers, + )? + else { + return Err(Http3Error::Done); + }; + + let headers_event = Http3Event::Headers { + headers, + fin: conn.stream_finished(stream_id), + }; + + Ok((stream_id, headers_event)) + } + + fn decode_qpack_field_section( + &mut self, + conn: &mut Connection, + stream_id: u64, + field_section: Vec, + kind: BlockedFieldSectionKind, + ) -> Result>> { // RFC9114: The default value of max_field_section_size is unlimited. let max_field_section_size = self .local_settings .max_field_section_size .unwrap_or(u64::MAX); - let headers = match self - .qpack_decoder - .decode(&field_section[..], max_field_section_size) - { - Ok(v) => v.0, + match self.qpack_decoder.decode_field_section( + stream_id, + &field_section, + max_field_section_size, + ) { + Ok(qpack::DecodeStatus::Decoded { + headers, + decoder_instructions, + .. + }) => { + self.queue_qpack_decoder_instructions(conn, decoder_instructions)?; + Ok(Some(headers)) + } + Ok(qpack::DecodeStatus::Blocked { + required_insert_count, + }) => { + if self.blocked_field_sections.contains_key(&stream_id) + || self.blocked_field_sections.len() as u64 + >= self.local_settings.qpack_blocked_streams.unwrap_or(0) + { + let e = Http3Error::QpackDecompressionFailed; + conn.close(true, e.to_wire(), b"too many QPACK blocked streams")?; + return Err(e); + } + + self.blocked_field_sections.insert( + stream_id, + BlockedFieldSection { + field_section, + required_insert_count, + kind, + }, + ); + if let Some(stream) = self.streams.get_mut(&stream_id) { + stream.set_qpack_blocked(true); + } + Ok(None) + } Err(e) => { error!( "{:?} stream {} qpack decode error: {:?}", @@ -1133,16 +1290,61 @@ impl Http3Connection { ); conn.close(true, e.to_wire(), b"qpack decompression failed")?; - return Err(e); + Err(e) } - }; + } + } - let headers_event = Http3Event::Headers { - headers, - fin: conn.stream_finished(stream_id), + fn queue_qpack_decoder_instructions( + &mut self, + conn: &mut Connection, + instructions: Vec, + ) -> Result<()> { + if !instructions.is_empty() { + self.pending_qpack_decoder_instructions + .push_back(Bytes::from(instructions)); + } + self.flush_qpack_decoder_instructions(conn) + } + + fn flush_qpack_decoder_instructions(&mut self, conn: &mut Connection) -> Result<()> { + let Some(stream_id) = self.local_qpack_streams.decoder_stream_id else { + return if self.pending_qpack_decoder_instructions.is_empty() { + Ok(()) + } else { + Err(Http3Error::InternalError) + }; }; - Ok((stream_id, headers_event)) + while let Some(instruction) = self.pending_qpack_decoder_instructions.pop_front() { + let capacity = conn.stream_capacity(stream_id)?; + if capacity == 0 { + self.pending_qpack_decoder_instructions + .push_front(instruction); + let _ = conn.stream_want_write(stream_id, true); + return Ok(()); + } + + let write_len = capacity.min(instruction.len()); + let written = conn.stream_write(stream_id, instruction.slice(..write_len), false)?; + if written < instruction.len() { + self.pending_qpack_decoder_instructions + .push_front(instruction.slice(written..)); + let _ = conn.stream_want_write(stream_id, true); + return Ok(()); + } + } + Ok(()) + } + + fn cancel_qpack_stream(&mut self, conn: &mut Connection, stream_id: u64) -> Result<()> { + self.blocked_field_sections.remove(&stream_id); + if let Some(stream) = self.streams.get_mut(&stream_id) { + stream.set_qpack_blocked(false); + } + + let instruction = self.qpack_decoder.stream_cancellation(stream_id)?; + self.queue_qpack_decoder_instructions(conn, instruction) } /// Receive an HTTP/3 DATA frame from the peer. @@ -1248,7 +1450,7 @@ impl Http3Connection { conn: &mut Connection, stream_id: u64, push_id: u64, - _field_section: Vec, + field_section: Vec, ) -> Result<(u64, Http3Event)> { // A client MUST NOT send a PUSH_PROMISE frame. A server MUST treat the receipt of // a PUSH_PROMISE frame as a connection error of type H3_FRAME_UNEXPECTED. @@ -1288,7 +1490,15 @@ impl Http3Connection { return Err(Http3Error::IdError); } - // Ignore the PUSH_PROMISE field_section temporarily. + // Decode the promised field section even though push delivery is not + // exposed yet, so QPACK acknowledgments and table references remain in + // sync with the peer. + let _ = self.decode_qpack_field_section( + conn, + stream_id, + field_section, + BlockedFieldSectionKind::PushPromise, + )?; Err(Http3Error::Done) } @@ -1444,6 +1654,13 @@ impl Http3Connection { raw, .. } => { + if let Err(e) = self + .qpack_encoder + .set_max_capacity(qpack_max_table_capacity.unwrap_or(0)) + { + conn.close(true, e.to_wire(), b"invalid QPACK table capacity")?; + return Err(e); + } self.peer_settings = Http3Settings { max_field_section_size, qpack_max_table_capacity, @@ -1520,15 +1737,108 @@ impl Http3Connection { conn: &mut Connection, stream_id: u64, ) -> Result<(u64, Http3Event)> { - let mut d: [u8; 4096] = unsafe { - #[allow(clippy::uninit_assumed_init, invalid_value)] - MaybeUninit::uninit().assume_init() - }; + let stream_type = self + .streams + .get(&stream_id) + .and_then(Http3Stream::stream_type) + .ok_or(Http3Error::InternalError)?; + let mut data = [0; 4096]; - // We don't support qpack dynamic table yet, so just read and discard all data. loop { - conn.stream_read(stream_id, &mut d)?; + let read = match conn.stream_read(stream_id, &mut data) { + Ok((0, _)) => break, + Ok((read, _)) => read, + Err(crate::Error::Done) => break, + Err(e) => return Err(e.into()), + }; + + let result = match stream_type { + Http3StreamType::QpackEncoder => { + match self + .qpack_decoder + .process_encoder_instructions(&data[..read]) + { + Ok(instructions) => { + self.queue_qpack_decoder_instructions(conn, instructions)?; + self.unblock_qpack_streams(conn) + } + Err(e) => Err(e), + } + } + Http3StreamType::QpackDecoder => self + .qpack_encoder + .process_decoder_instructions(&data[..read]), + _ => unreachable!(), + }; + + if let Err(e) = result { + conn.close(true, e.to_wire(), b"invalid QPACK instruction")?; + return Err(e); + } + } + + self.flush_qpack_encoder_instructions(conn)?; + self.flush_qpack_decoder_instructions(conn)?; + if let Some(event) = self.pending_qpack_events.pop_front() { + return Ok(event); } + Err(Http3Error::Done) + } + + fn unblock_qpack_streams(&mut self, conn: &mut Connection) -> Result<()> { + let insert_count = self.qpack_decoder.insert_count(); + let ready: Vec = self + .blocked_field_sections + .iter() + .filter_map(|(&stream_id, blocked)| { + (blocked.required_insert_count <= insert_count).then_some(stream_id) + }) + .collect(); + + for stream_id in ready { + let blocked = self.blocked_field_sections.remove(&stream_id).unwrap(); + let max_field_section_size = self + .local_settings + .max_field_section_size + .unwrap_or(u64::MAX); + let status = self.qpack_decoder.decode_field_section( + stream_id, + &blocked.field_section, + max_field_section_size, + )?; + let qpack::DecodeStatus::Decoded { + headers, + decoder_instructions, + .. + } = status + else { + return Err(Http3Error::QpackDecompressionFailed); + }; + self.queue_qpack_decoder_instructions(conn, decoder_instructions)?; + + if let Some(stream) = self.streams.get_mut(&stream_id) { + stream.set_qpack_blocked(false); + } + + if matches!(blocked.kind, BlockedFieldSectionKind::Headers) { + let fin = conn.stream_finished(stream_id); + self.pending_qpack_events + .push_back((stream_id, Http3Event::Headers { headers, fin })); + if fin { + let mut completed = false; + if let Some(stream) = self.streams.get_mut(&stream_id) { + stream.mark_read_finished(); + completed = stream.write_finished(); + } + self.pending_qpack_events + .push_back((stream_id, Http3Event::Finished)); + if completed { + self.stream_destroy(stream_id); + } + } + } + } + Ok(()) } /// Process readable HTTP/3 push stream. @@ -1545,6 +1855,9 @@ impl Http3Connection { // borrowing `self` for the entire duration of the loop, because we'll need to borrow it // again in inner block. while let Some(stream) = self.streams.get_mut(&stream_id) { + if stream.qpack_blocked() { + break; + } match stream.state() { Http3StreamState::PushId => { stream.parse_push_id(conn)?; @@ -1716,8 +2029,6 @@ impl Http3Connection { return self.process_readable_push_stream(conn, stream_id, polling); } - // Actually, Encoder and Decoder have different parsing instruction formats, - // but since we don't support dynamic table, we handle them here. Http3StreamType::QpackEncoder | Http3StreamType::QpackDecoder => { return self.process_readable_qpack_stream(conn, stream_id); } @@ -1747,6 +2058,9 @@ impl Http3Connection { // borrowing `self` for the entire duration of the loop, because we'll need to borrow it // again in inner block. while let Some(stream) = self.streams.get_mut(&stream_id) { + if stream.qpack_blocked() { + break; + } match stream.state() { Http3StreamState::FrameType => { stream.parse_frame_type(conn)?; @@ -1921,12 +2235,17 @@ impl Http3Connection { Err(Http3Error::Done) => None, // If the stream was reset, return a Reset event early, to avoid return a Finished event later. Err(Http3Error::TransportError(crate::Error::StreamReset(e))) => { - return Ok((stream_id, Http3Event::Reset(e))) + self.cancel_qpack_stream(conn, stream_id)?; + return Ok((stream_id, Http3Event::Reset(e))); } Err(e) => return Err(e), }; - if conn.stream_finished(stream_id) { + let qpack_blocked = self + .streams + .get(&stream_id) + .is_some_and(Http3Stream::qpack_blocked); + if conn.stream_finished(stream_id) && !qpack_blocked { trace!("{:?} stream {} finished", conn.trace_id(), stream_id); self.process_finished_stream(stream_id); @@ -1963,6 +2282,13 @@ impl Http3Connection { return Err(Http3Error::Done); } + self.flush_qpack_encoder_instructions(conn)?; + self.flush_qpack_decoder_instructions(conn)?; + + if let Some(event) = self.pending_qpack_events.pop_front() { + return Ok(event); + } + // Process finished HTTP/3 streams. if let Some(stream_id) = self.finished_streams.pop_front() { return Ok((stream_id, Http3Event::Finished)); @@ -1976,6 +2302,10 @@ impl Http3Connection { Err(e) => return Err(e), } + if let Some(event) = self.pending_qpack_events.pop_front() { + return Ok(event); + } + // Process all readable HTTP/3 streams. match self.process_readable_streams(conn) { Ok(ev) => return Ok(ev), @@ -2066,6 +2396,18 @@ struct QpackStreams { pub decoder_stream_id: Option, } +#[derive(Clone, Copy)] +enum BlockedFieldSectionKind { + Headers, + PushPromise, +} + +struct BlockedFieldSection { + field_section: Vec, + required_insert_count: u64, + kind: BlockedFieldSectionKind, +} + /// An extensible HTTP/3 Priority Parameters #[derive(Debug, PartialEq, Eq)] #[repr(C)] @@ -5645,6 +5987,151 @@ mod tests { assert_eq!(s.server.peer_settings.connect_protocol_enabled, None); } + #[test] + fn dynamic_table_round_trip_over_qpack_streams() { + let mut client_config = Session::new_test_config(false).unwrap(); + let mut server_config = Session::new_test_config(true).unwrap(); + let mut h3_config = Http3Config::new().unwrap(); + h3_config.set_qpack_max_table_capacity(512); + h3_config.set_qpack_blocked_streams(1); + + let mut s = + Session::new_with_test_config(&mut client_config, &mut server_config, &h3_config) + .unwrap(); + let headers = Session::default_request_headers(); + + let first_stream = s.send_request_with_custom_headers(&headers, true).unwrap(); + assert_eq!( + s.server_poll(), + Ok(( + first_stream, + Http3Event::Headers { + headers: headers.clone(), + fin: true, + }, + )) + ); + assert_eq!(s.server_poll(), Ok((first_stream, Http3Event::Finished))); + assert!(s.server.qpack_decoder.insert_count() > 0); + + // Deliver Insert Count Increment feedback so the next field section + // can safely reference the entries inserted by the first request. + s.move_forward().unwrap(); + assert_eq!(s.client_poll(), Err(Http3Error::Done)); + + let second_stream = s.send_request_with_custom_headers(&headers, true).unwrap(); + assert_eq!( + s.server_poll(), + Ok((second_stream, Http3Event::Headers { headers, fin: true },)) + ); + assert_eq!(s.server_poll(), Ok((second_stream, Http3Event::Finished))); + + // Deliver and process the Section Acknowledgment produced for the + // dynamically referenced second field section. + s.move_forward().unwrap(); + assert_eq!(s.client_poll(), Err(Http3Error::Done)); + } + + #[test] + fn qpack_encoder_instruction_resumes_after_flow_control() { + let mut client_config = Session::new_test_config(false).unwrap(); + let mut server_config = Session::new_test_config(true).unwrap(); + server_config.set_initial_max_stream_data_uni(24); + + let mut h3_config = Http3Config::new().unwrap(); + h3_config.set_qpack_max_table_capacity(512); + h3_config.set_qpack_blocked_streams(1); + let mut s = + Session::new_with_test_config(&mut client_config, &mut server_config, &h3_config) + .unwrap(); + let headers = vec![Header::new( + b"x-large-field", + b"abcdefghijklmnopqrstuvwxyz-abcdefghijklmnopqrstuvwxyz-abcdefghijklmnopqrstuvwxyz", + )]; + + let stream_id = s.send_request_with_custom_headers(&headers, true).unwrap(); + assert!(!s.client.pending_qpack_encoder_instructions.is_empty()); + assert_eq!( + s.server_poll(), + Ok((stream_id, Http3Event::Headers { headers, fin: true },)) + ); + assert_eq!(s.server_poll(), Ok((stream_id, Http3Event::Finished))); + + // Reading each partial encoder-stream chunk returns flow-control + // credit. Polling the sender flushes the queued remainder. + for _ in 0..8 { + s.move_forward().unwrap(); + let _ = s.client_poll(); + s.move_forward().unwrap(); + let _ = s.server_poll(); + if s.client.pending_qpack_encoder_instructions.is_empty() + && s.server.qpack_decoder.insert_count() == 1 + { + break; + } + } + assert!(s.client.pending_qpack_encoder_instructions.is_empty()); + assert_eq!(s.server.qpack_decoder.insert_count(), 1); + } + + #[test] + fn blocked_field_section_resumes_after_encoder_instructions() { + let mut client_config = Session::new_test_config(false).unwrap(); + let mut server_config = Session::new_test_config(true).unwrap(); + let mut h3_config = Http3Config::new().unwrap(); + h3_config.set_qpack_max_table_capacity(220); + h3_config.set_qpack_blocked_streams(1); + let mut s = + Session::new_with_test_config(&mut client_config, &mut server_config, &h3_config) + .unwrap(); + + // RFC 9204 Appendix B.2 field section references two entries that + // have not arrived on the encoder stream yet. + let stream_id = s.client.stream_new(&mut s.pair.client).unwrap(); + let field_section = hex::decode("03811011").unwrap(); + let mut frame_bytes = BytesMut::zeroed(32); + let frame_len = frame::Http3Frame::Headers { field_section } + .encode(frame_bytes.as_mut()) + .unwrap(); + frame_bytes.truncate(frame_len); + s.pair + .client + .stream_write(stream_id, frame_bytes.freeze(), true) + .unwrap(); + s.move_forward().unwrap(); + + assert_eq!(s.server_poll(), Err(Http3Error::Done)); + assert!(s.server.blocked_field_sections.contains_key(&stream_id)); + assert!(s.server.streams.get(&stream_id).unwrap().qpack_blocked()); + + let encoder_instructions = + hex::decode("3fbd01c00f7777772e6578616d706c652e636f6dc10c2f73616d706c652f70617468") + .unwrap(); + let encoder_stream_id = s.client.local_qpack_streams.encoder_stream_id.unwrap(); + s.pair + .client + .stream_write(encoder_stream_id, Bytes::from(encoder_instructions), false) + .unwrap(); + s.move_forward().unwrap(); + + assert_eq!( + s.server_poll(), + Ok(( + stream_id, + Http3Event::Headers { + headers: vec![ + Header::new(b":authority", b"www.example.com"), + Header::new(b":path", b"/sample/path"), + ], + fin: true, + }, + )) + ); + assert_eq!(s.server_poll(), Ok((stream_id, Http3Event::Finished))); + assert_eq!(s.server_poll(), Err(Http3Error::Done)); + assert!(!s.server.blocked_field_sections.contains_key(&stream_id)); + } + // Client try to open multiple control streams. #[test] fn client_open_multiple_control_streams() { diff --git a/src/h3/qpack/dynamic_table.rs b/src/h3/qpack/dynamic_table.rs new file mode 100644 index 000000000..841aa3762 --- /dev/null +++ b/src/h3/qpack/dynamic_table.rs @@ -0,0 +1,333 @@ +// Copyright (c) 2026 The TQUIC Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::VecDeque; + +/// The per-entry overhead defined by RFC 9204 Section 3.2.1. +const ENTRY_OVERHEAD: u64 = 32; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct DynamicEntry { + pub absolute_index: u64, + pub name: Vec, + pub value: Vec, + references: usize, +} + +impl DynamicEntry { + fn new(absolute_index: u64, name: Vec, value: Vec) -> Self { + Self { + absolute_index, + name, + value, + references: 0, + } + } + + pub fn size(&self) -> u64 { + entry_size(&self.name, &self.value) + } + + fn evictable(&self, known_received_count: u64) -> bool { + self.absolute_index < known_received_count && self.references == 0 + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum DynamicTableError { + CapacityTooLarge, + EntryTooLarge, + InvalidIndex, +} + +/// A FIFO QPACK dynamic table. +/// +/// Absolute indices remain stable for an entry's lifetime. Eviction only +/// removes entries from the front, so the entries retained by this structure +/// always form one contiguous range of absolute indices. +#[derive(Clone, Debug)] +pub struct DynamicTable { + entries: VecDeque, + size: u64, + capacity: u64, + max_capacity: u64, + insert_count: u64, +} + +impl DynamicTable { + pub fn new(max_capacity: u64) -> Self { + Self { + entries: VecDeque::new(), + size: 0, + capacity: 0, + max_capacity, + insert_count: 0, + } + } + + pub fn capacity(&self) -> u64 { + self.capacity + } + + pub fn max_capacity(&self) -> u64 { + self.max_capacity + } + + pub fn insert_count(&self) -> u64 { + self.insert_count + } + + pub fn set_max_capacity(&mut self, max_capacity: u64) { + self.max_capacity = max_capacity; + } + + /// Apply a capacity update on a decoder table. + pub fn set_capacity(&mut self, capacity: u64) -> Result<(), DynamicTableError> { + if capacity > self.max_capacity { + return Err(DynamicTableError::CapacityTooLarge); + } + + self.capacity = capacity; + while self.size > self.capacity { + self.evict_one()?; + } + + Ok(()) + } + + /// Apply a capacity update on an encoder table without evicting entries + /// that the decoder has not acknowledged or that are still referenced by + /// an outstanding field section. + pub fn set_encoder_capacity( + &mut self, + capacity: u64, + known_received_count: u64, + ) -> Result { + if capacity > self.max_capacity { + return Err(DynamicTableError::CapacityTooLarge); + } + + let mut projected_size = self.size; + let mut evict_count = 0; + for entry in &self.entries { + if projected_size <= capacity { + break; + } + if !entry.evictable(known_received_count) { + return Ok(false); + } + projected_size -= entry.size(); + evict_count += 1; + } + + for _ in 0..evict_count { + self.evict_one()?; + } + self.capacity = capacity; + Ok(true) + } + + /// Insert an entry on a decoder table. The encoder is responsible for + /// ensuring that any evicted entries are evictable. + pub fn insert(&mut self, name: Vec, value: Vec) -> Result { + let size = entry_size(&name, &value); + if size > self.capacity { + return Err(DynamicTableError::EntryTooLarge); + } + + while self.size > self.capacity - size { + self.evict_one()?; + } + + Ok(self.push(name, value)) + } + + /// Try to insert an entry on an encoder table. `None` means insertion is + /// currently prohibited because it would evict an unacknowledged or + /// referenced entry, or because the entry is larger than the capacity. + pub fn try_insert_encoder( + &mut self, + name: Vec, + value: Vec, + known_received_count: u64, + ) -> Result, DynamicTableError> { + let size = entry_size(&name, &value); + if size > self.capacity { + return Ok(None); + } + + let target_size = self.capacity - size; + let mut projected_size = self.size; + let mut evict_count = 0; + for entry in &self.entries { + if projected_size <= target_size { + break; + } + if !entry.evictable(known_received_count) { + return Ok(None); + } + projected_size -= entry.size(); + evict_count += 1; + } + + if projected_size > target_size { + return Ok(None); + } + + for _ in 0..evict_count { + self.evict_one()?; + } + + Ok(Some(self.push(name, value))) + } + + pub fn get_absolute(&self, absolute_index: u64) -> Option<&DynamicEntry> { + let first = self.first_index(); + let offset = usize::try_from(absolute_index.checked_sub(first)?).ok()?; + self.entries.get(offset) + } + + pub fn get_relative(&self, relative_index: u64) -> Option<&DynamicEntry> { + let relative = relative_index.checked_add(1)?; + let absolute_index = self.insert_count.checked_sub(relative)?; + self.get_absolute(absolute_index) + } + + pub fn find_exact(&self, name: &[u8], value: &[u8]) -> Option { + self.entries + .iter() + .rev() + .find(|entry| entry.name.eq_ignore_ascii_case(name) && entry.value == value) + .map(|entry| entry.absolute_index) + } + + pub fn find_name(&self, name: &[u8]) -> Option { + self.entries + .iter() + .rev() + .find(|entry| entry.name.eq_ignore_ascii_case(name)) + .map(|entry| entry.absolute_index) + } + + pub fn add_reference(&mut self, absolute_index: u64) -> Result<(), DynamicTableError> { + let entry = self + .get_absolute_mut(absolute_index) + .ok_or(DynamicTableError::InvalidIndex)?; + entry.references = entry + .references + .checked_add(1) + .ok_or(DynamicTableError::InvalidIndex)?; + Ok(()) + } + + pub fn release_reference(&mut self, absolute_index: u64) -> Result<(), DynamicTableError> { + let entry = self + .get_absolute_mut(absolute_index) + .ok_or(DynamicTableError::InvalidIndex)?; + entry.references = entry + .references + .checked_sub(1) + .ok_or(DynamicTableError::InvalidIndex)?; + Ok(()) + } + + fn first_index(&self) -> u64 { + self.insert_count - self.entries.len() as u64 + } + + fn get_absolute_mut(&mut self, absolute_index: u64) -> Option<&mut DynamicEntry> { + let first = self.first_index(); + let offset = usize::try_from(absolute_index.checked_sub(first)?).ok()?; + self.entries.get_mut(offset) + } + + fn push(&mut self, name: Vec, value: Vec) -> u64 { + let absolute_index = self.insert_count; + let entry = DynamicEntry::new(absolute_index, name, value); + self.size += entry.size(); + self.entries.push_back(entry); + self.insert_count += 1; + absolute_index + } + + fn evict_one(&mut self) -> Result<(), DynamicTableError> { + let entry = self + .entries + .pop_front() + .ok_or(DynamicTableError::InvalidIndex)?; + self.size -= entry.size(); + Ok(()) + } +} + +pub fn entry_size(name: &[u8], value: &[u8]) -> u64 { + name.len() as u64 + value.len() as u64 + ENTRY_OVERHEAD +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn insertion_and_fifo_eviction() { + let mut table = DynamicTable::new(84); + table.set_capacity(84).unwrap(); + assert_eq!(table.insert(b"a".to_vec(), b"1".to_vec()), Ok(0)); + assert_eq!(table.insert(b"b".to_vec(), b"2".to_vec()), Ok(1)); + assert_eq!(table.insert(b"c".to_vec(), b"3".to_vec()), Ok(2)); + assert!(table.get_absolute(0).is_none()); + assert_eq!(table.get_relative(0).unwrap().name, b"c"); + assert_eq!(table.get_relative(1).unwrap().name, b"b"); + } + + #[test] + fn encoder_preserves_referenced_entries() { + let mut table = DynamicTable::new(68); + assert!(table.set_encoder_capacity(68, 0).unwrap()); + let first = table + .try_insert_encoder(b"a".to_vec(), b"1".to_vec(), 0) + .unwrap() + .unwrap(); + table.add_reference(first).unwrap(); + assert_eq!( + table + .try_insert_encoder(b"b".to_vec(), b"2".to_vec(), 1) + .unwrap(), + Some(1) + ); + assert_eq!( + table + .try_insert_encoder(b"c".to_vec(), b"3".to_vec(), 2) + .unwrap(), + None + ); + table.release_reference(first).unwrap(); + assert_eq!( + table + .try_insert_encoder(b"c".to_vec(), b"3".to_vec(), 2) + .unwrap(), + Some(2) + ); + } + + #[test] + fn rejects_oversized_decoder_entry() { + let mut table = DynamicTable::new(32); + table.set_capacity(32).unwrap(); + assert_eq!( + table.insert(b"a".to_vec(), Vec::new()), + Err(DynamicTableError::EntryTooLarge) + ); + } +} diff --git a/src/h3/qpack/prefix_int.rs b/src/h3/qpack/prefix_int.rs index fc0933864..a283196ba 100644 --- a/src/h3/qpack/prefix_int.rs +++ b/src/h3/qpack/prefix_int.rs @@ -17,6 +17,8 @@ use crate::codec::Encoder; use crate::h3::Http3Error; use crate::h3::Result; +pub const MAX_QPACK_INT: u64 = (1 << 62) - 1; + /// Encode an integer using QPACK integer representation. /// /// An integer is represented in two parts: a prefix that fills the current @@ -71,17 +73,21 @@ pub fn decode_int(mut buf: &[u8], n: usize) -> Result<(u64, usize)> { let mut shift = 0; while !buf.is_empty() { let byte = buf.read_u8()?; + if shift >= 62 { + return Err(Http3Error::QpackDecompressionFailed); + } let inc = u64::from(byte & 0x7f) .checked_shl(shift) .ok_or(Http3Error::QpackDecompressionFailed)?; val = val .checked_add(inc) + .filter(|value| *value <= MAX_QPACK_INT) .ok_or(Http3Error::QpackDecompressionFailed)?; - shift += 7; if byte & 0x80 == 0 { return Ok((val, buf_len - buf.len())); } + shift += 7; } Err(Http3Error::QpackDecompressionFailed) @@ -131,4 +137,19 @@ mod test { ]; assert!(decode_int(&mut buf, 5).is_err()); } + + #[test] + fn decode_int_rejects_overlong_high_bits() { + let mut buf = vec![0b11111]; + buf.extend(std::iter::repeat_n(0b1000_0000, 9)); + buf.push(0b0000_0010); + assert!(decode_int(&buf, 5).is_err()); + } + + #[test] + fn prefix_int_supports_62_bit_values() { + let mut encoded = [0; 16]; + let len = encode_int(MAX_QPACK_INT, 0, 5, &mut encoded).unwrap(); + assert_eq!(decode_int(&encoded[..len], 5), Ok((MAX_QPACK_INT, len))); + } } diff --git a/src/h3/qpack/qpack.rs b/src/h3/qpack/qpack.rs index 99f0b382e..84af8d845 100644 --- a/src/h3/qpack/qpack.rs +++ b/src/h3/qpack/qpack.rs @@ -14,6 +14,9 @@ //! HTTP/3 header compression (QPACK). +use std::collections::HashMap; +use std::collections::VecDeque; + use log::trace; use crate::codec::Decoder; @@ -25,6 +28,17 @@ use crate::h3::Http3Error; use crate::h3::NameValue; use crate::h3::Result; +use self::dynamic_table::DynamicTable; + +// Encoder stream instructions (RFC 9204 Section 4.3). +const INSERT_WITH_NAME_REF: u8 = 0b1000_0000; +const INSERT_WITH_LITERAL_NAME: u8 = 0b0100_0000; +const SET_DYNAMIC_TABLE_CAPACITY: u8 = 0b0010_0000; + +// Decoder stream instructions (RFC 9204 Section 4.4). +const SECTION_ACKNOWLEDGMENT: u8 = 0b1000_0000; +const STREAM_CANCELLATION: u8 = 0b0100_0000; + /// An indexed field line representation starts with the '1' 1-bit pattern, /// followed by the 'T' bit, indicating whether the reference is into the /// static or dynamic table. @@ -130,108 +144,540 @@ impl Representation { } } +#[derive(Clone, Debug)] +struct OutstandingSection { + required_insert_count: u64, + references: Vec, +} + +#[derive(Clone, Debug)] +enum PlannedField { + StaticIndexed(u64), + DynamicIndexed(u64), + StaticName { + index: u64, + value: Vec, + never_index: bool, + }, + DynamicName { + absolute_index: u64, + value: Vec, + never_index: bool, + }, + Literal { + name: Vec, + value: Vec, + never_index: bool, + }, +} + /// A QPACK encoder. -#[derive(Default)] -pub struct QpackEncoder {} +/// +/// The encoder deliberately references only entries covered by the Known +/// Received Count. This conservative policy avoids creating blocked streams +/// while still compressing repeated field lines after the decoder confirms an +/// insertion. +#[derive(Clone, Debug)] +pub struct QpackEncoder { + table: DynamicTable, + known_received_count: u64, + pending_capacity: Option, + decoder_stream_buf: Vec, + outstanding: HashMap>, +} + +impl Default for QpackEncoder { + fn default() -> Self { + Self { + table: DynamicTable::new(0), + known_received_count: 0, + pending_capacity: None, + decoder_stream_buf: Vec::new(), + outstanding: HashMap::new(), + } + } +} impl QpackEncoder { pub fn new() -> QpackEncoder { QpackEncoder::default() } - /// Encode a list of headers into a QPACK field section. + /// Update the maximum capacity advertised by the peer. The matching Set + /// Dynamic Table Capacity instruction is emitted with the first insertion. + pub fn set_max_capacity(&mut self, capacity: u64) -> Result<()> { + self.table.set_max_capacity(capacity); + if !self + .table + .set_encoder_capacity(capacity, self.known_received_count) + .map_err(|_| Http3Error::QpackDecoderStreamError)? + { + return Err(Http3Error::QpackDecoderStreamError); + } + self.pending_capacity = (capacity > 0).then_some(capacity); + Ok(()) + } + + /// Encode a list of headers into a QPACK field section without associating + /// the section with an HTTP stream. This preserves the original unit-level + /// API and uses the static table when no peer capacity has been configured. pub fn encode(&mut self, headers: &[T], out: &mut [u8]) -> Result { - // Required Insert Count. - let mut off = encode_int(0, 0, 8, out)?; + self.encode_transactional(None, headers, out) + .map(|(field_len, _)| field_len) + } - // Base. - off += encode_int(0, 0, 7, &mut out[off..])?; + /// Encode a field section for an HTTP stream and return encoder-stream + /// instructions that must be written before the field section. + pub fn encode_for_stream( + &mut self, + stream_id: u64, + headers: &[T], + out: &mut [u8], + ) -> Result<(usize, Vec)> { + self.encode_transactional(Some(stream_id), headers, out) + } + + fn encode_transactional( + &mut self, + stream_id: Option, + headers: &[T], + out: &mut [u8], + ) -> Result<(usize, Vec)> { + let mut next = self.clone(); + let result = next.encode_inner(stream_id, headers, out)?; + *self = next; + Ok(result) + } + + fn encode_inner( + &mut self, + stream_id: Option, + headers: &[T], + out: &mut [u8], + ) -> Result<(usize, Vec)> { + let allow_dynamic = stream_id.is_some(); + let mut fields = Vec::with_capacity(headers.len()); + let mut references = Vec::new(); + let mut instructions = Vec::new(); for hdr in headers { - match encode_static(hdr) { - // Encode as statically indexed. - Some((idx, true)) => { - const STATIC: u8 = 0x40; - off += encode_int(idx, INDEXED | STATIC, 6, &mut out[off..])?; - trace!("QpackEncoder Indexed index={} static=true", idx); + let never_index = is_sensitive(hdr.name()); + let static_match = encode_static(hdr); + + let field = if !never_index && matches!(static_match, Some((_, true))) { + PlannedField::StaticIndexed(static_match.unwrap().0) + } else if !never_index && allow_dynamic { + match self.table.find_exact(hdr.name(), hdr.value()) { + Some(absolute_index) if absolute_index < self.known_received_count => { + self.add_reference(absolute_index, &mut references)?; + PlannedField::DynamicIndexed(absolute_index) + } + _ => self.plan_literal( + hdr, + static_match, + never_index, + allow_dynamic, + &mut references, + )?, } + } else { + self.plan_literal( + hdr, + static_match, + never_index, + allow_dynamic, + &mut references, + )? + }; + + if !never_index && allow_dynamic { + self.maybe_insert(hdr, &mut instructions)?; + } + fields.push(field); + } - // Encode value as literal with static name reference. - Some((idx, false)) => { + let required_insert_count = match references.iter().copied().max() { + Some(index) => index.checked_add(1).ok_or(Http3Error::InternalError)?, + None => 0, + }; + let base = required_insert_count; + let encoded_insert_count = + encode_required_insert_count(required_insert_count, self.table.max_capacity())?; + + let mut off = encode_int(encoded_insert_count, 0, 8, out)?; + off += encode_int(0, 0, 7, &mut out[off..])?; + + for field in fields { + match field { + PlannedField::StaticIndexed(index) => { + const STATIC: u8 = 0x40; + off += encode_int(index, INDEXED | STATIC, 6, &mut out[off..])?; + trace!("QpackEncoder Indexed index={} static=true", index); + } + PlannedField::DynamicIndexed(absolute_index) => { + let relative = absolute_index + .checked_add(1) + .ok_or(Http3Error::InternalError)?; + let index = base + .checked_sub(relative) + .ok_or(Http3Error::InternalError)?; + off += encode_int(index, INDEXED, 6, &mut out[off..])?; + trace!("QpackEncoder Indexed index={} static=false", index); + } + PlannedField::StaticName { + index, + value, + never_index, + } => { const STATIC: u8 = 0x10; - off += encode_int(idx, LITERAL_WITH_NAME_REF | STATIC, 4, &mut out[off..])?; - off += self.encode_str(hdr.value(), 7, &mut out[off..])?; - trace!( - "QpackDecoder Literal with name refer name_idx={} static=true", - idx - ); + let first = LITERAL_WITH_NAME_REF | STATIC | if never_index { 0x20 } else { 0 }; + off += encode_int(index, first, 4, &mut out[off..])?; + off += encode_string(&value, 0, 7, false, &mut out[off..])?; + } + PlannedField::DynamicName { + absolute_index, + value, + never_index, + } => { + let relative = absolute_index + .checked_add(1) + .ok_or(Http3Error::InternalError)?; + let index = base + .checked_sub(relative) + .ok_or(Http3Error::InternalError)?; + let first = LITERAL_WITH_NAME_REF | if never_index { 0x20 } else { 0 }; + off += encode_int(index, first, 4, &mut out[off..])?; + off += encode_string(&value, 0, 7, false, &mut out[off..])?; + } + PlannedField::Literal { + name, + value, + never_index, + } => { + let first = LITERAL | if never_index { 0x10 } else { 0 }; + off += encode_string(&name, first, 3, true, &mut out[off..])?; + off += encode_string(&value, 0, 7, false, &mut out[off..])?; } + } + } - // Encode as fully literal. - None => { - let len = huffman::encode_output_length(hdr.name(), true); - if len < hdr.name().len() { - off += encode_int(len as u64, LITERAL | 0x08, 3, &mut out[off..])?; - off += huffman::encode(hdr.name(), &mut out[off..], true)?; - } else { - off += encode_int(hdr.name().len() as u64, LITERAL, 3, &mut out[off..])?; - let mut buf = &mut out[off..]; - off += buf.write(&hdr.name().to_ascii_lowercase())?; - } - off += self.encode_str(hdr.value(), 7, &mut out[off..])?; - trace!( - "QpackDecoder Literal name={:?} value={:?}", - hdr.name(), - hdr.value() - ); + if let (Some(stream_id), false) = (stream_id, references.is_empty()) { + self.outstanding + .entry(stream_id) + .or_default() + .push_back(OutstandingSection { + required_insert_count, + references, + }); + } + + Ok((off, instructions)) + } + + fn plan_literal( + &mut self, + hdr: &T, + static_match: Option<(u64, bool)>, + never_index: bool, + allow_dynamic: bool, + references: &mut Vec, + ) -> Result { + if let Some((index, _)) = static_match { + return Ok(PlannedField::StaticName { + index, + value: hdr.value().to_vec(), + never_index, + }); + } + + if !never_index && allow_dynamic { + if let Some(absolute_index) = self.table.find_name(hdr.name()) { + if absolute_index < self.known_received_count { + self.add_reference(absolute_index, references)?; + return Ok(PlannedField::DynamicName { + absolute_index, + value: hdr.value().to_vec(), + never_index, + }); } - }; + } } - Ok(off) + Ok(PlannedField::Literal { + name: hdr.name().to_ascii_lowercase(), + value: hdr.value().to_vec(), + never_index, + }) + } + + fn add_reference(&mut self, absolute_index: u64, references: &mut Vec) -> Result<()> { + self.table + .add_reference(absolute_index) + .map_err(|_| Http3Error::InternalError)?; + references.push(absolute_index); + Ok(()) } - /// Encode a string in huffman encoding or literal. - fn encode_str(&mut self, v: &[u8], prefix: usize, buf: &mut [u8]) -> Result { - let len = huffman::encode_output_length(v, false); - if len < v.len() { - let mut off = encode_int(len as u64, 0x80, prefix, buf)?; - off += huffman::encode(v, &mut buf[off..], false)?; - Ok(off) + fn maybe_insert(&mut self, hdr: &T, instructions: &mut Vec) -> Result<()> { + if self.table.capacity() == 0 + || matches!(encode_static(hdr), Some((_, true))) + || self.table.find_exact(hdr.name(), hdr.value()).is_some() + { + return Ok(()); + } + + let static_name = encode_static(hdr).map(|(index, _)| index); + let dynamic_name = self.table.find_name(hdr.name()).map(|absolute_index| { + ( + absolute_index, + self.table.insert_count() - absolute_index - 1, + ) + }); + let name = hdr.name().to_ascii_lowercase(); + let value = hdr.value().to_vec(); + + let inserted = self + .table + .try_insert_encoder(name.clone(), value.clone(), self.known_received_count) + .map_err(|_| Http3Error::InternalError)?; + if inserted.is_none() { + return Ok(()); + } + + if let Some(capacity) = self.pending_capacity.take() { + append_int(instructions, capacity, SET_DYNAMIC_TABLE_CAPACITY, 5)?; + } + + if let Some(index) = static_name { + append_int(instructions, index, INSERT_WITH_NAME_REF | 0x40, 6)?; + append_string(instructions, &value, 0, 7, false)?; + } else if let Some((_, relative_index)) = dynamic_name { + append_int(instructions, relative_index, INSERT_WITH_NAME_REF, 6)?; + append_string(instructions, &value, 0, 7, false)?; } else { - let mut off = encode_int(v.len() as u64, 0, prefix, buf)?; - let mut buf = &mut buf[off..]; - off += buf.write(v)?; - Ok(off) + append_string(instructions, &name, INSERT_WITH_LITERAL_NAME, 5, true)?; + append_string(instructions, &value, 0, 7, false)?; + } + + Ok(()) + } + + /// Process decoder-stream acknowledgments and insert-count updates. + pub fn process_decoder_instructions(&mut self, data: &[u8]) -> Result<()> { + self.decoder_stream_buf.extend_from_slice(data); + + let mut consumed = 0; + while consumed < self.decoder_stream_buf.len() { + let buf = &self.decoder_stream_buf[consumed..]; + let first = buf[0]; + let (value, len) = if first & SECTION_ACKNOWLEDGMENT != 0 { + match decode_int_partial(buf, 7).map_err(|_| Http3Error::QpackDecoderStreamError)? { + Some(v) => v, + None => break, + } + } else { + match decode_int_partial(buf, 6).map_err(|_| Http3Error::QpackDecoderStreamError)? { + Some(v) => v, + None => break, + } + }; + + if first & SECTION_ACKNOWLEDGMENT != 0 { + self.acknowledge_section(value)?; + } else if first & 0xc0 == STREAM_CANCELLATION { + self.cancel_stream(value)?; + } else { + self.increment_known_received(value)?; + } + consumed += len; + } + + self.decoder_stream_buf.drain(..consumed); + Ok(()) + } + + fn acknowledge_section(&mut self, stream_id: u64) -> Result<()> { + let section = { + let sections = self + .outstanding + .get_mut(&stream_id) + .ok_or(Http3Error::QpackDecoderStreamError)?; + sections + .pop_front() + .ok_or(Http3Error::QpackDecoderStreamError)? + }; + if self + .outstanding + .get(&stream_id) + .is_some_and(VecDeque::is_empty) + { + self.outstanding.remove(&stream_id); + } + + self.known_received_count = self.known_received_count.max(section.required_insert_count); + self.release_references(section.references) + } + + fn cancel_stream(&mut self, stream_id: u64) -> Result<()> { + let Some(sections) = self.outstanding.remove(&stream_id) else { + return Ok(()); + }; + for section in sections { + self.release_references(section.references)?; + } + Ok(()) + } + + fn increment_known_received(&mut self, increment: u64) -> Result<()> { + let known_received_count = self + .known_received_count + .checked_add(increment) + .ok_or(Http3Error::QpackDecoderStreamError)?; + if increment == 0 || known_received_count > self.table.insert_count() { + return Err(Http3Error::QpackDecoderStreamError); + } + self.known_received_count = known_received_count; + Ok(()) + } + + fn release_references(&mut self, references: Vec) -> Result<()> { + for absolute_index in references { + self.table + .release_reference(absolute_index) + .map_err(|_| Http3Error::QpackDecoderStreamError)?; } + Ok(()) } } +#[derive(Debug, PartialEq, Eq)] +pub enum DecodeStatus { + Decoded { + headers: Vec
, + consumed: usize, + decoder_instructions: Vec, + }, + Blocked { + required_insert_count: u64, + }, +} + +#[derive(Debug)] +enum EncoderInstruction { + SetCapacity(u64), + InsertWithNameReference { + static_table: bool, + index: u64, + value: Vec, + }, + InsertWithLiteralName { + name: Vec, + value: Vec, + }, + Duplicate(u64), +} + /// A QPACK decoder. -#[derive(Default)] -pub struct QpackDecoder {} +#[derive(Debug)] +pub struct QpackDecoder { + table: DynamicTable, + encoder_stream_buf: Vec, + reported_insert_count: u64, +} + +impl Default for QpackDecoder { + fn default() -> Self { + Self::with_max_capacity(0) + } +} impl QpackDecoder { pub fn new() -> QpackDecoder { QpackDecoder::default() } + pub fn with_max_capacity(max_capacity: u64) -> QpackDecoder { + QpackDecoder { + table: DynamicTable::new(max_capacity), + encoder_stream_buf: Vec::new(), + reported_insert_count: 0, + } + } + + pub fn insert_count(&self) -> u64 { + self.table.insert_count() + } + + pub fn stream_cancellation(&self, stream_id: u64) -> Result> { + if self.table.max_capacity() == 0 { + return Ok(Vec::new()); + } + let mut instruction = Vec::new(); + append_int(&mut instruction, stream_id, STREAM_CANCELLATION, 6)?; + Ok(instruction) + } + /// Decode a QPACK header block into a list of headers. - pub fn decode(&mut self, mut buf: &[u8], max_size: u64) -> Result<(Vec
, usize)> { + pub fn decode(&mut self, buf: &[u8], max_size: u64) -> Result<(Vec
, usize)> { + match self.decode_field_section(0, buf, max_size)? { + DecodeStatus::Decoded { + headers, consumed, .. + } => Ok((headers, consumed)), + DecodeStatus::Blocked { .. } => Err(Http3Error::Done), + } + } + + pub fn decode_field_section( + &mut self, + stream_id: u64, + mut buf: &[u8], + max_size: u64, + ) -> Result { let buf_len = buf.len(); let mut out = Vec::new(); let mut left = max_size; + let mut largest_reference = None; - let (req_insert_count, off) = decode_int(buf, 8)?; + let (encoded_insert_count, off) = decode_int(buf, 8)?; buf = &buf[off..]; - let (base, off) = decode_int(buf, 7)?; + + if buf.is_empty() { + return Err(Http3Error::QpackDecompressionFailed); + } + let sign = buf[0] & 0x80 != 0; + let (delta_base, off) = decode_int(buf, 7)?; buf = &buf[off..]; + + let required_insert_count = decode_required_insert_count( + encoded_insert_count, + self.table.insert_count(), + self.table.max_capacity(), + )?; + let base = if sign { + let delta_base = delta_base + .checked_add(1) + .ok_or(Http3Error::QpackDecompressionFailed)?; + required_insert_count + .checked_sub(delta_base) + .ok_or(Http3Error::QpackDecompressionFailed)? + } else { + required_insert_count + .checked_add(delta_base) + .ok_or(Http3Error::QpackDecompressionFailed)? + }; + trace!( "QpackDecoder Header count={} base={}", - req_insert_count, + required_insert_count, base ); + if required_insert_count > self.table.insert_count() { + return Ok(DecodeStatus::Blocked { + required_insert_count, + }); + } + while !buf.is_empty() { let first = buf[0]; match Representation::from(first) { @@ -241,24 +687,36 @@ impl QpackDecoder { let (index, off) = decode_int(buf, 6)?; buf = &buf[off..]; - trace!("QpackDecoder Indexed index={} static={}", index, static_idx); - if !static_idx { - // TODO: implement dynamic table - return Err(Http3Error::QpackDecompressionFailed); - } + let (name, value) = if static_idx { + let (name, value) = decode_static(index)?; + (name.to_vec(), value.to_vec()) + } else { + let absolute_index = relative_absolute(base, index, required_insert_count)?; + update_largest_reference(&mut largest_reference, absolute_index); + let entry = self + .table + .get_absolute(absolute_index) + .ok_or(Http3Error::QpackDecompressionFailed)?; + (entry.name.clone(), entry.value.clone()) + }; - let (name, value) = decode_static(index)?; - left = left - .checked_sub((name.len() + value.len() + 32) as u64) - .ok_or(Http3Error::QpackDecompressionFailed)?; - out.push(Header(name.to_vec(), value.to_vec())); + charge_field(&mut left, &name, &value)?; + out.push(Header(name, value)); } Representation::IndexedWithPostBase => { - let (index, _) = decode_int(buf, 4)?; - trace!("QpackDecoder Indexed With Post Base index={}", index); - // TODO: implement dynamic table - return Err(Http3Error::QpackDecompressionFailed); + let (index, off) = decode_int(buf, 4)?; + buf = &buf[off..]; + let absolute_index = post_base_absolute(base, index, required_insert_count)?; + update_largest_reference(&mut largest_reference, absolute_index); + let entry = self + .table + .get_absolute(absolute_index) + .ok_or(Http3Error::QpackDecompressionFailed)?; + let name = entry.name.clone(); + let value = entry.value.clone(); + charge_field(&mut left, &name, &value)?; + out.push(Header(name, value)); } Representation::LiteralWithNameRef => { @@ -268,29 +726,39 @@ impl QpackDecoder { buf = &buf[off..]; let (value, off) = self.decode_str(buf)?; buf = &buf[off..]; - trace!( - "QpackDecoder Literal With Name refer name_idx={} static={} value={:?}", - name_idx, - static_idx, - value - ); - - if !static_idx { - // TODO: implement dynamic table - return Err(Http3Error::QpackDecompressionFailed); - } - let (name, _) = decode_static(name_idx)?; - left = left - .checked_sub((name.len() + value.len() + 32) as u64) - .ok_or(Http3Error::QpackDecompressionFailed)?; - out.push(Header(name.to_vec(), value)); + let name = if static_idx { + decode_static(name_idx)?.0.to_vec() + } else { + let absolute_index = + relative_absolute(base, name_idx, required_insert_count)?; + update_largest_reference(&mut largest_reference, absolute_index); + self.table + .get_absolute(absolute_index) + .ok_or(Http3Error::QpackDecompressionFailed)? + .name + .clone() + }; + + charge_field(&mut left, &name, &value)?; + out.push(Header(name, value)); } Representation::LiteralWithPostBase => { - trace!("QpackDecoder Literal With Post Base"); - // TODO: implement dynamic table - return Err(Http3Error::QpackDecompressionFailed); + let (name_idx, off) = decode_int(buf, 3)?; + buf = &buf[off..]; + let (value, off) = self.decode_str(buf)?; + buf = &buf[off..]; + let absolute_index = post_base_absolute(base, name_idx, required_insert_count)?; + update_largest_reference(&mut largest_reference, absolute_index); + let name = self + .table + .get_absolute(absolute_index) + .ok_or(Http3Error::QpackDecompressionFailed)? + .name + .clone(); + charge_field(&mut left, &name, &value)?; + out.push(Header(name, value)); } Representation::Literal => { @@ -304,49 +772,388 @@ impl QpackDecoder { } else { name.to_vec() }; - - let name = name.to_vec(); let (value, off) = self.decode_str(buf)?; buf = &buf[off..]; - trace!("QpackDecoder Literal name={:?} value={:?}", name, value); - left = left - .checked_sub((name.len() + value.len() + 32) as u64) - .ok_or(Http3Error::QpackDecompressionFailed)?; + charge_field(&mut left, &name, &value)?; out.push(Header(name, value)); } } } - Ok((out, buf_len - buf.len())) + let expected_required_insert_count = match largest_reference { + Some(index) => index + .checked_add(1) + .ok_or(Http3Error::QpackDecompressionFailed)?, + None => 0, + }; + if expected_required_insert_count != required_insert_count { + return Err(Http3Error::QpackDecompressionFailed); + } + + let mut decoder_instructions = Vec::new(); + if required_insert_count != 0 { + append_int( + &mut decoder_instructions, + stream_id, + SECTION_ACKNOWLEDGMENT, + 7, + )?; + } + + Ok(DecodeStatus::Decoded { + headers: out, + consumed: buf_len - buf.len(), + decoder_instructions, + }) } - /// Decode a string in huffman encoding or literal. - fn decode_str(&self, mut buf: &[u8]) -> Result<(Vec, usize)> { - if buf.is_empty() { - return Err(Http3Error::QpackDecompressionFailed); + /// Decode a string in Huffman encoding or literal form. + fn decode_str(&self, buf: &[u8]) -> Result<(Vec, usize)> { + decode_string_partial(buf, 7)?.ok_or(Http3Error::QpackDecompressionFailed) + } + + /// Process control instructions from the encoder and return decoder-stream + /// Insert Count Increment feedback. + pub fn process_encoder_instructions(&mut self, data: &[u8]) -> Result> { + self.encoder_stream_buf.extend_from_slice(data); + + let mut consumed = 0; + while consumed < self.encoder_stream_buf.len() { + let Some((instruction, len)) = + parse_encoder_instruction(&self.encoder_stream_buf[consumed..]) + .map_err(|_| Http3Error::QpackEncoderStreamError)? + else { + break; + }; + self.apply_encoder_instruction(instruction)?; + consumed += len; } + self.encoder_stream_buf.drain(..consumed); - let buf_len = buf.len(); - let huff = buf[0] & 0x80 == 0x80; - let (str_len, off) = decode_int(buf, 7)?; - buf = &buf[off..]; + let mut feedback = Vec::new(); + let increment = self.table.insert_count() - self.reported_insert_count; + if increment != 0 { + append_int(&mut feedback, increment, 0, 6) + .map_err(|_| Http3Error::QpackEncoderStreamError)?; + self.reported_insert_count = self.table.insert_count(); + } + Ok(feedback) + } + + fn apply_encoder_instruction(&mut self, instruction: EncoderInstruction) -> Result<()> { + match instruction { + EncoderInstruction::SetCapacity(capacity) => self + .table + .set_capacity(capacity) + .map_err(|_| Http3Error::QpackEncoderStreamError), + EncoderInstruction::InsertWithNameReference { + static_table, + index, + value, + } => { + let name = if static_table { + decode_static(index) + .map_err(|_| Http3Error::QpackEncoderStreamError)? + .0 + .to_vec() + } else { + self.table + .get_relative(index) + .ok_or(Http3Error::QpackEncoderStreamError)? + .name + .clone() + }; + self.table + .insert(name, value) + .map(|_| ()) + .map_err(|_| Http3Error::QpackEncoderStreamError) + } + EncoderInstruction::InsertWithLiteralName { name, value } => self + .table + .insert(name, value) + .map(|_| ()) + .map_err(|_| Http3Error::QpackEncoderStreamError), + EncoderInstruction::Duplicate(index) => { + let entry = self + .table + .get_relative(index) + .ok_or(Http3Error::QpackEncoderStreamError)?; + let name = entry.name.clone(); + let value = entry.value.clone(); + self.table + .insert(name, value) + .map(|_| ()) + .map_err(|_| Http3Error::QpackEncoderStreamError) + } + } + } + + /// Backwards-compatible control-stream entry point. + pub fn process(&mut self, buf: &mut [u8]) -> Result<()> { + self.process_encoder_instructions(buf).map(|_| ()) + } +} - let str_val = buf.read(str_len as usize)?; - let val = if huff { - huffman::decode(&str_val)? +fn is_sensitive(name: &[u8]) -> bool { + [ + b"authorization".as_slice(), + b"proxy-authorization".as_slice(), + b"cookie".as_slice(), + b"set-cookie".as_slice(), + ] + .iter() + .any(|sensitive| name.eq_ignore_ascii_case(sensitive)) +} + +fn encode_string( + value: &[u8], + first: u8, + prefix: usize, + lower_case: bool, + out: &mut [u8], +) -> Result { + let huffman_len = huffman::encode_output_length(value, lower_case); + if huffman_len < value.len() { + let huffman_bit = 1u8 + .checked_shl(prefix as u32) + .ok_or(Http3Error::InternalError)?; + let mut off = encode_int(huffman_len as u64, first | huffman_bit, prefix, out)?; + off += huffman::encode(value, &mut out[off..], lower_case)?; + Ok(off) + } else { + let encoded = if lower_case { + value.to_ascii_lowercase() } else { - str_val.to_vec() + value.to_vec() }; + let mut off = encode_int(encoded.len() as u64, first, prefix, out)?; + let mut buf = &mut out[off..]; + off += buf.write(&encoded)?; + Ok(off) + } +} + +fn append_int(out: &mut Vec, value: u64, first: u8, prefix: usize) -> Result<()> { + let mut buf = [0; 16]; + let len = encode_int(value, first, prefix, &mut buf)?; + out.extend_from_slice(&buf[..len]); + Ok(()) +} - Ok((val, buf_len - buf.len())) +fn append_string( + out: &mut Vec, + value: &[u8], + first: u8, + prefix: usize, + lower_case: bool, +) -> Result<()> { + let mut buf = vec![0; value.len() + 32]; + let len = encode_string(value, first, prefix, lower_case, &mut buf)?; + out.extend_from_slice(&buf[..len]); + Ok(()) +} + +fn decode_int_partial(buf: &[u8], prefix: usize) -> Result> { + if buf.is_empty() { + return Ok(None); } - /// Process control instructions from the encoder. - pub fn process(&mut self, _buf: &mut [u8]) -> Result<()> { - // TODO: support instructions - Ok(()) + let mask = 2u64 + .checked_pow(prefix as u32) + .and_then(|value| value.checked_sub(1)) + .ok_or(Http3Error::QpackDecompressionFailed)?; + let mut value = u64::from(buf[0]) & mask; + if value < mask { + return Ok(Some((value, 1))); + } + + let mut shift = 0; + for (offset, byte) in buf[1..].iter().copied().enumerate() { + if shift >= 62 { + return Err(Http3Error::QpackDecompressionFailed); + } + let increment = u64::from(byte & 0x7f) + .checked_shl(shift) + .ok_or(Http3Error::QpackDecompressionFailed)?; + value = value + .checked_add(increment) + .filter(|value| *value <= MAX_QPACK_INT) + .ok_or(Http3Error::QpackDecompressionFailed)?; + if byte & 0x80 == 0 { + return Ok(Some((value, offset + 2))); + } + shift += 7; + } + + Ok(None) +} + +fn decode_string_partial(buf: &[u8], prefix: usize) -> Result, usize)>> { + if buf.is_empty() { + return Ok(None); + } + + let huffman_bit = 1u8 + .checked_shl(prefix as u32) + .ok_or(Http3Error::QpackDecompressionFailed)?; + let huffman_encoded = buf[0] & huffman_bit != 0; + let Some((length, prefix_len)) = decode_int_partial(buf, prefix)? else { + return Ok(None); + }; + let length = usize::try_from(length).map_err(|_| Http3Error::QpackDecompressionFailed)?; + let end = prefix_len + .checked_add(length) + .ok_or(Http3Error::QpackDecompressionFailed)?; + if buf.len() < end { + return Ok(None); + } + + let encoded = &buf[prefix_len..end]; + let value = if huffman_encoded { + huffman::decode(encoded)? + } else { + encoded.to_vec() + }; + Ok(Some((value, end))) +} + +fn encode_required_insert_count(required_insert_count: u64, max_capacity: u64) -> Result { + if required_insert_count == 0 { + return Ok(0); + } + + let max_entries = max_capacity / 32; + let full_range = max_entries + .checked_mul(2) + .filter(|value| *value != 0) + .ok_or(Http3Error::InternalError)?; + Ok(required_insert_count % full_range + 1) +} + +fn decode_required_insert_count( + encoded_insert_count: u64, + total_insert_count: u64, + max_capacity: u64, +) -> Result { + if encoded_insert_count == 0 { + return Ok(0); + } + + let max_entries = max_capacity / 32; + let full_range = max_entries + .checked_mul(2) + .filter(|value| *value != 0) + .ok_or(Http3Error::QpackDecompressionFailed)?; + if encoded_insert_count > full_range { + return Err(Http3Error::QpackDecompressionFailed); + } + + let max_value = total_insert_count + .checked_add(max_entries) + .ok_or(Http3Error::QpackDecompressionFailed)?; + let max_wrapped = max_value / full_range * full_range; + let mut required_insert_count = max_wrapped + .checked_add(encoded_insert_count - 1) + .ok_or(Http3Error::QpackDecompressionFailed)?; + if required_insert_count > max_value { + if required_insert_count <= full_range { + return Err(Http3Error::QpackDecompressionFailed); + } + required_insert_count -= full_range; + } + if required_insert_count == 0 { + return Err(Http3Error::QpackDecompressionFailed); } + Ok(required_insert_count) +} + +fn parse_encoder_instruction(buf: &[u8]) -> Result> { + if buf.is_empty() { + return Ok(None); + } + + let first = buf[0]; + if first & INSERT_WITH_NAME_REF != 0 { + let Some((index, prefix_len)) = decode_int_partial(buf, 6)? else { + return Ok(None); + }; + let Some((value, value_len)) = decode_string_partial(&buf[prefix_len..], 7)? else { + return Ok(None); + }; + return Ok(Some(( + EncoderInstruction::InsertWithNameReference { + static_table: first & 0x40 != 0, + index, + value, + }, + prefix_len + value_len, + ))); + } + + if first & 0xc0 == INSERT_WITH_LITERAL_NAME { + let Some((name, name_len)) = decode_string_partial(buf, 5)? else { + return Ok(None); + }; + let Some((value, value_len)) = decode_string_partial(&buf[name_len..], 7)? else { + return Ok(None); + }; + return Ok(Some(( + EncoderInstruction::InsertWithLiteralName { name, value }, + name_len + value_len, + ))); + } + + if first & 0xe0 == SET_DYNAMIC_TABLE_CAPACITY { + let Some((capacity, len)) = decode_int_partial(buf, 5)? else { + return Ok(None); + }; + return Ok(Some((EncoderInstruction::SetCapacity(capacity), len))); + } + + let Some((index, len)) = decode_int_partial(buf, 5)? else { + return Ok(None); + }; + Ok(Some((EncoderInstruction::Duplicate(index), len))) +} + +fn relative_absolute(base: u64, index: u64, required_insert_count: u64) -> Result { + let relative = index + .checked_add(1) + .ok_or(Http3Error::QpackDecompressionFailed)?; + let absolute_index = base + .checked_sub(relative) + .ok_or(Http3Error::QpackDecompressionFailed)?; + if absolute_index >= required_insert_count { + return Err(Http3Error::QpackDecompressionFailed); + } + Ok(absolute_index) +} + +fn post_base_absolute(base: u64, index: u64, required_insert_count: u64) -> Result { + let absolute_index = base + .checked_add(index) + .ok_or(Http3Error::QpackDecompressionFailed)?; + if absolute_index >= required_insert_count { + return Err(Http3Error::QpackDecompressionFailed); + } + Ok(absolute_index) +} + +fn update_largest_reference(largest_reference: &mut Option, absolute_index: u64) { + *largest_reference = + Some(largest_reference.map_or(absolute_index, |current| current.max(absolute_index))); +} + +fn charge_field(left: &mut u64, name: &[u8], value: &[u8]) -> Result<()> { + let field_size = (name.len() as u64) + .checked_add(value.len() as u64) + .and_then(|size| size.checked_add(32)) + .ok_or(Http3Error::QpackDecompressionFailed)?; + *left = left + .checked_sub(field_size) + .ok_or(Http3Error::QpackDecompressionFailed)?; + Ok(()) } #[cfg(test)] @@ -563,8 +1370,267 @@ mod tests { let mut dec = QpackDecoder::new(); assert!(dec.decode(&buf, 1024 * 16).is_err()); } + + #[test] + fn rfc9204_appendix_b_dynamic_table() { + let encoder_instructions = + hex::decode("3fbd01c00f7777772e6578616d706c652e636f6dc10c2f73616d706c652f70617468") + .unwrap(); + let field_section = hex::decode("03811011").unwrap(); + + let mut decoder = QpackDecoder::with_max_capacity(220); + assert_eq!( + decoder + .process_encoder_instructions(&encoder_instructions) + .unwrap(), + vec![0x02] + ); + + let status = decoder + .decode_field_section(4, &field_section, u64::MAX) + .unwrap(); + assert_eq!( + status, + DecodeStatus::Decoded { + headers: vec![ + Header::new(b":authority", b"www.example.com"), + Header::new(b":path", b"/sample/path"), + ], + consumed: field_section.len(), + decoder_instructions: vec![0x84], + } + ); + + // Appendix B.3: speculative insertion with a literal name. + let speculative_insert = + hex::decode("4a637573746f6d2d6b65790c637573746f6d2d76616c7565").unwrap(); + assert_eq!( + decoder + .process_encoder_instructions(&speculative_insert) + .unwrap(), + vec![0x01] + ); + + // Appendix B.4: the field section blocks until the delayed Duplicate + // arrives, and can be canceled while it is blocked. + let blocked_field_section = hex::decode("050080c181").unwrap(); + assert_eq!( + decoder + .decode_field_section(8, &blocked_field_section, u64::MAX) + .unwrap(), + DecodeStatus::Blocked { + required_insert_count: 4, + } + ); + assert_eq!(decoder.stream_cancellation(8).unwrap(), vec![0x48]); + assert_eq!( + decoder.process_encoder_instructions(&[0x02]).unwrap(), + vec![0x01] + ); + assert_eq!( + decoder + .decode_field_section(8, &blocked_field_section, u64::MAX) + .unwrap(), + DecodeStatus::Decoded { + headers: vec![ + Header::new(b":authority", b"www.example.com"), + Header::new(b":path", b"/"), + Header::new(b"custom-key", b"custom-value"), + ], + consumed: blocked_field_section.len(), + decoder_instructions: vec![0x88], + } + ); + + // Appendix B.5: an insertion using a dynamic name reference evicts + // the oldest entry while absolute indices remain stable. + let insert_with_dynamic_name = hex::decode("810d637573746f6d2d76616c756532").unwrap(); + assert_eq!( + decoder + .process_encoder_instructions(&insert_with_dynamic_name) + .unwrap(), + vec![0x01] + ); + assert_eq!(decoder.insert_count(), 5); + assert!(decoder.table.get_absolute(0).is_none()); + assert_eq!(decoder.table.get_absolute(4).unwrap().name, b"custom-key"); + assert_eq!( + decoder.table.get_absolute(4).unwrap().value, + b"custom-value2" + ); + } + + #[test] + fn dynamic_table_encoder_round_trip() { + let headers = vec![Header::new(b"x-foo", b"bar")]; + let mut encoder = QpackEncoder::new(); + encoder.set_max_capacity(256).unwrap(); + let mut decoder = QpackDecoder::with_max_capacity(256); + let mut field_section = [0; 128]; + + let (first_len, encoder_instructions) = encoder + .encode_for_stream(0, &headers, &mut field_section) + .unwrap(); + assert!(!encoder_instructions.is_empty()); + let feedback = decoder + .process_encoder_instructions(&encoder_instructions) + .unwrap(); + assert_eq!(feedback, vec![0x01]); + assert_eq!( + decoder + .decode(&field_section[..first_len], u64::MAX) + .unwrap() + .0, + headers + ); + + encoder.process_decoder_instructions(&feedback).unwrap(); + let (second_len, encoder_instructions) = encoder + .encode_for_stream(0, &headers, &mut field_section) + .unwrap(); + assert!(encoder_instructions.is_empty()); + assert!(second_len < first_len); + + let status = decoder + .decode_field_section(0, &field_section[..second_len], u64::MAX) + .unwrap(); + assert_eq!( + status, + DecodeStatus::Decoded { + headers, + consumed: second_len, + decoder_instructions: vec![0x80], + } + ); + encoder.process_decoder_instructions(&[0x80]).unwrap(); + + let renamed_value = vec![Header::new(b"x-foo", b"baz")]; + let (third_len, encoder_instructions) = encoder + .encode_for_stream(4, &renamed_value, &mut field_section) + .unwrap(); + assert!(!encoder_instructions.is_empty()); + assert_eq!( + decoder + .process_encoder_instructions(&encoder_instructions) + .unwrap(), + vec![0x01] + ); + assert_eq!( + decoder + .decode_field_section(4, &field_section[..third_len], u64::MAX) + .unwrap(), + DecodeStatus::Decoded { + headers: renamed_value, + consumed: third_len, + decoder_instructions: vec![0x84], + } + ); + } + + #[test] + fn literal_with_post_base_name_reference() { + let mut decoder = QpackDecoder::with_max_capacity(64); + let mut encoder_instructions = Vec::new(); + append_int(&mut encoder_instructions, 64, SET_DYNAMIC_TABLE_CAPACITY, 5).unwrap(); + append_string( + &mut encoder_instructions, + b"x-name", + INSERT_WITH_LITERAL_NAME, + 5, + false, + ) + .unwrap(); + append_string(&mut encoder_instructions, b"old", 0, 7, false).unwrap(); + decoder + .process_encoder_instructions(&encoder_instructions) + .unwrap(); + + // Required Insert Count=1, Base=0, followed by a literal whose name + // uses post-Base index 0 (absolute index 0). + let field_section = [0x02, 0x80, 0x00, 0x03, b'n', b'e', b'w']; + assert_eq!( + decoder + .decode_field_section(4, &field_section, u64::MAX) + .unwrap(), + DecodeStatus::Decoded { + headers: vec![Header::new(b"x-name", b"new")], + consumed: field_section.len(), + decoder_instructions: vec![0x84], + } + ); + } + + #[test] + fn rejects_overflowing_delta_base() { + let mut field_section = vec![0]; + append_int(&mut field_section, u64::MAX, 0x80, 7).unwrap(); + + let mut decoder = QpackDecoder::new(); + assert_eq!( + decoder.decode_field_section(0, &field_section, u64::MAX), + Err(Http3Error::QpackDecompressionFailed) + ); + } + + #[test] + fn partial_integer_rejects_overlong_high_bits() { + let mut encoded = vec![0b11111]; + encoded.extend(std::iter::repeat_n(0b1000_0000, 9)); + encoded.push(0b0000_0010); + assert_eq!( + decode_int_partial(&encoded, 5), + Err(Http3Error::QpackDecompressionFailed) + ); + } + + #[test] + fn encoder_instructions_can_arrive_incrementally() { + let instructions = + hex::decode("3fbd014a637573746f6d2d6b65790c637573746f6d2d76616c7565").unwrap(); + let mut decoder = QpackDecoder::with_max_capacity(220); + for byte in &instructions[..instructions.len() - 1] { + assert!(decoder + .process_encoder_instructions(&[*byte]) + .unwrap() + .is_empty()); + } + assert_eq!( + decoder + .process_encoder_instructions(&instructions[instructions.len() - 1..]) + .unwrap(), + vec![0x01] + ); + } + + #[test] + fn rejects_capacity_above_advertised_limit() { + let mut decoder = QpackDecoder::with_max_capacity(64); + let mut instruction = Vec::new(); + append_int(&mut instruction, 65, SET_DYNAMIC_TABLE_CAPACITY, 5).unwrap(); + assert_eq!( + decoder.process_encoder_instructions(&instruction), + Err(Http3Error::QpackEncoderStreamError) + ); + } + + #[test] + fn sensitive_fields_are_never_inserted() { + let mut encoder = QpackEncoder::new(); + encoder.set_max_capacity(256).unwrap(); + let mut field_section = [0; 128]; + let (_, instructions) = encoder + .encode_for_stream( + 0, + &[Header::new(b"authorization", b"secret")], + &mut field_section, + ) + .unwrap(); + assert!(instructions.is_empty()); + assert_ne!(field_section[2] & 0x20, 0); + } } +mod dynamic_table; mod huffman; mod prefix_int; mod static_table; diff --git a/src/h3/stream.rs b/src/h3/stream.rs index 63a8b55f7..422cb16c4 100644 --- a/src/h3/stream.rs +++ b/src/h3/stream.rs @@ -73,6 +73,10 @@ pub struct Http3Stream { /// Stream header blocked by flow control, buffered here until it can be sent fully. /// The tuple contains the encoded header block and whether it carries the fin flag. header_block: Option<(Bytes, bool)>, + + /// Whether processing is suspended until the QPACK decoder receives the + /// dynamic table entries required by the current field section. + qpack_blocked: bool, } impl Http3Stream { @@ -102,6 +106,7 @@ impl Http3Stream { priority_initialized: false, priority_update: None, header_block: None, + qpack_blocked: false, } } @@ -687,6 +692,14 @@ impl Http3Stream { pub fn has_header_block(&self) -> bool { self.header_block.is_some() } + + pub fn set_qpack_blocked(&mut self, blocked: bool) { + self.qpack_blocked = blocked; + } + + pub fn qpack_blocked(&self) -> bool { + self.qpack_blocked + } } /// HTTP/3 stream types.