h3/server/
request.rs

1use std::{convert::TryFrom, sync::Arc};
2
3use bytes::Buf;
4use http::{Request, StatusCode};
5
6use tokio::sync::mpsc::UnboundedSender;
7#[cfg(feature = "tracing")]
8use tracing::instrument;
9
10use crate::{
11    connection::{self},
12    error::{
13        connection_error_creators::{CloseStream, HandleFrameStreamErrorOnRequestStream},
14        internal_error::InternalConnectionError,
15        Code, StreamError,
16    },
17    frame::{FrameStream, FrameStreamError},
18    proto::{
19        frame::{Frame, PayloadLen},
20        headers::Header,
21    },
22    qpack,
23    quic::{self, SendStream, StreamId},
24    shared_state::{ConnectionState, SharedState},
25};
26
27use super::{connection::RequestEnd, stream::RequestStream};
28
29/// Helper struct to await the request headers and return a `Request` object
30pub struct RequestResolver<C, B>
31where
32    C: quic::Connection<B>,
33    C::BidiStream: quic::SendStream<B>,
34    B: Buf,
35{
36    #[doc(hidden)]
37    // TODO: make this private
38    pub frame_stream: FrameStream<C::BidiStream, B>,
39    pub(super) request_end_send: UnboundedSender<StreamId>,
40    pub(super) send_grease_frame: bool,
41    pub(super) max_field_section_size: u64,
42    pub(super) shared: Arc<SharedState>,
43}
44
45impl<C, B> ConnectionState for RequestResolver<C, B>
46where
47    C: quic::Connection<B>,
48    B: Buf,
49{
50    fn shared_state(&self) -> &SharedState {
51        &self.shared
52    }
53}
54
55impl<C, B> CloseStream for RequestResolver<C, B>
56where
57    C: quic::Connection<B>,
58    B: Buf,
59{
60}
61
62impl<C, B> RequestResolver<C, B>
63where
64    C: quic::Connection<B>,
65    B: Buf,
66{
67    /// Returns a future to await the request headers and return a `Request` object
68    #[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
69    pub async fn resolve_request(
70        mut self,
71    ) -> Result<(Request<()>, RequestStream<C::BidiStream, B>), StreamError> {
72        let frame = std::future::poll_fn(|cx| self.frame_stream.poll_next(cx)).await;
73        let req = self.accept_with_frame(frame)?;
74        Ok(req.resolve().await?)
75    }
76
77    /// Accepts a http request where the first frame has already been read and decoded.
78    ///
79    /// This is needed as a bidirectional stream may be read as part of incoming webtransport
80    /// bi-streams. If it turns out that the stream is *not* a `WEBTRANSPORT_STREAM` the request
81    /// may still want to be handled and passed to the user.
82    #[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
83    pub fn accept_with_frame(
84        mut self,
85        frame: Result<Option<Frame<PayloadLen>>, FrameStreamError>,
86    ) -> Result<ResolvedRequest<C, B>, StreamError> {
87        let mut encoded = match frame {
88            Ok(Some(Frame::Headers(h))) => h,
89
90            //= https://www.rfc-editor.org/rfc/rfc9114#section-4.1
91            //# If a client-initiated
92            //# stream terminates without enough of the HTTP message to provide a
93            //# complete response, the server SHOULD abort its response stream with
94            //# the error code H3_REQUEST_INCOMPLETE.
95            Ok(None) => {
96                self.frame_stream.reset(Code::H3_REQUEST_INCOMPLETE.value());
97                return Err(StreamError::StreamError {
98                    code: Code::H3_REQUEST_INCOMPLETE,
99                    reason: "stream terminated without headers".to_string(),
100                });
101            }
102
103            //= https://www.rfc-editor.org/rfc/rfc9114#section-4.1
104            //# Receipt of an invalid sequence of frames MUST be treated as a
105            //# connection error of type H3_FRAME_UNEXPECTED.
106
107            //= https://www.rfc-editor.org/rfc/rfc9114#section-7.2.5
108            //# A server MUST treat the
109            //# receipt of a PUSH_PROMISE frame as a connection error of type
110            //# H3_FRAME_UNEXPECTED.
111            Ok(Some(_)) => {
112                //= https://www.rfc-editor.org/rfc/rfc9114#section-4.1
113                //# Receipt of an invalid sequence of frames MUST be treated as a
114                //# connection error of type H3_FRAME_UNEXPECTED.
115                // Close if the first frame is not a header frame
116                return Err(
117                    self.handle_connection_error_on_stream(InternalConnectionError::new(
118                        Code::H3_FRAME_UNEXPECTED,
119                        "first request frame is not headers".to_string(),
120                    )),
121                );
122            }
123            Err(e) => {
124                return Err(self.handle_frame_stream_error_on_request_stream(e));
125            }
126        };
127
128        let decoded = match qpack::decode_stateless(&mut encoded, self.max_field_section_size) {
129            //= https://www.rfc-editor.org/rfc/rfc9114#section-4.2.2
130            //# An HTTP/3 implementation MAY impose a limit on the maximum size of
131            //# the message header it will accept on an individual HTTP message.
132            Err(qpack::DecoderError::HeaderTooLong(cancel_size)) => Err(cancel_size),
133            Ok(decoded) => Ok(decoded),
134            Err(_e) => {
135                return Err(
136                    self.handle_connection_error_on_stream(InternalConnectionError {
137                        code: Code::QPACK_DECOMPRESSION_FAILED,
138                        message: "Failed to decode headers".to_string(),
139                    }),
140                );
141            }
142        };
143
144        let request_stream = RequestStream {
145            request_end: Arc::new(RequestEnd {
146                request_end: self.request_end_send.clone(),
147                stream_id: self.frame_stream.send_id(),
148            }),
149            inner: connection::RequestStream::new(
150                self.frame_stream,
151                self.max_field_section_size,
152                self.shared.clone(),
153                self.send_grease_frame,
154            ),
155        };
156
157        Ok(ResolvedRequest::new(
158            request_stream,
159            decoded,
160            self.max_field_section_size,
161        ))
162    }
163}
164
165pub struct ResolvedRequest<C, B>
166where
167    C: quic::Connection<B>,
168    B: Buf,
169{
170    request_stream: RequestStream<C::BidiStream, B>,
171    // Ok or `REQUEST_HEADER_FIELDS_TO_LARGE` which needs to be sent
172    decoded: Result<qpack::Decoded, u64>,
173    max_field_section_size: u64,
174}
175
176impl<B, C> ResolvedRequest<C, B>
177where
178    C: quic::Connection<B>,
179    B: Buf,
180{
181    pub fn new(
182        request_stream: RequestStream<C::BidiStream, B>,
183        decoded: Result<qpack::Decoded, u64>,
184        max_field_section_size: u64,
185    ) -> Self {
186        Self {
187            request_stream,
188            decoded,
189            max_field_section_size,
190        }
191    }
192
193    /// Finishes the resolution of the request
194    #[cfg_attr(feature = "tracing", instrument(skip_all, level = "trace"))]
195    pub async fn resolve(
196        mut self,
197    ) -> Result<(Request<()>, RequestStream<C::BidiStream, B>), StreamError> {
198        let fields = match self.decoded {
199            Ok(v) => v.fields,
200            Err(cancel_size) => {
201                // Send and await the error response
202                self.request_stream
203                    .send_response(
204                        http::Response::builder()
205                            .status(StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE)
206                            .body(())
207                            .expect("header too big response"),
208                    )
209                    .await?;
210
211                return Err(StreamError::HeaderTooBig {
212                    actual_size: cancel_size,
213                    max_size: self.max_field_section_size,
214                });
215            }
216        };
217
218        // Parse the request headers
219        let result = match Header::try_from(fields) {
220            Ok(header) => match header.into_request_parts() {
221                Ok(parts) => Ok(parts),
222                Err(err) => Err(err),
223            },
224            Err(err) => Err(err),
225        };
226        let (method, uri, protocol, headers) = match result {
227            Ok(parts) => parts,
228            Err(err) => {
229                //= https://www.rfc-editor.org/rfc/rfc9114#section-4.1.2
230                //# Malformed requests or responses that are
231                //# detected MUST be treated as a stream error of type H3_MESSAGE_ERROR.
232                let error_code = Code::H3_MESSAGE_ERROR;
233                self.request_stream.stop_stream(error_code);
234                self.request_stream.stop_sending(error_code);
235
236                return Err(StreamError::StreamError {
237                    code: error_code,
238                    reason: format!("Malformed request with error: {}", err),
239                });
240            }
241        };
242
243        //  request_stream.stop_stream(Code::H3_MESSAGE_ERROR).await;
244
245        let mut req = http::Request::new(());
246        *req.method_mut() = method;
247        *req.uri_mut() = uri;
248        *req.headers_mut() = headers;
249        // NOTE: insert `Protocol` and not `Option<Protocol>`
250        if let Some(protocol) = protocol {
251            req.extensions_mut().insert(protocol);
252        }
253        *req.version_mut() = http::Version::HTTP_3;
254        #[cfg(feature = "tracing")]
255        tracing::trace!("replying with: {:?}", req);
256
257        Ok((req, self.request_stream))
258    }
259}
h3/server/request.rs

h3/server/
request.rs
