nexora_proto/http2/
connection.rs

1//! Sans-IO HTTP/2 connection and multiplexed stream lifecycle.
2use std::collections::{BTreeMap, VecDeque};
3
4use super::{
5    HpackDecoder, HpackEncoder, Http2Frame, Http2Settings, connection_error, explicit_push_enabled,
6    frame::{self, CLIENT_PREFACE, DEFAULT_INITIAL_WINDOW_SIZE, MAX_FLOW_CONTROL_WINDOW},
7    stream_error, validate_headers, validate_settings,
8};
9use crate::{
10    Bytes, BytesMut, Headers, Http2ErrorCode, Method, ProtoError, Request, Response, Result,
11    StatusCode, Uri, Version,
12};
13
14/// Endpoint role; server push is intentionally disabled.
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub enum Http2Role {
17    Client,
18    Server,
19}
20/// Local receive limits and advertised connection settings.
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub struct Http2Config {
23    pub local_settings: Http2Settings,
24    pub max_body_size: usize,
25}
26#[rustfmt::skip]
27impl Default for Http2Config {
28    fn default() -> Self {
29        Self { local_settings: Http2Settings { enable_push: false, ..Http2Settings::default() }, max_body_size: 16 * 1024 * 1024 }
30    }
31}
32/// Completed messages and connection or stream lifecycle notifications.
33#[derive(Debug, Clone, PartialEq, Eq)]
34#[rustfmt::skip]
35pub enum Http2Event {
36    RequestReceived { stream_id: u32, request: Request },
37    ResponseReceived { stream_id: u32, response: Response },
38    StreamReset { stream_id: u32, error_code: Http2ErrorCode },
39    StreamBodyTooLarge { stream_id: u32, size: usize, max: usize },
40    SettingsAcknowledged,
41    PingAcknowledged([u8; 8]),
42    GoAwayReceived { last_stream_id: u32, error_code: Http2ErrorCode, debug_data: Bytes },
43}
44struct Stream {
45    send_window: i64,
46    recv_window: i64,
47    outgoing: Bytes,
48    offset: usize,
49    local_closed: bool,
50    remote_closed: bool,
51    send_started: bool,
52    headers: Option<Vec<(String, String)>>,
53    body: BytesMut,
54    head_request: bool,
55}
56#[rustfmt::skip]
57impl Stream {
58    fn new(send: i64, recv: i64) -> Self {
59        Self { send_window: send, recv_window: recv, outgoing: Bytes::new(), offset: 0,
60            local_closed: false, remote_closed: false, send_started: false,
61            headers: None, body: BytesMut::new(), head_request: false }
62    }
63}
64struct HeaderBlock {
65    stream_id: u32,
66    end_stream: bool,
67    bytes: BytesMut,
68    discard: bool,
69}
70/// A transport-independent, single-threaded HTTP/2 multiplexer.
71pub struct Http2Connection {
72    role: Http2Role,
73    config: Http2Config,
74    peer: Http2Settings,
75    streams: BTreeMap<u32, Stream>,
76    seen_peer_streams: BTreeMap<u32, u32>,
77    events: VecDeque<Http2Event>,
78    tx: BytesMut,
79    encoder: HpackEncoder,
80    decoder: HpackDecoder,
81    preface: usize,
82    received_settings: bool,
83    awaiting_ack: bool,
84    next_stream: u32,
85    last_peer_stream: u32,
86    processed_peer_stream: u32,
87    send_window: i64,
88    recv_window: i64,
89    block: Option<HeaderBlock>,
90    sent_goaway: Option<u32>,
91    received_goaway: Option<u32>,
92    failure: Option<ProtoError>,
93}
94#[rustfmt::skip]
95impl Http2Connection {
96    /// Creates a connection and queues its initial SETTINGS and client preface.
97    /// # Errors
98    /// This constructor does not return errors; invalid settings fail on protocol operations.
99    /// # Panics
100    /// Does not panic.
101    pub fn new(role: Http2Role, mut config: Http2Config) -> Self {
102        config.local_settings.enable_push = false;
103        let mut tx = BytesMut::new();
104        if role == Http2Role::Client {
105            tx.extend_from_slice(CLIENT_PREFACE);
106        }
107        let failure =
108            validate_settings(config.local_settings).and_then(|()| frame::encode_frame(&Http2Frame::Settings { ack: false, settings: config.local_settings }, &mut tx)).err();
109        Self {
110            role,
111            config,
112            peer: Http2Settings::default(),
113            streams: BTreeMap::new(),
114            seen_peer_streams: BTreeMap::new(),
115            events: VecDeque::new(),
116            tx,
117            encoder: HpackEncoder::new(4096),
118            decoder: HpackDecoder::new(config.local_settings.header_table_size.max(4096) as usize, config.local_settings.max_header_list_size as usize),
119            preface: if role == Http2Role::Client { 24 } else { 0 },
120            received_settings: false,
121            awaiting_ack: true,
122            next_stream: 1,
123            last_peer_stream: 0,
124            processed_peer_stream: 0,
125            send_window: DEFAULT_INITIAL_WINDOW_SIZE,
126            recv_window: DEFAULT_INITIAL_WINDOW_SIZE,
127            block: None,
128            sent_goaway: None,
129            received_goaway: None,
130            failure,
131        }
132    }
133    /// Creates a client with default limits.
134    /// # Errors
135    /// This operation is infallible.
136    /// # Panics
137    /// Does not panic.
138    pub fn client_default() -> Self { Self::new(Http2Role::Client, Http2Config::default()) }
139    /// Creates a server with default limits.
140    /// # Errors
141    /// This operation is infallible.
142    /// # Panics
143    /// Does not panic.
144    pub fn server_default() -> Self { Self::new(Http2Role::Server, Http2Config::default()) }
145    /// Returns the local endpoint role.
146    /// # Errors
147    /// This operation is infallible.
148    /// # Panics
149    /// Does not panic.
150    pub fn role(&self) -> Http2Role { self.role }
151    /// Consumes complete frames, retaining incomplete frames in `src`.
152    /// # Errors
153    /// Returns protocol, compression, stream, or flow-control errors for invalid input.
154    /// # Panics
155    /// Does not panic.
156    pub fn feed_bytes(&mut self, src: &mut BytesMut) -> Result<()> {
157        self.check_live()?;
158        let result = self.feed_inner(src);
159        if let Err(error) = &result {
160            match error {
161                ProtoError::Http2ConnectionError { code, .. } => {
162                    let _queued = self.queue(Http2Frame::GoAway { last_stream_id: self.last_peer_stream, error_code: *code, debug_data: Bytes::new() });
163                    self.failure = Some(error.clone());
164                }
165                ProtoError::Http2StreamError { stream_id, code, .. } => {
166                    self.streams.remove(stream_id);
167                    let _queued = self.queue(Http2Frame::RstStream { stream_id: *stream_id, error_code: *code });
168                }
169                _ => {}
170            }
171        }
172        result
173    }
174    /// Removes the next completed-message or lifecycle notification.
175    /// # Errors
176    /// This operation is infallible.
177    /// # Panics
178    /// Does not panic.
179    pub fn poll_event(&mut self) -> Option<Http2Event> {
180        self.events.pop_front()
181    }
182    /// Appends all currently queued wire bytes to `out`.
183    /// # Errors
184    /// This operation is infallible.
185    /// # Panics
186    /// Does not panic.
187    pub fn take_tx_bytes(&mut self, out: &mut BytesMut) {
188        out.extend_from_slice(self.tx.as_slice());
189        self.tx.clear();
190    }
191    /// Reports whether encoded wire bytes are available.
192    /// # Errors
193    /// This operation is infallible.
194    /// # Panics
195    /// Does not panic.
196    pub fn has_tx_bytes(&self) -> bool {
197        !self.tx.is_empty()
198    }
199    /// Opens an odd-numbered client stream and queues a request.
200    /// # Errors
201    /// Returns an error for the wrong role, draining connections, exhausted limits, or invalid headers.
202    /// # Panics
203    /// Does not panic.
204    pub fn submit_request(&mut self, req: &Request) -> Result<u32> {
205        self.check_live()?;
206        if !self.can_open_stream() {
207            return Err(connection_error(Http2ErrorCode::RefusedStream, "cannot open request stream"));
208        }
209        let id = self.next_stream;
210        let mut fields = vec![(":method".into(), req.method.as_str().into())];
211        if req.method != Method::Connect {
212            fields.push((":scheme".into(), req.uri.scheme().as_str().into()));
213            fields.push((":path".into(), req.uri.path_and_query().into()));
214        }
215        fields.push((":authority".into(), req.uri.authority()));
216        fields.extend(req.headers.iter().filter(|field| field.name != "host").map(|field| (field.name.clone(), field.value.clone())));
217        validate_headers(&fields, Http2Role::Server, false, id)?;
218        if req.headers.content_length()?.is_some_and(|size| size != req.body.len()) {
219            return Err(stream_error(id, Http2ErrorCode::ProtocolError, "request content length disagrees with body"));
220        }
221        if req.headers.get("host").is_some_and(|host| !host.eq_ignore_ascii_case(&req.uri.authority())) {
222            return Err(stream_error(id, Http2ErrorCode::ProtocolError, "host disagrees with request authority"));
223        }
224        self.encode_headers(id, &fields, req.body.is_empty())?;
225        let mut stream = self.new_stream();
226        stream.outgoing = req.body.clone();
227        stream.send_started = true;
228        stream.local_closed = req.body.is_empty();
229        stream.head_request = req.method == Method::Head;
230        self.streams.insert(id, stream);
231        self.next_stream += 2;
232        self.flush_data()?;
233        Ok(id)
234    }
235    /// Queues a final response on an existing server stream.
236    /// # Errors
237    /// Returns an error for closed streams, invalid roles, or invalid response headers.
238    /// # Panics
239    /// Does not panic.
240    pub fn submit_response(&mut self, stream_id: u32, resp: &Response) -> Result<()> {
241        self.check_live()?;
242        let stream = self.streams.get(&stream_id).ok_or_else(|| stream_error(stream_id, Http2ErrorCode::StreamClosed, "unknown response stream"))?;
243        if self.role != Http2Role::Server || stream.send_started || !stream.remote_closed {
244            return Err(stream_error(stream_id, Http2ErrorCode::ProtocolError, "response stream is not ready"));
245        }
246        let mut fields = vec![(":status".into(), resp.status.as_u16().to_string())];
247        fields.extend(resp.headers.iter().map(|field| (field.name.clone(), field.value.clone())));
248        validate_headers(&fields, Http2Role::Client, false, stream_id)?;
249        if resp.status.is_informational() {
250            return Err(stream_error(stream_id, Http2ErrorCode::ProtocolError, "final response required"));
251        }
252        let no_body = stream.head_request || matches!(resp.status.as_u16(), 204 | 304);
253        let length = resp.headers.content_length()?;
254        if (resp.status.as_u16() == 204 && length.is_some()) || (!no_body && length.is_some_and(|size| size != resp.body.len())) {
255            return Err(stream_error(stream_id, Http2ErrorCode::ProtocolError, "response content length disagrees with semantics"));
256        }
257        let body = if no_body { Bytes::new() } else { resp.body.clone() };
258        self.encode_headers(stream_id, &fields, body.is_empty())?;
259        if let Some(stream) = self.streams.get_mut(&stream_id) {
260            stream.local_closed = body.is_empty();
261            stream.outgoing = body;
262            stream.send_started = true;
263        }
264        self.flush_data()?;
265        self.retire();
266        Ok(())
267    }
268    /// Cancels one stream without closing sibling streams.
269    /// # Errors
270    /// Returns an error if the connection failed or the stream is not active.
271    /// # Panics
272    /// Does not panic.
273    pub fn cancel_stream(&mut self, stream_id: u32) -> Result<()> {
274        self.check_live()?;
275        if self.streams.remove(&stream_id).is_none() {
276            return Err(stream_error(stream_id, Http2ErrorCode::StreamClosed, "cannot cancel inactive stream"));
277        }
278        self.queue(Http2Frame::RstStream { stream_id, error_code: Http2ErrorCode::Cancel })
279    }
280    /// Queues a connection liveness probe.
281    /// # Errors
282    /// Returns an error if the connection failed.
283    /// # Panics
284    /// Does not panic.
285    pub fn send_ping(&mut self, opaque_data: [u8; 8]) -> Result<()> {
286        self.check_live()?;
287        self.queue(Http2Frame::Ping { ack: false, opaque_data })
288    }
289    /// Stops accepting new streams while existing accepted streams drain.
290    /// # Errors
291    /// Returns an error for an invalid cutoff or debug data exceeding the peer frame limit.
292    /// # Panics
293    /// Does not panic.
294    pub fn send_goaway(&mut self, last_stream_id: u32, code: Http2ErrorCode, debug_data: Bytes) -> Result<()> {
295        self.check_live()?;
296        if last_stream_id > 0x7fff_ffff || last_stream_id < self.processed_peer_stream || self.sent_goaway.is_some_and(|old| last_stream_id > old) {
297            return Err(connection_error(Http2ErrorCode::ProtocolError, "invalid GOAWAY stream identifier"));
298        }
299        if debug_data.len().saturating_add(8) > self.peer.max_frame_size as usize {
300            return Err(connection_error(Http2ErrorCode::FrameSizeError, "GOAWAY exceeds peer frame size"));
301        }
302        self.queue(Http2Frame::GoAway { last_stream_id, error_code: code, debug_data })?;
303        self.sent_goaway = Some(last_stream_id);
304        Ok(())
305    }
306    /// Counts streams that have not closed both directions.
307    /// # Errors
308    /// This operation is infallible.
309    /// # Panics
310    /// Does not panic.
311    pub fn active_stream_count(&self) -> usize { self.streams.len() }
312    /// Reports whether a new client request can be submitted.
313    /// # Errors
314    /// This operation is infallible.
315    /// # Panics
316    /// Does not panic.
317    pub fn can_open_stream(&self) -> bool {
318        self.role == Http2Role::Client
319            && self.failure.is_none()
320            && self.sent_goaway.is_none()
321            && self.received_goaway.is_none()
322            && self.next_stream <= 0x7fff_ffff
323            && self.streams.len() < self.peer.max_concurrent_streams as usize
324    }
325    /// Returns remaining connection-level send credit.
326    /// # Errors
327    /// This operation is infallible.
328    /// # Panics
329    /// Does not panic.
330    pub fn connection_send_window(&self) -> i64 { self.send_window }
331    /// Returns signed send credit for an active stream.
332    /// # Errors
333    /// This operation is infallible.
334    /// # Panics
335    /// Does not panic.
336    pub fn stream_send_window(&self, stream_id: u32) -> Option<i64> { self.streams.get(&stream_id).map(|s| s.send_window) }
337    fn check_live(&self) -> Result<()> {
338        match &self.failure {
339            Some(error) => Err(error.clone()),
340            None => Ok(()),
341        }
342    }
343    fn queue(&mut self, frame: Http2Frame) -> Result<()> { frame::encode_frame(&frame, &mut self.tx) }
344    fn new_stream(&self) -> Stream {
345        Stream::new(i64::from(self.peer.initial_window_size), i64::from(self.config.local_settings.initial_window_size))
346    }
347    fn retire(&mut self) { self.streams.retain(|_, stream| !(stream.local_closed && stream.remote_closed)); }
348    fn feed_inner(&mut self, src: &mut BytesMut) -> Result<()> {
349        while self.preface < CLIENT_PREFACE.len() && !src.is_empty() {
350            if src[0] != CLIENT_PREFACE[self.preface] {
351                return Err(connection_error(Http2ErrorCode::ProtocolError, "invalid client connection preface"));
352            }
353            src.advance(1);
354            self.preface += 1;
355        }
356        if self.preface < CLIENT_PREFACE.len() {
357            return Ok(());
358        }
359        loop {
360            if src.len() >= 9 {
361                let id = u32::from_be_bytes([src[5], src[6], src[7], src[8]]) & 0x7fff_ffff;
362                if !self.received_settings && (src[3] != 4 || src[4] & 1 != 0) {
363                    return Err(connection_error(Http2ErrorCode::ProtocolError, "first peer frame must be SETTINGS"));
364                }
365                if self.block.as_ref().is_some_and(|block| src[3] != 9 || id != block.stream_id) {
366                    return Err(connection_error(Http2ErrorCode::ProtocolError, "interleaved header block"));
367                }
368            }
369            let merged = merged_settings(src.as_slice(), self.peer, &self.streams)?;
370            if self.role == Http2Role::Client && explicit_push_enabled(src.as_slice()) {
371                return Err(connection_error(Http2ErrorCode::ProtocolError, "server enabled unsupported push"));
372            }
373            let flow_len = if src.len() >= 9 { (usize::from(src[0]) << 16) | (usize::from(src[1]) << 8) | usize::from(src[2]) } else { 0 };
374            let decoded = frame::decode_frame(src, self.config.local_settings.max_frame_size);
375            if matches!(decoded, Err(ProtoError::Http2StreamError { .. })) {
376                if src.len() < 9 + flow_len { break; }
377                if src[3] == 1 {
378                    return Err(connection_error(Http2ErrorCode::ProtocolError, "invalid prioritized header block"));
379                }
380                if src[3] != 2 && self.is_idle(u32::from_be_bytes([src[5], src[6], src[7], src[8]]) & 0x7fff_ffff) {
381                    return Err(connection_error(Http2ErrorCode::ProtocolError, "invalid frame on idle stream"));
382                }
383                src.advance(9 + flow_len);
384            }
385            let Some(mut frame) = decoded? else {
386                break;
387            };
388            if let (Some((settings, table_sizes)), Http2Frame::Settings { ack: false, .. }) = (merged, &frame) {
389                for size in table_sizes { self.encoder.set_max_table_size(size); }
390                frame = Http2Frame::Settings { ack: false, settings };
391            }
392            self.handle_frame(frame, flow_len)?;
393        }
394        Ok(())
395    }
396    fn handle_frame(&mut self, frame: Http2Frame, flow_len: usize) -> Result<()> {
397        match frame {
398            Http2Frame::Settings { ack: true, .. } => {
399                if !self.awaiting_ack {
400                    return Err(connection_error(Http2ErrorCode::ProtocolError, "unexpected SETTINGS acknowledgement"));
401                }
402                self.awaiting_ack = false;
403                self.decoder.set_max_table_size(self.config.local_settings.header_table_size as usize);
404                self.events.push_back(Http2Event::SettingsAcknowledged);
405            }
406            Http2Frame::Settings { ack: false, settings } => {
407                validate_settings(settings)?;
408                let delta = i64::from(settings.initial_window_size) - i64::from(self.peer.initial_window_size);
409                if self.streams.values().any(|stream| stream.send_window + delta > MAX_FLOW_CONTROL_WINDOW) {
410                    return Err(connection_error(Http2ErrorCode::FlowControlError, "SETTINGS increases stream window beyond limit"));
411                }
412                for stream in self.streams.values_mut() {
413                    stream.send_window += delta;
414                }
415                self.encoder.set_max_table_size(settings.header_table_size as usize);
416                self.peer = settings;
417                self.received_settings = true;
418                self.queue(Http2Frame::Settings { ack: true, settings: self.config.local_settings })?;
419                self.flush_data()?;
420            }
421            Http2Frame::Headers { stream_id, end_stream, end_headers, header_block } => {
422                let discard = !self.accept_headers(stream_id)?;
423                let mut bytes = BytesMut::new();
424                bytes.extend_from_slice(header_block.as_slice());
425                self.block = Some(HeaderBlock { stream_id, end_stream, bytes, discard });
426                if end_headers {
427                    self.finish_headers()?;
428                }
429            }
430            Http2Frame::Continuation { stream_id, end_headers, header_block } => {
431                let block = self.block.as_mut().ok_or_else(|| connection_error(Http2ErrorCode::ProtocolError, "unexpected CONTINUATION"))?;
432                if block.stream_id != stream_id {
433                    return Err(connection_error(Http2ErrorCode::ProtocolError, "wrong CONTINUATION stream"));
434                }
435                block.bytes.extend_from_slice(header_block.as_slice());
436                if block.bytes.len() > (self.config.local_settings.max_header_list_size as usize).saturating_mul(2).saturating_add(1024) {
437                    return Err(connection_error(Http2ErrorCode::EnhanceYourCalm, "compressed header block too large"));
438                }
439                if end_headers {
440                    self.finish_headers()?;
441                }
442            }
443            Http2Frame::Data { stream_id, end_stream, data } => self.receive_data(stream_id, end_stream, data, flow_len)?,
444            Http2Frame::WindowUpdate { stream_id, increment } => self.window_update(stream_id, increment)?,
445            Http2Frame::RstStream { stream_id, error_code } => {
446                if self.is_idle(stream_id) {
447                    return Err(connection_error(Http2ErrorCode::ProtocolError, "RST_STREAM on idle stream"));
448                }
449                if self.streams.remove(&stream_id).is_some() {
450                    self.events.push_back(Http2Event::StreamReset { stream_id, error_code });
451                }
452            }
453            Http2Frame::Ping { ack, opaque_data } => {
454                if ack {
455                    self.events.push_back(Http2Event::PingAcknowledged(opaque_data));
456                } else {
457                    self.queue(Http2Frame::Ping { ack: true, opaque_data })?;
458                }
459            }
460            Http2Frame::GoAway { last_stream_id, error_code, debug_data } => {
461                if self.received_goaway.is_some_and(|old| last_stream_id > old) {
462                    return Err(connection_error(Http2ErrorCode::ProtocolError, "GOAWAY increases last stream identifier"));
463                }
464                self.received_goaway = Some(last_stream_id);
465                if self.role == Http2Role::Client {
466                    let refused: Vec<_> = self.streams.range((last_stream_id.saturating_add(1))..).map(|(&id, _)| id).collect();
467                    for stream_id in refused {
468                        self.streams.remove(&stream_id);
469                        self.events.push_back(Http2Event::StreamReset { stream_id, error_code: Http2ErrorCode::RefusedStream });
470                    }
471                }
472                self.events.push_back(Http2Event::GoAwayReceived { last_stream_id, error_code, debug_data });
473            }
474            Http2Frame::Priority { .. } => {}
475            Http2Frame::Unknown { header, .. } => {
476                if matches!(header.frame_type, super::Http2FrameType::PushPromise) {
477                    return Err(connection_error(Http2ErrorCode::ProtocolError, "server push is disabled"));
478                }
479            }
480        }
481        self.retire();
482        Ok(())
483    }
484    fn was_peer_stream(&self, id: u32) -> bool {
485        self.seen_peer_streams.range(..=id).next_back().is_some_and(|(_, &end)| id <= end)
486    }
487    fn record_peer_stream(&mut self, id: u32) {
488        match self.seen_peer_streams.last_entry() {
489            Some(mut entry) if entry.get().saturating_add(2) == id => { *entry.get_mut() = id; }
490            _ => { self.seen_peer_streams.insert(id, id); }
491        }
492    }
493    fn is_idle(&self, id: u32) -> bool {
494        id == 0 || id.is_multiple_of(2) || if self.role == Http2Role::Client { id >= self.next_stream } else { id > self.last_peer_stream }
495    }
496    fn accept_headers(&mut self, id: u32) -> Result<bool> {
497        if let Some(stream) = self.streams.get(&id) {
498            if stream.remote_closed {
499                self.streams.remove(&id);
500                self.queue(Http2Frame::RstStream { stream_id: id, error_code: Http2ErrorCode::StreamClosed })?;
501                return Ok(false);
502            }
503            return Ok(true);
504        }
505        if id == 0 || id.is_multiple_of(2) || self.is_idle(id) && self.role == Http2Role::Client {
506            return Err(connection_error(Http2ErrorCode::ProtocolError, "invalid new stream identifier"));
507        }
508        if !self.is_idle(id) {
509            if self.role == Http2Role::Server && !self.was_peer_stream(id) {
510                return Err(connection_error(Http2ErrorCode::ProtocolError, "new stream identifier is not increasing"));
511            }
512            return Ok(false);
513        }
514        self.last_peer_stream = id;
515        self.record_peer_stream(id);
516        if self.sent_goaway.is_some_and(|last| id > last) || self.streams.len() >= self.config.local_settings.max_concurrent_streams as usize {
517            self.queue(Http2Frame::RstStream { stream_id: id, error_code: Http2ErrorCode::RefusedStream })?;
518            return Ok(false);
519        }
520        self.streams.insert(id, self.new_stream());
521        Ok(true)
522    }
523    fn finish_headers(&mut self) -> Result<()> {
524        let block = self.block.take().ok_or_else(|| connection_error(Http2ErrorCode::ProtocolError, "missing header block"))?;
525        let fields = match self.decoder.decode(block.bytes.as_slice()) {
526            Ok(fields) => fields,
527            Err(ProtoError::HeaderListTooLarge { .. }) if block.discard => return Ok(()),
528            Err(ProtoError::HeaderListTooLarge { .. }) => return Err(stream_error(block.stream_id, Http2ErrorCode::EnhanceYourCalm, "header list too large")),
529            Err(error) => return Err(connection_error(Http2ErrorCode::CompressionError, &error.to_string())),
530        };
531        if block.discard {
532            return Ok(());
533        }
534        let stream = self.streams.get_mut(&block.stream_id).ok_or_else(|| stream_error(block.stream_id, Http2ErrorCode::StreamClosed, "headers on closed stream"))?;
535        let trailers = stream.headers.is_some();
536        validate_headers(&fields, self.role, trailers, block.stream_id)?;
537        if trailers && !block.end_stream {
538            return Err(stream_error(block.stream_id, Http2ErrorCode::ProtocolError, "trailers must end stream"));
539        }
540        if !trailers && self.role == Http2Role::Client && fields.iter().any(|(name, value)| name == ":status" && value.starts_with('1')) {
541            if block.end_stream {
542                return Err(stream_error(block.stream_id, Http2ErrorCode::ProtocolError, "informational response ends stream"));
543            }
544            return Ok(());
545        }
546        if let Some(headers) = &mut stream.headers {
547            headers.extend(fields);
548        } else {
549            stream.headers = Some(fields);
550        }
551        if block.end_stream {
552            self.complete_message(block.stream_id)?;
553        }
554        Ok(())
555    }
556    fn receive_data(&mut self, id: u32, end: bool, data: Bytes, flow_len: usize) -> Result<()> {
557        let credit = flow_len as i64;
558        if credit > self.recv_window {
559            return Err(connection_error(Http2ErrorCode::FlowControlError, "connection receive window exhausted"));
560        }
561        self.recv_window -= credit;
562        if credit != 0 {
563            self.queue(Http2Frame::WindowUpdate { stream_id: 0, increment: flow_len as u32 })?;
564            self.recv_window += credit;
565        }
566        let Some(stream) = self.streams.get_mut(&id) else {
567            if self.is_idle(id) {
568                return Err(connection_error(Http2ErrorCode::ProtocolError, "DATA on idle stream"));
569            }
570            return Ok(());
571        };
572        if stream.remote_closed || stream.headers.is_none() {
573            return Err(stream_error(id, Http2ErrorCode::StreamClosed, "DATA outside open message"));
574        }
575        if credit > stream.recv_window {
576            return Err(stream_error(id, Http2ErrorCode::FlowControlError, "stream receive window exhausted"));
577        }
578        stream.recv_window -= credit;
579        let size = stream.body.len().saturating_add(data.len());
580        if size > self.config.max_body_size {
581            self.events.push_back(Http2Event::StreamBodyTooLarge { stream_id: id, size, max: self.config.max_body_size });
582            self.streams.remove(&id);
583            return self.queue(Http2Frame::RstStream { stream_id: id, error_code: Http2ErrorCode::Cancel });
584        }
585        stream.body.extend_from_slice(data.as_slice());
586        stream.recv_window += credit;
587        if credit != 0 {
588            self.queue(Http2Frame::WindowUpdate { stream_id: id, increment: flow_len as u32 })?;
589        }
590        if end {
591            self.complete_message(id)?;
592        }
593        Ok(())
594    }
595    fn complete_message(&mut self, id: u32) -> Result<()> {
596        let stream = self.streams.get_mut(&id).ok_or_else(|| stream_error(id, Http2ErrorCode::StreamClosed, "message stream closed"))?;
597        let fields = stream.headers.as_ref().ok_or_else(|| stream_error(id, Http2ErrorCode::ProtocolError, "message has no headers"))?;
598        let mut headers = Headers::new();
599        for (name, value) in fields {
600            if !name.starts_with(':') {
601                headers.append(name.clone(), value.clone());
602            }
603        }
604        let pseudo = |key: &str| fields.iter().find(|(name, _)| name == key).map(|(_, value)| value.as_str());
605        let length = headers.content_length().map_err(|_| stream_error(id, Http2ErrorCode::ProtocolError, "invalid content length"))?;
606        let no_body = self.role == Http2Role::Client && (stream.head_request || matches!(pseudo(":status"), Some("204" | "304")));
607        if self.role == Http2Role::Client && pseudo(":status") == Some("204") && length.is_some() {
608            return Err(stream_error(id, Http2ErrorCode::ProtocolError, "204 response includes content length"));
609        }
610        if (!no_body && length.is_some_and(|length| length != stream.body.len())) || (no_body && !stream.body.is_empty()) {
611            return Err(stream_error(id, Http2ErrorCode::ProtocolError, "message body disagrees with content length or semantics"));
612        }
613        let body = Bytes::new();
614        let mut event = if self.role == Http2Role::Server {
615            let method = Method::parse(pseudo(":method").unwrap_or_default()).map_err(|_| stream_error(id, Http2ErrorCode::ProtocolError, "invalid request method"))?;
616            let authority = pseudo(":authority").or_else(|| headers.get("host")).unwrap_or_default();
617            let scheme = pseudo(":scheme").unwrap_or("http");
618            let path = pseudo(":path").unwrap_or("/");
619            let uri = Uri::parse(&format!("{scheme}://{authority}{path}")).map_err(|_| stream_error(id, Http2ErrorCode::ProtocolError, "invalid request URI"))?;
620            stream.head_request = method == Method::Head;
621            self.processed_peer_stream = self.processed_peer_stream.max(id);
622            Http2Event::RequestReceived { stream_id: id, request: Request { method, uri, version: Version::Http2, headers, body } }
623        } else {
624            let status = pseudo(":status").and_then(|value| value.parse::<u16>().ok()).ok_or_else(|| stream_error(id, Http2ErrorCode::ProtocolError, "missing response status"))?;
625            Http2Event::ResponseReceived {
626                stream_id: id,
627                response: Response {
628                    status: StatusCode::new(status).map_err(|_| stream_error(id, Http2ErrorCode::ProtocolError, "invalid response status"))?,
629                    version: Version::Http2,
630                    headers,
631                    body,
632                },
633            }
634        };
635        let body = std::mem::take(&mut stream.body).freeze();
636        match &mut event {
637            Http2Event::RequestReceived { request, .. } => request.body = body,
638            Http2Event::ResponseReceived { response, .. } => response.body = body,
639            _ => {}
640        }
641        stream.remote_closed = true;
642        self.events.push_back(event);
643        Ok(())
644    }
645    fn window_update(&mut self, id: u32, increment: u32) -> Result<()> {
646        if increment == 0 || increment > 0x7fff_ffff {
647            return Err(if id == 0 {
648                connection_error(Http2ErrorCode::ProtocolError, "invalid window increment")
649            } else {
650                stream_error(id, Http2ErrorCode::ProtocolError, "invalid window increment")
651            });
652        }
653        if id == 0 {
654            if self.send_window + i64::from(increment) > MAX_FLOW_CONTROL_WINDOW {
655                return Err(connection_error(Http2ErrorCode::FlowControlError, "connection window overflow"));
656            }
657            self.send_window += i64::from(increment);
658        } else if let Some(stream) = self.streams.get_mut(&id) {
659            if stream.send_window + i64::from(increment) > MAX_FLOW_CONTROL_WINDOW {
660                return Err(stream_error(id, Http2ErrorCode::FlowControlError, "stream window overflow"));
661            }
662            stream.send_window += i64::from(increment);
663        } else if self.is_idle(id) {
664            return Err(connection_error(Http2ErrorCode::ProtocolError, "WINDOW_UPDATE on idle stream"));
665        }
666        self.flush_data()
667    }
668    fn encode_headers(&mut self, id: u32, fields: &[(String, String)], end_stream: bool) -> Result<()> {
669        let size = fields.iter().fold(0usize, |size, (name, value)| size.saturating_add(name.len()).saturating_add(value.len()).saturating_add(32));
670        if size > self.peer.max_header_list_size as usize {
671            return Err(ProtoError::HeaderListTooLarge { size, max: self.peer.max_header_list_size as usize });
672        }
673        let mut block = BytesMut::new();
674        self.encoder.encode(fields, &mut block);
675        let bytes = block.freeze();
676        let max = self.peer.max_frame_size as usize;
677        if bytes.is_empty() {
678            return self.queue(Http2Frame::Headers { stream_id: id, end_stream, end_headers: true, header_block: bytes });
679        }
680        let mut offset = 0;
681        while offset < bytes.len() {
682            let end = (offset + max).min(bytes.len());
683            let header_block = bytes.slice(offset..end);
684            let frame = if offset == 0 {
685                Http2Frame::Headers { stream_id: id, end_stream, end_headers: end == bytes.len(), header_block }
686            } else {
687                Http2Frame::Continuation { stream_id: id, end_headers: end == bytes.len(), header_block }
688            };
689            self.queue(frame)?;
690            offset = end;
691        }
692        Ok(())
693    }
694    fn flush_data(&mut self) -> Result<()> {
695        loop {
696            let mut progress = false;
697            for (&id, stream) in &mut self.streams {
698                if stream.local_closed || !stream.send_started || stream.offset == stream.outgoing.len() {
699                    continue;
700                }
701                let credit = self.send_window.min(stream.send_window).min(i64::from(self.peer.max_frame_size));
702                if credit <= 0 {
703                    continue;
704                }
705                let size = (credit as usize).min(stream.outgoing.len() - stream.offset);
706                let end = stream.offset + size;
707                let end_stream = end == stream.outgoing.len();
708                frame::encode_frame(&Http2Frame::Data { stream_id: id, end_stream, data: stream.outgoing.slice(stream.offset..end) }, &mut self.tx)?;
709                stream.offset = end;
710                stream.send_window -= size as i64;
711                self.send_window -= size as i64;
712                stream.local_closed = end_stream;
713                progress = true;
714                if end_stream {
715                    stream.outgoing.clear();
716                    stream.offset = 0;
717                }
718            }
719            if !progress {
720                break;
721            }
722        }
723        self.retire();
724        Ok(())
725    }
726}
727#[rustfmt::skip]
728fn merged_settings(bytes: &[u8], mut settings: Http2Settings, streams: &BTreeMap<u32, Stream>) -> Result<Option<(Http2Settings, Vec<usize>)>> {
729    let initial_window = i64::from(settings.initial_window_size);
730    let mut table_sizes = Vec::new();
731    if bytes.len() < 9 || bytes[3] != 4 || bytes[4] & 1 != 0 {
732        return Ok(None);
733    }
734    let length = (usize::from(bytes[0]) << 16) | (usize::from(bytes[1]) << 8) | usize::from(bytes[2]);
735    if bytes.len() < 9 + length || !length.is_multiple_of(6) {
736        return Ok(None);
737    }
738    for item in bytes[9..9 + length].as_chunks::<6>().0 {
739        let id = u16::from_be_bytes([item[0], item[1]]);
740        let value = u32::from_be_bytes([item[2], item[3], item[4], item[5]]);
741        match id {
742            1 => { settings.header_table_size = value; table_sizes.push(value as usize); },
743            2 => settings.enable_push = value != 0,
744            3 => settings.max_concurrent_streams = value,
745            4 => {
746                if i64::from(value) > MAX_FLOW_CONTROL_WINDOW || streams.values().any(|stream| stream.send_window + i64::from(value) - initial_window > MAX_FLOW_CONTROL_WINDOW) {
747                    return Err(connection_error(Http2ErrorCode::FlowControlError, "SETTINGS occurrence overflows stream window"));
748                }
749                settings.initial_window_size = value;
750            },
751            5 => settings.max_frame_size = value,
752            6 => settings.max_header_list_size = value,
753            _ => {}
754        }
755    }
756    Ok(Some((settings, table_sizes)))
757}
nexora_proto/http2/connection.rs

nexora_proto/http2/
connection.rs
