use std::io;
use std::io::Write;
use std::net::ToSocketAddrs;
use std::time::{Duration, Instant};
use crate::net::udp::{open_udp_transport, UdpTransport};
use compcol::hpack::HeaderField;
use compcol::qpack::{QpackDecoder, QpackEncoder};
use purecrypto::quic::transport_params::TransportParameters;
use purecrypto::quic::{QuicConfig, QuicConnection, StreamId};
use crate::error::{Error, Result};
use crate::{Request, Response};
pub(crate) mod varint {
use crate::error::{Error, Result};
pub const MAX: u64 = (1u64 << 62) - 1;
#[allow(dead_code)]
pub const fn encoded_len(value: u64) -> usize {
if value < 1 << 6 {
1
} else if value < 1 << 14 {
2
} else if value < 1 << 30 {
4
} else {
8
}
}
pub fn encode(value: u64, out: &mut Vec<u8>) {
debug_assert!(value <= MAX, "QUIC varint out of range: {value:#x}");
if value < 1 << 6 {
out.push(value as u8);
} else if value < 1 << 14 {
let bytes = (value as u16).to_be_bytes();
out.push(bytes[0] | 0x40);
out.push(bytes[1]);
} else if value < 1 << 30 {
let bytes = (value as u32).to_be_bytes();
out.push(bytes[0] | 0x80);
out.push(bytes[1]);
out.push(bytes[2]);
out.push(bytes[3]);
} else {
let bytes = value.to_be_bytes();
out.push(bytes[0] | 0xC0);
out.extend_from_slice(&bytes[1..]);
}
}
pub fn decode(buf: &[u8]) -> Result<(u64, usize)> {
if buf.is_empty() {
return Err(Error::BadResponse("varint: empty input".into()));
}
let tag = buf[0] >> 6;
let n: usize = 1 << tag; if buf.len() < n {
return Err(Error::BadResponse(format!(
"varint: need {n} bytes, have {}",
buf.len()
)));
}
let mut v: u64 = (buf[0] & 0x3F) as u64;
for &b in &buf[1..n] {
v = (v << 8) | (b as u64);
}
Ok((v, n))
}
}
#[allow(dead_code)]
pub(crate) mod frame_type {
pub const DATA: u64 = 0x00;
pub const HEADERS: u64 = 0x01;
pub const CANCEL_PUSH: u64 = 0x03;
pub const SETTINGS: u64 = 0x04;
pub const PUSH_PROMISE: u64 = 0x05;
pub const GOAWAY: u64 = 0x07;
pub const MAX_PUSH_ID: u64 = 0x0D;
}
#[allow(dead_code)]
pub(crate) mod uni_stream_type {
pub const CONTROL: u64 = 0x00;
pub const PUSH: u64 = 0x01;
pub const QPACK_ENCODER: u64 = 0x02;
pub const QPACK_DECODER: u64 = 0x03;
}
#[allow(dead_code)]
pub(crate) mod settings_id {
pub const QPACK_MAX_TABLE_CAPACITY: u64 = 0x01;
pub const MAX_FIELD_SECTION_SIZE: u64 = 0x06;
pub const QPACK_BLOCKED_STREAMS: u64 = 0x07;
}
pub(crate) const QPACK_MAX_TABLE_CAPACITY: u64 = 4096;
pub(crate) const QPACK_BLOCKED_STREAMS: u64 = 0;
#[derive(Debug, PartialEq, Eq)]
pub(crate) struct Frame {
pub ty: u64,
pub len: u64,
}
impl Frame {
pub fn encode_header(ty: u64, len: u64, out: &mut Vec<u8>) {
varint::encode(ty, out);
varint::encode(len, out);
}
pub fn decode_header(buf: &[u8]) -> Result<(Frame, usize)> {
let (ty, n1) = varint::decode(buf)?;
let (len, n2) = varint::decode(&buf[n1..])?;
Ok((Frame { ty, len }, n1 + n2))
}
}
const MAX_DECODED_HEADER_LIST: usize = 256 * 1024;
type Fields = Vec<(String, String)>;
fn decode_header_block(decoder: &mut QpackDecoder, block: &[u8]) -> Result<Fields> {
let decoded = decoder
.decode_field_section(block)
.map_err(|e| Error::BadResponse(format!("qpack: decode failed: {e}")))?;
let mut out: Fields = Vec::with_capacity(decoded.len());
let mut list_size: usize = 0;
for f in decoded {
if !header_octets_ok(&f.name, &f.value) {
return Err(Error::BadResponse(
"qpack: forbidden octet in decoded header".into(),
));
}
list_size = list_size
.saturating_add(f.name.len())
.saturating_add(f.value.len())
.saturating_add(32);
if list_size > MAX_DECODED_HEADER_LIST {
return Err(Error::BadResponse(
"qpack: decoded header list exceeds limit".into(),
));
}
let name = String::from_utf8(f.name)
.map_err(|_| Error::BadResponse("qpack: header name not utf-8".into()))?;
let value = String::from_utf8(f.value)
.map_err(|_| Error::BadResponse("qpack: header value not utf-8".into()))?;
out.push((name, value));
}
Ok(out)
}
fn encode_header_block(fields: &[(String, String)]) -> Vec<u8> {
let hfields: Vec<HeaderField> = fields
.iter()
.map(|(n, v)| HeaderField::new(n.as_bytes(), v.as_bytes()))
.collect();
let mut enc = QpackEncoder::new();
enc.set_huffman(true);
enc.encode_field_section(&hfields)
}
fn block_references_dynamic_table(block: &[u8]) -> bool {
!block.is_empty() && block[0] != 0
}
fn header_octets_ok(name: &[u8], value: &[u8]) -> bool {
if name.is_empty() {
return false;
}
let name_rest = if name[0] == b':' { &name[1..] } else { name };
if name_rest.is_empty() {
return false;
}
if !name_rest.iter().all(|&c| is_token_char(c)) {
return false;
}
!value.iter().any(|&c| c == 0x00 || c == 0x0a || c == 0x0d)
}
fn is_token_char(c: u8) -> bool {
c.is_ascii_digit()
|| c.is_ascii_lowercase()
|| matches!(
c,
b'!' | b'#'
| b'$'
| b'%'
| b'&'
| b'\''
| b'*'
| b'+'
| b'-'
| b'.'
| b'^'
| b'_'
| b'`'
| b'|'
| b'~'
)
}
fn encode_prefixed_int(value: u64, prefix_bits: u8, pattern: u8, out: &mut Vec<u8>) {
debug_assert!((1..=8).contains(&prefix_bits));
let max_prefix = (1u64 << prefix_bits) - 1;
if value < max_prefix {
out.push(pattern | value as u8);
} else {
out.push(pattern | max_prefix as u8);
let mut rem = value - max_prefix;
while rem >= 128 {
out.push(((rem & 0x7f) as u8) | 0x80);
rem >>= 7;
}
out.push(rem as u8);
}
}
fn complete_encoder_instructions_len(buf: &[u8]) -> usize {
let mut pos = 0;
while let Some(end) = next_instruction_end(buf, pos) {
pos = end;
}
pos
}
fn next_instruction_end(buf: &[u8], pos: usize) -> Option<usize> {
let b = *buf.get(pos)?;
if b & 0b1000_0000 != 0 {
let p = skip_int(buf, pos, 6)?;
skip_string(buf, p, 7)
} else if b & 0b0100_0000 != 0 {
let p = skip_string(buf, pos, 5)?;
skip_string(buf, p, 7)
} else {
skip_int(buf, pos, 5)
}
}
fn skip_int(buf: &[u8], pos: usize, prefix_bits: u32) -> Option<usize> {
let mask = ((1u16 << prefix_bits) - 1) as u8;
let first = *buf.get(pos)?;
if first & mask != mask {
return Some(pos + 1);
}
let mut p = pos + 1;
loop {
let b = *buf.get(p)?;
p += 1;
if b & 0x80 == 0 {
return Some(p);
}
}
}
fn skip_string(buf: &[u8], pos: usize, prefix_bits: u32) -> Option<usize> {
let mask = ((1u16 << prefix_bits) - 1) as u8;
let first = *buf.get(pos)?;
let (len, mut p) = if first & mask != mask {
((first & mask) as u64, pos + 1)
} else {
let mut value = mask as u64;
let mut shift = 0u32;
let mut q = pos + 1;
loop {
let b = *buf.get(q)?;
q += 1;
value = match value.checked_add(((b & 0x7f) as u64) << shift) {
Some(v) => v,
None => return Some(q), };
if b & 0x80 == 0 {
break;
}
shift += 7;
if shift > 63 {
return Some(q); }
}
(value, q)
};
let len = usize::try_from(len).ok()?;
p = p.checked_add(len)?;
if p > buf.len() {
return None;
}
Some(p)
}
const MAX_RESPONSE_BYTES: usize = 256 * 1024 * 1024;
const MAX_HEADERS_FRAME_LEN: u64 = 256 * 1024;
const MAX_TOTAL_DEADLINE: Duration = Duration::from_secs(300);
const PUMP_READ_CAP: Duration = Duration::from_millis(100);
const MAX_DATAGRAM: usize = 65_535;
struct Http3State {
decoder: QpackDecoder,
decoder_stream: Option<StreamId>,
uni: std::collections::HashMap<u64, UniStreamState>,
}
#[derive(Default)]
struct UniStreamState {
buf: Vec<u8>,
ty: Option<u64>,
}
const MAX_UNI_BUFFER: usize = 64 * 1024;
impl Http3State {
fn new(decoder_stream: Option<StreamId>) -> Self {
Http3State {
decoder: QpackDecoder::with_max_table_capacity(QPACK_MAX_TABLE_CAPACITY as usize),
decoder_stream,
uni: std::collections::HashMap::new(),
}
}
}
pub fn send(req: Request, trace: &mut dyn Write) -> Result<Response> {
send_inner(req, None, None, trace)
}
pub fn send_to(
req: Request,
sink: &mut dyn Write,
on_head: Option<crate::http::HeadObserver<'_>>,
trace: &mut dyn Write,
) -> Result<Response> {
send_inner(req, Some(sink), on_head, trace)
}
fn send_inner(
req: Request,
sink: Option<&mut dyn Write>,
on_head: Option<crate::http::HeadObserver<'_>>,
trace: &mut dyn Write,
) -> Result<Response> {
if req.url.scheme != "https" {
return Err(Error::UnsupportedScheme(format!(
"http/3 requires https://, not {}://",
req.url.scheme
)));
}
let pins = match &req.pinned_pubkey {
Some(spec) => crate::tls::parse_pinned_pubkey(spec)?,
None => Vec::new(),
};
let dial_start = Instant::now();
let mut conn = build_client(&req)?;
let (sock, peer) = open_udp(&req)?;
let connect = dial_start.elapsed();
let _ = writeln!(trace, "* Trying {peer} (UDP)...");
handshake(&mut conn, &*sock, peer, req.read_timeout, dial_start)?;
let appconnect = dial_start.elapsed();
verify_peer_certificates(&conn, &req, &pins)?;
let _ = writeln!(
trace,
"* Connected to {} ({}) port {} (QUIC)",
req.url.host,
peer.ip(),
peer.port()
);
let _ = writeln!(trace, "* QUIC connected, TLS 1.3 handshake complete");
match conn.alpn_protocol() {
Some(p) => {
let _ = writeln!(
trace,
"* ALPN: server accepted {}",
String::from_utf8_lossy(p)
);
}
None => {
let _ = writeln!(trace, "* ALPN: no protocol negotiated");
}
}
let _ = writeln!(trace, "* using HTTP/3");
let _ = open_control_stream(&mut conn);
let decoder_stream = open_qpack_streams(&mut conn);
let mut state = Http3State::new(decoder_stream);
let request_stream = conn
.open_bidi()
.map_err(|e| Error::BadResponse(format!("http3: open_bidi failed: {e:?}")))?;
let tls_info = crate::http::TlsInfo {
version: Some(crate::tls::ProtocolVersion::TLSv1_3),
cipher_suite: conn.negotiated_cipher_suite(),
alpn: conn.alpn_protocol().map(|p| p.to_vec()),
peer_certificates: conn.peer_certificates().to_vec(),
};
write_request(&mut conn, request_stream, &req, trace)?;
if !req.body.is_empty() {
let _ = writeln!(trace, "* uploading {} body bytes", req.body.len());
}
pump(&mut conn, &*sock, peer, req.read_timeout, dial_start)?;
let mut resp = read_response(
&mut conn,
&*sock,
peer,
request_stream,
&req,
&mut state,
sink,
on_head,
trace,
dial_start,
)?;
resp.tls = Some(tls_info);
resp.timing.connect = Some(connect);
resp.timing.appconnect = Some(appconnect);
resp.timing.pretransfer = Some(appconnect);
Ok(resp)
}
fn drain_uni_streams(conn: &mut QuicConnection, state: &mut Http3State) -> Result<()> {
let ids: Vec<StreamId> = conn
.readable_streams()
.filter(|s| s.is_uni() && s.is_server_initiated())
.collect();
for sid in ids {
let mut tmp = vec![0u8; 16 * 1024];
while let Ok((n, _fin)) = conn.read(sid, &mut tmp) {
if n == 0 {
break;
}
let entry = state.uni.entry(sid.value()).or_default();
if entry.buf.len() + n > MAX_UNI_BUFFER {
return Err(Error::BadResponse(
"http3: server uni-stream buffer exceeded limit".into(),
));
}
entry.buf.extend_from_slice(&tmp[..n]);
}
process_uni_stream(state, sid.value())?;
}
Ok(())
}
fn process_uni_stream(state: &mut Http3State, sid: u64) -> Result<()> {
let entry = state.uni.entry(sid).or_default();
if entry.ty.is_none() {
match varint::decode(&entry.buf) {
Ok((ty, used)) => {
entry.ty = Some(ty);
entry.buf.drain(..used);
}
Err(_) => return Ok(()), }
}
match entry.ty {
Some(uni_stream_type::QPACK_ENCODER) => {
let consumed = complete_encoder_instructions_len(&entry.buf);
if consumed > 0 {
state
.decoder
.feed_encoder_stream(&entry.buf[..consumed])
.map_err(|e| Error::BadResponse(format!("qpack: encoder stream: {e}")))?;
let entry = state.uni.get_mut(&sid).expect("entry present");
entry.buf.drain(..consumed);
}
}
_ => {
entry.buf.clear();
}
}
Ok(())
}
fn verify_peer_certificates(conn: &QuicConnection, req: &Request, pins: &[[u8; 32]]) -> Result<()> {
let leaf = conn.peer_certificates().first().map(Vec::as_slice);
if !pins.is_empty() {
match leaf {
Some(der) if crate::tls::client_auth::spki_pin_matches(der, pins) => {}
_ => {
return Err(Error::BadResponse(
"pinned public key does not match server certificate".into(),
))
}
}
}
if let Some(cb) = &req.tls_verify_callback {
let chain = conn.peer_certificates().to_vec();
let verdict = cb.call(&crate::tls::CertVerify {
server_name: &req.url.host,
chain_der: &chain,
});
if verdict == crate::tls::CertVerdict::Reject {
return Err(Error::BadResponse(
"server certificate rejected by verify callback".into(),
));
}
return Ok(());
}
if req.verify_tls {
match leaf {
Some(der) if crate::tls::client_auth::leaf_has_san(der) => {}
Some(_) => {
return Err(Error::BadResponse(
"server certificate has no Subject Alternative Name \
(CN fallback is not accepted)"
.into(),
))
}
None => {}
}
}
Ok(())
}
fn build_client(req: &Request) -> Result<QuicConnection> {
let mut roots = match &req.ca_bundle {
Some(path) => crate::tls::pc_roots::load_from_file(path)?,
None => crate::tls::pc_roots::embedded_roots(),
};
if let Some(dir) = &req.ca_path {
crate::tls::pc_roots::add_from_dir(&mut roots, dir)?;
}
let mut builder = purecrypto::tls::Config::builder()
.tls_only()
.roots(roots)
.server_name(req.url.host.clone())
.verify_certificates(req.verify_tls && req.tls_verify_callback.is_none())
.rng(std::sync::Arc::new(purecrypto::rng::OsRng))
.alpn(vec![b"h3".to_vec()]);
if let Some(path) = &req.crl_file {
let crl_bytes = std::fs::read(path).map_err(Error::Io)?;
let mut store = purecrypto::tls::CrlStore::new();
let blocks = std::str::from_utf8(&crl_bytes)
.ok()
.map(|pem| crate::tls::pc_roots::pem_blocks_labelled(pem, "X509 CRL"))
.unwrap_or_default();
if !blocks.is_empty() {
for block in &blocks {
store
.add_pem(block)
.map_err(|_| Error::BadResponse("--crlfile: invalid PEM CRL block".into()))?;
}
} else {
store
.add_der(crl_bytes)
.map_err(|_| Error::BadResponse("--crlfile: not a valid PEM or DER CRL".into()))?;
}
builder = builder.crls(store);
}
let mut cipher_ids: Vec<u16> = Vec::new();
if let Some(spec) = &req.ciphers {
cipher_ids.extend(crate::tls::cipher_names_to_ids(spec)?);
}
if let Some(spec) = &req.tls13_ciphers {
cipher_ids.extend(crate::tls::cipher_names_to_ids(spec)?);
}
if !cipher_ids.is_empty() {
builder = builder.cipher_suites(&cipher_ids);
}
let tls = builder.build();
let transport_params = TransportParameters {
max_idle_timeout_ms: Some(30_000),
max_udp_payload_size: Some(1452),
initial_max_data: Some(10 * 1024 * 1024),
initial_max_stream_data_bidi_local: Some(2 * 1024 * 1024),
initial_max_stream_data_bidi_remote: Some(2 * 1024 * 1024),
initial_max_stream_data_uni: Some(2 * 1024 * 1024),
initial_max_streams_bidi: Some(100),
initial_max_streams_uni: Some(100),
active_connection_id_limit: Some(2),
..Default::default()
};
#[allow(clippy::field_reassign_with_default)]
let cfg = {
let mut cfg = QuicConfig::default();
cfg.tls = tls;
cfg.transport_params = transport_params;
cfg
};
QuicConnection::client(cfg, &req.url.host)
.map_err(|e| Error::BadResponse(format!("http3: build client: {e:?}")))
}
fn open_udp(req: &Request) -> Result<(Box<dyn UdpTransport>, std::net::SocketAddr)> {
let host_port = format!("{}:{}", req.url.host, req.url.port);
let peer = host_port
.to_socket_addrs()?
.next()
.ok_or_else(|| Error::InvalidUrl(req.url.host.clone()))?;
let sock = open_udp_transport(req.connector.udp_proxy(), peer)?;
sock.set_read_timeout(Some(Duration::from_millis(100)))?;
sock.set_write_timeout(req.read_timeout)?;
Ok((sock, peer))
}
fn pump_once(
conn: &mut QuicConnection,
sock: &dyn UdpTransport,
peer: std::net::SocketAddr,
can_block: bool,
start: Instant,
) -> Result<bool> {
let mut sent_anything = false;
loop {
let dg = conn.pop_datagram();
if dg.is_empty() {
break;
}
sock.send_to(&dg, peer)?;
sent_anything = true;
}
let mut buf = vec![0u8; MAX_DATAGRAM];
let mut got_anything = false;
if can_block {
let wait = conn
.next_timeout()
.unwrap_or(PUMP_READ_CAP)
.clamp(Duration::from_millis(1), PUMP_READ_CAP);
sock.set_read_timeout(Some(wait))?;
match sock.recv_from(&mut buf) {
Ok((n, _from)) => {
conn.feed_datagram_from(peer, &buf[..n])
.map_err(|e| Error::BadResponse(format!("http3: feed: {e:?}")))?;
got_anything = true;
}
Err(e)
if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut => {
}
Err(e) => return Err(Error::Io(e)),
}
}
conn.on_timeout(start.elapsed());
loop {
let dg = conn.pop_datagram();
if dg.is_empty() {
break;
}
sock.send_to(&dg, peer)?;
sent_anything = true;
}
Ok(sent_anything || got_anything)
}
fn handshake(
conn: &mut QuicConnection,
sock: &dyn UdpTransport,
peer: std::net::SocketAddr,
deadline_hint: Option<Duration>,
start: Instant,
) -> Result<()> {
let total_deadline = deadline_hint
.unwrap_or(MAX_TOTAL_DEADLINE)
.min(MAX_TOTAL_DEADLINE);
while !conn.is_handshake_complete() {
if start.elapsed() > total_deadline {
return Err(Error::Io(io::Error::new(
io::ErrorKind::TimedOut,
"http3: QUIC handshake timed out",
)));
}
pump_once(conn, sock, peer, true, start)?;
if conn.is_closed() {
return Err(Error::BadResponse(
"http3: connection closed mid-handshake".into(),
));
}
}
Ok(())
}
fn open_control_stream(conn: &mut QuicConnection) -> Result<()> {
let sid = conn
.open_uni()
.map_err(|e| Error::BadResponse(format!("http3: open_uni: {e:?}")))?;
let mut prefix = Vec::with_capacity(16);
varint::encode(uni_stream_type::CONTROL, &mut prefix);
let mut settings = Vec::with_capacity(8);
varint::encode(settings_id::QPACK_MAX_TABLE_CAPACITY, &mut settings);
varint::encode(QPACK_MAX_TABLE_CAPACITY, &mut settings);
varint::encode(settings_id::QPACK_BLOCKED_STREAMS, &mut settings);
varint::encode(QPACK_BLOCKED_STREAMS, &mut settings);
Frame::encode_header(frame_type::SETTINGS, settings.len() as u64, &mut prefix);
prefix.extend_from_slice(&settings);
write_all(conn, sid, &prefix)?;
Ok(())
}
fn open_qpack_streams(conn: &mut QuicConnection) -> Option<StreamId> {
if let Ok(enc) = conn.open_uni() {
let mut buf = Vec::with_capacity(1);
varint::encode(uni_stream_type::QPACK_ENCODER, &mut buf);
let _ = write_all(conn, enc, &buf);
}
let dec = conn.open_uni().ok()?;
let mut buf = Vec::with_capacity(1);
varint::encode(uni_stream_type::QPACK_DECODER, &mut buf);
if write_all(conn, dec, &buf).is_err() {
return None;
}
Some(dec)
}
fn encode_section_ack(stream_id: u64, out: &mut Vec<u8>) {
encode_prefixed_int(stream_id, 7, 0b1000_0000, out);
}
fn write_all(conn: &mut QuicConnection, sid: StreamId, mut data: &[u8]) -> Result<()> {
while !data.is_empty() {
let n = conn
.write(sid, data)
.map_err(|e| Error::BadResponse(format!("http3: stream write: {e:?}")))?;
if n == 0 {
return Err(Error::BadResponse(
"http3: stream write blocked (flow control)".into(),
));
}
data = &data[n..];
}
Ok(())
}
fn write_request(
conn: &mut QuicConnection,
sid: StreamId,
req: &Request,
trace: &mut dyn Write,
) -> Result<()> {
let host_port = if req.url.port == 443 {
req.url.host.clone()
} else {
format!("{}:{}", req.url.host, req.url.port)
};
let mut fields: Vec<(String, String)> = Vec::with_capacity(req.headers.len() + 5);
fields.push((":method".into(), crate::http::effective_method(req)));
fields.push((":scheme".into(), "https".into()));
fields.push((":authority".into(), host_port));
fields.push((":path".into(), req.url.path.clone()));
let mut have_ua = false;
let mut have_accept_enc = false;
for (k, v) in &req.headers {
let kl = k.to_ascii_lowercase();
if kl.starts_with(':')
|| kl == "host"
|| kl == "connection"
|| kl == "transfer-encoding"
|| kl == "upgrade"
|| kl == "keep-alive"
|| kl == "proxy-connection"
{
continue;
}
if kl == "user-agent" {
have_ua = true;
}
if kl == "accept-encoding" {
have_accept_enc = true;
}
fields.push((kl, v.clone()));
}
if !req.strict_headers {
if !have_ua {
fields.push((
"user-agent".into(),
format!("rsurl/{}", env!("CARGO_PKG_VERSION")),
));
}
if !have_accept_enc {
fields.push(("accept-encoding".into(), "gzip, deflate".into()));
}
}
if !req.body.is_empty() {
fields.push(("content-length".into(), req.body.len().to_string()));
}
{
let path = fields
.iter()
.find(|(k, _)| k == ":path")
.map(|(_, v)| v.as_str())
.unwrap_or("/");
let _ = writeln!(
trace,
"> {} {path} HTTP/3",
crate::http::effective_method(req)
);
if let Some((_, authority)) = fields.iter().find(|(k, _)| k == ":authority") {
let _ = writeln!(trace, "> Host: {authority}");
}
for (k, v) in &fields {
if !k.starts_with(':') {
let _ = writeln!(trace, "> {k}: {v}");
}
}
let _ = writeln!(trace, "> ");
}
let qpack_payload = encode_header_block(&fields);
let mut out = Vec::with_capacity(qpack_payload.len() + 16);
Frame::encode_header(frame_type::HEADERS, qpack_payload.len() as u64, &mut out);
out.extend_from_slice(&qpack_payload);
if !req.body.is_empty() {
Frame::encode_header(frame_type::DATA, req.body.len() as u64, &mut out);
out.extend_from_slice(&req.body);
}
write_all(conn, sid, &out)?;
conn.finish(sid)
.map_err(|e| Error::BadResponse(format!("http3: stream finish: {e:?}")))?;
Ok(())
}
fn pump(
conn: &mut QuicConnection,
sock: &dyn UdpTransport,
peer: std::net::SocketAddr,
_read_timeout: Option<Duration>,
start: Instant,
) -> Result<()> {
for _ in 0..3 {
pump_once(conn, sock, peer, false, start)?;
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
fn read_response(
conn: &mut QuicConnection,
sock: &dyn UdpTransport,
peer: std::net::SocketAddr,
sid: StreamId,
req: &Request,
state: &mut Http3State,
mut sink: Option<&mut dyn Write>,
mut on_head: Option<crate::http::HeadObserver<'_>>,
trace: &mut dyn Write,
conn_start: Instant,
) -> Result<Response> {
let mut streamed_len: u64 = 0;
let total_deadline = req
.read_timeout
.unwrap_or(MAX_TOTAL_DEADLINE)
.min(MAX_TOTAL_DEADLINE);
let start = Instant::now();
let mut stream_buf: Vec<u8> = Vec::new();
let mut headers: Option<Fields> = None;
let mut body: Vec<u8> = Vec::new();
loop {
if start.elapsed() > total_deadline {
return Err(Error::Io(io::Error::new(
io::ErrorKind::TimedOut,
"http3: response timed out",
)));
}
if conn.is_closed() {
return Err(Error::BadResponse("http3: peer closed connection".into()));
}
drain_uni_streams(conn, state)?;
let mut tmp = vec![0u8; 16 * 1024];
let (n, fin) = match conn.read(sid, &mut tmp) {
Ok(x) => x,
Err(e) => return Err(Error::BadResponse(format!("http3: stream read: {e:?}"))),
};
if n > 0 {
if stream_buf.len() + n > MAX_RESPONSE_BYTES {
return Err(Error::BadResponse("http3: response too large".into()));
}
stream_buf.extend_from_slice(&tmp[..n]);
}
loop {
let encoded = req.decompress
&& headers.as_ref().is_some_and(|f| {
f.iter()
.any(|(k, _)| k.eq_ignore_ascii_case("content-encoding"))
});
let frame_sink: Option<&mut dyn Write> = if encoded {
None
} else {
match &mut sink {
Some(w) => Some(&mut **w),
None => None,
}
};
let (consumed, ack_owed) = match try_consume_frame(
&stream_buf,
&mut headers,
&mut body,
&mut state.decoder,
frame_sink,
&mut streamed_len,
) {
FrameOutcome::Consumed(n, ack) => (n, ack),
FrameOutcome::NeedMore => break,
FrameOutcome::Err(e) => return Err(e),
};
if ack_owed {
send_section_ack(conn, sid, state);
}
stream_buf.drain(..consumed);
if on_head.is_some() {
if let Some(fields) = headers.as_ref() {
fire_h3_head(fields, &mut on_head);
}
}
if stream_buf.is_empty() {
break;
}
}
if fin {
if !stream_buf.is_empty() {
return Err(Error::BadResponse(
"http3: stream FIN with partial frame in buffer".into(),
));
}
break;
}
pump_once(conn, sock, peer, true, conn_start)?;
}
let fields = headers.ok_or_else(|| Error::BadResponse("http3: no HEADERS frame".into()))?;
finalize_response(fields, body, streamed_len, req.decompress, sink, trace)
}
enum FrameOutcome {
Consumed(usize, bool),
NeedMore,
Err(Error),
}
#[allow(clippy::too_many_arguments)]
fn try_consume_frame(
buf: &[u8],
headers: &mut Option<Fields>,
body: &mut Vec<u8>,
decoder: &mut QpackDecoder,
sink: Option<&mut dyn Write>,
streamed_len: &mut u64,
) -> FrameOutcome {
let (frame, hdr_len) = match Frame::decode_header(buf) {
Ok(x) => x,
Err(_) => return FrameOutcome::NeedMore,
};
match frame.ty {
frame_type::HEADERS if frame.len > MAX_HEADERS_FRAME_LEN => {
return FrameOutcome::Err(Error::BadResponse(
"http3: HEADERS frame length exceeds limit".into(),
));
}
frame_type::DATA => {
let remaining = MAX_RESPONSE_BYTES.saturating_sub(body.len()) as u64;
if frame.len > remaining {
return FrameOutcome::Err(Error::BadResponse(
"http3: DATA frame length exceeds response budget".into(),
));
}
}
frame_type::SETTINGS
| frame_type::GOAWAY
| frame_type::CANCEL_PUSH
| frame_type::MAX_PUSH_ID
| frame_type::PUSH_PROMISE => {
return FrameOutcome::Err(Error::BadResponse(format!(
"http3: frame type {:#x} not allowed on a request stream",
frame.ty
)));
}
_ => {}
}
let frame_len = match usize::try_from(frame.len) {
Ok(n) => n,
Err(_) => {
return FrameOutcome::Err(Error::BadResponse("http3: frame length too large".into()));
}
};
let total = hdr_len.saturating_add(frame_len);
if buf.len() < total {
return FrameOutcome::NeedMore;
}
let payload = &buf[hdr_len..total];
match frame.ty {
frame_type::HEADERS => match decode_header_block(decoder, payload) {
Ok(fields) => {
let ack_owed = block_references_dynamic_table(payload);
if headers.is_some() {
} else if header_status(&fields).is_some_and(|s| (100..200).contains(&s)) {
} else {
*headers = Some(fields);
}
FrameOutcome::Consumed(total, ack_owed)
}
Err(e) => FrameOutcome::Err(e),
},
frame_type::DATA => {
if let Some(w) = sink {
if body.is_empty() {
if let Err(e) = w.write_all(payload) {
return FrameOutcome::Err(Error::Io(e));
}
*streamed_len += payload.len() as u64;
return FrameOutcome::Consumed(total, false);
}
}
body.extend_from_slice(payload);
FrameOutcome::Consumed(total, false)
}
_ => FrameOutcome::Consumed(total, false),
}
}
fn send_section_ack(conn: &mut QuicConnection, request_sid: StreamId, state: &Http3State) {
if let Some(dec) = state.decoder_stream {
let mut out = Vec::with_capacity(4);
encode_section_ack(request_sid.value(), &mut out);
let _ = write_all(conn, dec, &out);
}
}
fn header_status(fields: &Fields) -> Option<u16> {
fields
.iter()
.find(|(k, _)| k == ":status")
.and_then(|(_, v)| v.parse::<u16>().ok())
}
fn fire_h3_head(fields: &Fields, on_head: &mut Option<crate::http::HeadObserver<'_>>) {
let mut status: Option<u16> = None;
let mut hdrs: Vec<(String, String)> = Vec::with_capacity(fields.len());
for (k, v) in fields {
if k == ":status" {
status = v.parse::<u16>().ok();
} else if !k.starts_with(':') {
hdrs.push((k.clone(), v.clone()));
}
}
let Some(status) = status.filter(|s| *s >= 200) else {
return;
};
if let Some(obs) = on_head.take() {
obs(&crate::http::ResponseHead {
status,
reason: String::new(),
version: "HTTP/3".to_string(),
headers: hdrs,
});
}
}
fn finalize_response(
fields: Fields,
body: Vec<u8>,
streamed_len: u64,
decompress: bool,
sink: Option<&mut dyn Write>,
trace: &mut dyn Write,
) -> Result<Response> {
let mut status: Option<u16> = None;
let mut hdrs: Vec<(String, String)> = Vec::with_capacity(fields.len());
for (k, v) in fields {
if k == ":status" {
status = Some(
v.parse()
.map_err(|_| Error::BadResponse(format!("http3: bad :status {v:?}")))?,
);
} else if k.starts_with(':') {
continue;
} else {
hdrs.push((k, v));
}
}
let status = status.ok_or_else(|| Error::BadResponse("http3: missing :status".into()))?;
let _ = writeln!(trace, "< HTTP/3 {status}");
for (k, v) in &hdrs {
let _ = writeln!(trace, "< {k}: {v}");
}
let _ = writeln!(trace, "< ");
let _ = writeln!(
trace,
"* Received {} body bytes",
body.len() as u64 + streamed_len
);
let (hdrs, body) = crate::http::maybe_decode_body(hdrs, body, decompress, trace)?;
if let Some(w) = sink {
if !body.is_empty() {
w.write_all(&body)?;
}
return Ok(Response {
status,
reason: String::new(),
version: "HTTP/3".to_string(),
headers: hdrs,
body: Vec::new(),
timing: crate::http::Timing::default(),
final_url: String::new(),
tls: None,
});
}
Ok(Response {
status,
reason: String::new(),
version: "HTTP/3".to_string(),
headers: hdrs,
body,
timing: crate::http::Timing::default(),
final_url: String::new(),
tls: None,
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn varint_round_trip_size_classes() {
let cases: &[(u64, usize)] = &[
(0, 1),
(63, 1),
(64, 2),
(16_383, 2),
(16_384, 4),
((1 << 30) - 1, 4),
(1 << 30, 8),
(varint::MAX, 8),
];
for &(value, expected_len) in cases {
assert_eq!(varint::encoded_len(value), expected_len, "len({value})");
let mut buf = Vec::new();
varint::encode(value, &mut buf);
assert_eq!(buf.len(), expected_len, "encoded bytes for {value}");
let (decoded, n) = varint::decode(&buf).expect("decode");
assert_eq!(decoded, value, "round-trip value");
assert_eq!(n, expected_len, "round-trip length");
}
}
#[test]
fn varint_rejects_empty_and_truncated() {
assert!(varint::decode(&[]).is_err());
assert!(varint::decode(&[0x40]).is_err());
assert!(varint::decode(&[0xC0, 0x00, 0x00]).is_err());
}
#[test]
fn varint_accepts_non_minimal_encoding() {
let (v, n) = varint::decode(&[0x40, 0x00]).unwrap();
assert_eq!(v, 0);
assert_eq!(n, 2);
}
#[test]
fn http3_frame_header_round_trip() {
let cases: &[(u64, u64)] = &[
(frame_type::DATA, 0),
(frame_type::HEADERS, 17),
(frame_type::SETTINGS, 63),
(frame_type::HEADERS, 64),
(frame_type::DATA, 16_383),
(frame_type::DATA, 16_384),
(frame_type::DATA, 1 << 20),
];
for &(ty, len) in cases {
let mut buf = Vec::new();
Frame::encode_header(ty, len, &mut buf);
let (parsed, used) = Frame::decode_header(&buf).expect("decode_header");
assert_eq!(parsed, Frame { ty, len });
assert_eq!(used, buf.len(), "exact consumption for ({ty},{len})");
}
}
fn decoder() -> QpackDecoder {
QpackDecoder::with_max_table_capacity(QPACK_MAX_TABLE_CAPACITY as usize)
}
fn enc_prefix(enc_ric: u64, sign: bool, delta_base: u64, out: &mut Vec<u8>) {
encode_prefixed_int(enc_ric, 8, 0x00, out);
let pat = if sign { 0b1000_0000 } else { 0 };
encode_prefixed_int(delta_base, 7, pat, out);
}
fn enc_set_capacity(cap: u64, out: &mut Vec<u8>) {
encode_prefixed_int(cap, 5, 0b0010_0000, out);
}
fn enc_insert_literal(name: &str, value: &str, out: &mut Vec<u8>) {
encode_prefixed_int(name.len() as u64, 5, 0b0100_0000, out);
out.extend_from_slice(name.as_bytes());
encode_prefixed_int(value.len() as u64, 7, 0x00, out);
out.extend_from_slice(value.as_bytes());
}
#[test]
fn qpack_encode_decode_round_trip_indexed_and_literal() {
let fields: Fields = vec![
(":method".to_string(), "GET".to_string()),
(":scheme".to_string(), "https".to_string()),
(":authority".to_string(), "example.com".to_string()),
(":path".to_string(), "/index.html".to_string()),
("user-agent".to_string(), "rsurl/test".to_string()),
("x-custom".to_string(), "hello".to_string()),
];
let wire = encode_header_block(&fields);
let decoded = decode_header_block(&mut decoder(), &wire).expect("decode");
assert_eq!(decoded, fields);
}
#[test]
fn qpack_decode_rejects_crlf_in_value() {
let buf =
encode_header_block(&[("x-h".to_string(), "evil\r\nset-cookie: x=1".to_string())]);
let err = decode_header_block(&mut decoder(), &buf).unwrap_err();
assert!(matches!(err, Error::BadResponse(_)), "got {err:?}");
}
#[test]
fn qpack_decode_rejects_lf_in_value() {
let buf = encode_header_block(&[("x-h".to_string(), "a\nb".to_string())]);
assert!(matches!(
decode_header_block(&mut decoder(), &buf).unwrap_err(),
Error::BadResponse(_)
));
}
#[test]
fn qpack_decode_rejects_nul_in_value() {
let buf = encode_header_block(&[("x-h".to_string(), "a\x00b".to_string())]);
assert!(matches!(
decode_header_block(&mut decoder(), &buf).unwrap_err(),
Error::BadResponse(_)
));
}
#[test]
fn qpack_decode_rejects_uppercase_name() {
let buf = encode_header_block(&[("X-Bad".to_string(), "ok".to_string())]);
assert!(matches!(
decode_header_block(&mut decoder(), &buf).unwrap_err(),
Error::BadResponse(_)
));
}
#[test]
fn qpack_decode_rejects_empty_name() {
let buf = encode_header_block(&[("".to_string(), "ok".to_string())]);
assert!(matches!(
decode_header_block(&mut decoder(), &buf).unwrap_err(),
Error::BadResponse(_)
));
}
#[test]
fn qpack_decode_accepts_normal_header_and_pseudo() {
let buf = encode_header_block(&[
(
"content-type".to_string(),
"text/html; charset=utf-8".to_string(),
),
("x-h".to_string(), "a\tb".to_string()),
(":status".to_string(), "200".to_string()),
]);
let fields = decode_header_block(&mut decoder(), &buf).expect("decode");
assert_eq!(
fields[0],
(
"content-type".to_string(),
"text/html; charset=utf-8".to_string()
)
);
assert_eq!(fields[1], ("x-h".to_string(), "a\tb".to_string()));
assert_eq!(fields[2], (":status".to_string(), "200".to_string()));
}
#[test]
fn qpack_oversized_literal_value_length_does_not_panic() {
let mut buf = Vec::new();
enc_prefix(0, false, 0, &mut buf); encode_prefixed_int(1, 3, 0b0010_0000, &mut buf);
buf.push(b'a'); encode_prefixed_int(u64::MAX - 1, 7, 0x00, &mut buf);
let err = decode_header_block(&mut decoder(), &buf).unwrap_err();
assert!(matches!(err, Error::BadResponse(_)), "got {err:?}");
}
#[test]
fn qpack_decompression_bomb_is_rejected() {
let mut buf = Vec::new();
enc_prefix(0, false, 0, &mut buf); let name = b"a";
let value = vec![b'x'; 1024];
for _ in 0..512 {
encode_prefixed_int(name.len() as u64, 3, 0b0010_0000, &mut buf);
buf.extend_from_slice(name);
encode_prefixed_int(value.len() as u64, 7, 0x00, &mut buf);
buf.extend_from_slice(&value);
}
let err = decode_header_block(&mut decoder(), &buf).unwrap_err();
match err {
Error::BadResponse(m) => assert!(m.contains("header list"), "msg: {m}"),
other => panic!("expected header-list-cap error, got {other:?}"),
}
}
#[test]
fn qpack_block_references_dynamic_table_predicate() {
let mut zero = Vec::new();
enc_prefix(0, false, 0, &mut zero);
assert!(!block_references_dynamic_table(&zero));
let mut nonzero = Vec::new();
enc_prefix(2, false, 0, &mut nonzero);
assert!(block_references_dynamic_table(&nonzero));
}
#[test]
fn qpack_section_ack_encoding() {
let mut out = Vec::new();
encode_section_ack(0, &mut out);
assert_eq!(out, vec![0x80]);
let mut out = Vec::new();
encode_section_ack(4, &mut out);
assert_eq!(out, vec![0x84]);
}
#[test]
fn qpack_rfc9204_appendix_b2_cross_check() {
let mut wire: Vec<u8> = vec![0x3f, 0xbd, 0x01];
wire.extend_from_slice(&[0xc0, 0x0f]);
wire.extend_from_slice(b"www.example.com");
wire.extend_from_slice(&[0xc1, 0x0c]);
wire.extend_from_slice(b"/sample/path");
assert_eq!(
complete_encoder_instructions_len(&wire),
wire.len(),
"framing consumes the entire Appendix B.2 stream"
);
let mut dec = decoder();
dec.feed_encoder_stream(&wire).expect("feed encoder stream");
assert_eq!(dec.insert_count(), 2);
let block: [u8; 4] = [0x03, 0x81, 0x10, 0x11];
let fields = decode_header_block(&mut dec, &block).expect("decode block");
assert_eq!(
fields,
vec![
(":authority".to_string(), "www.example.com".to_string()),
(":path".to_string(), "/sample/path".to_string()),
]
);
assert!(block_references_dynamic_table(&block));
}
#[test]
fn qpack_decode_unsatisfiable_required_insert_count_errors() {
let mut dec = decoder(); let mut block = Vec::new();
enc_prefix(2, false, 0, &mut block);
encode_prefixed_int(0, 6, 0b1000_0000, &mut block); let err = decode_header_block(&mut dec, &block).unwrap_err();
assert!(matches!(err, Error::BadResponse(_)), "got {err:?}");
}
#[test]
fn qpack_decode_dynamic_reference_bomb_trips_list_cap() {
let mut dec = decoder();
let mut enc = Vec::new();
enc_set_capacity(QPACK_MAX_TABLE_CAPACITY, &mut enc);
let big = "x".repeat(3000);
enc_insert_literal("a", &big, &mut enc); dec.feed_encoder_stream(&enc).expect("inserts");
assert_eq!(dec.insert_count(), 1);
let mut block = Vec::new();
enc_prefix(2, false, 0, &mut block); for _ in 0..100 {
encode_prefixed_int(0, 6, 0b1000_0000, &mut block);
}
let err = decode_header_block(&mut dec, &block).unwrap_err();
match err {
Error::BadResponse(m) => assert!(m.contains("header list"), "msg: {m}"),
other => panic!("expected header-list-cap error, got {other:?}"),
}
}
#[test]
fn qpack_encoder_stream_partial_instruction_is_held() {
let mut full = Vec::new();
enc_set_capacity(QPACK_MAX_TABLE_CAPACITY, &mut full);
enc_insert_literal("name", "value", &mut full);
let truncated = &full[..full.len() - 1];
let complete = complete_encoder_instructions_len(truncated);
assert!(complete > 0 && complete < truncated.len());
let mut dec = decoder();
dec.feed_encoder_stream(&truncated[..complete])
.expect("feed complete prefix");
assert_eq!(dec.insert_count(), 0, "no insert applied from a partial");
assert_eq!(complete_encoder_instructions_len(&full), full.len());
let mut dec = decoder();
dec.feed_encoder_stream(&full).expect("feed full");
assert_eq!(dec.insert_count(), 1);
}
#[test]
fn send_rejects_non_https() {
let req = Request::get("http://example.com/").unwrap();
let err = send(req, &mut std::io::sink()).unwrap_err();
match err {
Error::UnsupportedScheme(_) => {}
other => panic!("expected UnsupportedScheme, got {other:?}"),
}
}
#[test]
fn send_rejects_malformed_pinned_pubkey() {
let req = Request::get("https://example.com/")
.unwrap()
.pinned_pubkey("not-a-valid-pin-spec");
let err = send(req, &mut std::io::sink()).unwrap_err();
assert!(
!matches!(err, Error::Io(_)),
"expected a pin-parse error before any network I/O, got {err:?}"
);
}
#[test]
fn oversized_headers_frame_len_is_rejected() {
let mut buf = Vec::new();
Frame::encode_header(frame_type::HEADERS, MAX_HEADERS_FRAME_LEN + 1, &mut buf);
let mut headers = None;
let mut body = Vec::new();
let mut dec = decoder();
assert!(matches!(
try_consume_frame(&buf, &mut headers, &mut body, &mut dec, None, &mut 0),
FrameOutcome::Err(Error::BadResponse(_))
));
}
#[test]
fn data_frame_len_past_budget_is_rejected() {
let mut buf = Vec::new();
Frame::encode_header(frame_type::DATA, (MAX_RESPONSE_BYTES + 1) as u64, &mut buf);
let mut headers = None;
let mut body = Vec::new();
let mut dec = decoder();
assert!(matches!(
try_consume_frame(&buf, &mut headers, &mut body, &mut dec, None, &mut 0),
FrameOutcome::Err(Error::BadResponse(_))
));
}
#[test]
fn data_frame_streams_to_sink_when_present() {
let payload = b"h3-streamed-body";
let mut buf = Vec::new();
Frame::encode_header(frame_type::DATA, payload.len() as u64, &mut buf);
buf.extend_from_slice(payload);
let mut headers = None;
let mut body = Vec::new();
let mut dec = decoder();
let mut sink: Vec<u8> = Vec::new();
let mut streamed: u64 = 0;
let outcome = try_consume_frame(
&buf,
&mut headers,
&mut body,
&mut dec,
Some(&mut sink),
&mut streamed,
);
assert!(
matches!(outcome, FrameOutcome::Consumed(_, _)),
"expected the DATA frame to be consumed"
);
assert_eq!(sink, payload);
assert!(body.is_empty(), "streamed body must not be buffered");
assert_eq!(streamed, payload.len() as u64);
}
#[test]
fn grease_frame_len_exceeding_usize_is_rejected() {
#[cfg(target_pointer_width = "32")]
{
let mut buf = Vec::new();
Frame::encode_header(0x21, 0x1_0000_0001, &mut buf);
let mut headers = None;
let mut body = Vec::new();
let mut dec = decoder();
assert!(matches!(
try_consume_frame(&buf, &mut headers, &mut body, &mut dec, None, &mut 0),
FrameOutcome::Err(Error::BadResponse(_))
));
}
}
#[test]
fn grease_frame_len_within_usize_needs_full_buffer() {
let mut buf = Vec::new();
Frame::encode_header(0x21, 4096, &mut buf);
let mut headers = None;
let mut body = Vec::new();
let mut dec = decoder();
assert!(matches!(
try_consume_frame(&buf, &mut headers, &mut body, &mut dec, None, &mut 0),
FrameOutcome::NeedMore
));
}
fn headers_frame(fields: &[(&str, &str)]) -> Vec<u8> {
let owned: Vec<(String, String)> = fields
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect();
let block = encode_header_block(&owned);
let mut buf = Vec::new();
Frame::encode_header(frame_type::HEADERS, block.len() as u64, &mut buf);
buf.extend_from_slice(&block);
buf
}
#[test]
fn control_frame_on_request_stream_is_rejected() {
for ty in [
frame_type::SETTINGS,
frame_type::GOAWAY,
frame_type::CANCEL_PUSH,
frame_type::MAX_PUSH_ID,
frame_type::PUSH_PROMISE,
] {
let mut buf = Vec::new();
Frame::encode_header(ty, 8, &mut buf); let mut headers = None;
let mut body = Vec::new();
let mut dec = decoder();
assert!(
matches!(
try_consume_frame(&buf, &mut headers, &mut body, &mut dec, None, &mut 0),
FrameOutcome::Err(Error::BadResponse(_))
),
"frame type {ty:#x} must be rejected on a request stream"
);
}
}
#[test]
fn interim_1xx_headers_is_skipped_then_final_used() {
let interim = headers_frame(&[(":status", "103")]);
let final_block = headers_frame(&[(":status", "200"), ("content-type", "text/plain")]);
let mut headers = None;
let mut body = Vec::new();
let mut dec = decoder();
assert!(matches!(
try_consume_frame(&interim, &mut headers, &mut body, &mut dec, None, &mut 0),
FrameOutcome::Consumed(_, _)
));
assert!(
headers.is_none(),
"interim 1xx must not be stored as the head"
);
assert!(matches!(
try_consume_frame(
&final_block,
&mut headers,
&mut body,
&mut dec,
None,
&mut 0
),
FrameOutcome::Consumed(_, _)
));
let fields = headers.expect("final HEADERS should be stored");
assert_eq!(header_status(&fields), Some(200));
}
#[test]
fn trailers_after_final_headers_are_discarded() {
let final_block = headers_frame(&[(":status", "200")]);
let trailers = headers_frame(&[("x-trailer", "v")]);
let mut headers = None;
let mut body = Vec::new();
let mut dec = decoder();
let _ = try_consume_frame(
&final_block,
&mut headers,
&mut body,
&mut dec,
None,
&mut 0,
);
assert_eq!(header_status(headers.as_ref().unwrap()), Some(200));
assert!(matches!(
try_consume_frame(&trailers, &mut headers, &mut body, &mut dec, None, &mut 0),
FrameOutcome::Consumed(_, _)
));
let fields = headers.unwrap();
assert_eq!(header_status(&fields), Some(200));
assert!(
!fields.iter().any(|(k, _)| k == "x-trailer"),
"trailers must not be merged into the head"
);
}
}