ntex/http/h2/
service.rs

1use std::{cell::Cell, cell::RefCell, future::poll_fn, io, mem, rc::Rc};
2
3use ntex_h2::{self as h2, control::ExpectResult, frame::StreamId, server};
4use urly::{Authority, Scheme, Url};
5
6use crate::error::{Error, IntoFailure};
7use crate::http::body::{Body, BodySize, MessageBody, ResponseBody};
8use crate::http::config::DispatcherConfig;
9use crate::http::error::{DispatchError, H2Error, ResponseError};
10use crate::http::header::{self, HeaderMap, HeaderName, HeaderValue};
11use crate::http::helpers::is_valid_target;
12use crate::http::message::{CurrentIo, ResponseHead};
13use crate::http::{DateService, Method, Request, Response, StatusCode, Version};
14use crate::io::{Filter, Io, IoBoxed, IoRef, types};
15use crate::service::pipeline::{Pipeline, PipelineBinding, PipelineFactory};
16use crate::service::{Ctx, IntoServiceFactory, RequestState, Service, ServiceFactory};
17use crate::util::{Bytes, BytesMut, HashMap};
18
19use super::{DefaultControlService, payload::Payload, payload::PayloadSender};
20
21/// An HTTP/2 transport service.
22#[derive(derive_more::Debug)]
23#[debug("H2Service")]
24pub struct H2Service<F, Req: RequestState<Io<F>>, Err> {
25    sf: crate::http::HttpPipeline<Req::State, Err>,
26    ctl: crate::http::Ctl2Pipeline<Req::State>,
27    config: DispatcherConfig,
28}
29
30impl<F, Req, Err> H2Service<F, Req, Err>
31where
32    F: Filter,
33    Req: RequestState<Io<F>>,
34    Req::State: Clone,
35    Err: ResponseError + 'static,
36{
37    /// Create new `H2Service` instance.
38    pub(crate) fn new<Sf>(sf: impl IntoServiceFactory<Sf, Req::State, Request>) -> Self
39    where
40        Sf: ServiceFactory<Req::State, Request, Error = Err> + 'static,
41        Sf::Res: Into<Response>,
42        Sf::InitError: IntoFailure,
43    {
44        H2Service {
45            sf: PipelineFactory::new(
46                sf.into_factory()
47                    .map(Into::into)
48                    .map_init_err(|e| DispatchError::Control(e.fail())),
49            ),
50            ctl: PipelineFactory::new(DefaultControlService),
51            config: DispatcherConfig::default(),
52        }
53    }
54}
55
56impl<F, Req, Err> H2Service<F, Req, Err>
57where
58    F: Filter,
59    Req: RequestState<Io<F>>,
60    Req::State: Clone,
61    Err: ResponseError + 'static,
62{
63    #[must_use]
64    /// Provides the HTTP/2 control service.
65    pub fn control<I, Sf>(self, ctl: I) -> Self
66    where
67        I: IntoServiceFactory<Sf, Req::State, h2::Control<Error<H2Error>>>,
68        Sf: ServiceFactory<Req::State, h2::Control<Error<H2Error>>, Res = h2::ControlAck> + 'static,
69        Sf::Error: IntoFailure,
70        Sf::InitError: IntoFailure,
71    {
72        H2Service {
73            sf: self.sf,
74            ctl: PipelineFactory::new(
75                ctl.into_factory()
76                    .map_err(|e| DispatchError::Service(e.fail()))
77                    .map_init_err(|e| DispatchError::Service(e.fail())),
78            ),
79            config: self.config,
80        }
81    }
82}
83
84impl<St, F, Req, Err> Service<St, Req> for H2Service<F, Req, Err>
85where
86    F: Filter,
87    Req: RequestState<Io<F>>,
88    Req::State: Clone,
89    Err: ResponseError + 'static,
90{
91    type Res = ();
92    type Error = DispatchError;
93
94    async fn call(&self, req: Req, _: Ctx<'_, Self, St>) -> Result<Self::Res, Self::Error> {
95        let (st, io) = req.unpack();
96
97        let svc = self.sf.create(st.clone()).await?;
98        let ctl = self.ctl.create(st).await?;
99
100        let id = self.config.next_id();
101        let ioref = io.get_ref();
102        let (_guard, inflight) = self.config.insert_io(&ioref);
103        log::trace!(
104            "{}: New http2 connection {id}, peer address {:?}, inflight: {inflight}",
105            io.tag(),
106            io.query::<types::PeerAddr>().get()
107        );
108
109        handle(id, io.into(), svc, ctl).await
110    }
111
112    #[inline]
113    async fn ready(&self, _: Ctx<'_, Self, St>) -> Result<(), Self::Error> {
114        Ok(())
115    }
116
117    #[inline]
118    async fn shutdown(&self, _: Ctx<'_, Self, St>) {
119        // check inflight connections
120        let inflight = self.config.shutdown();
121        if inflight != 0 {
122            log::trace!("Shutting down service, in-flight connections: {inflight}");
123
124            self.config.wait_shutdown().await;
125            log::trace!("Shutting down is complected");
126        }
127    }
128}
129
130pub(in crate::http) async fn handle<Err>(
131    id: usize,
132    io: IoBoxed,
133    svc: Pipeline<Request, Response, Err>,
134    control: Pipeline<h2::Control<Error<H2Error>>, h2::ControlAck, DispatchError>,
135) -> Result<(), DispatchError>
136where
137    Err: ResponseError + 'static,
138{
139    let ioref = io.get_ref();
140    let control = Pipeline::new((), ControlService { inner: control });
141
142    let _ = server::handle_one(
143        io,
144        Pipeline::new((), PublishService::new(id, ioref, svc, control.bind())),
145        control,
146    )
147    .await;
148
149    Ok(())
150}
151
152/// Sets `GOAWAY` reason codes for connection level request errors.
153struct ControlService {
154    inner: Pipeline<h2::Control<Error<H2Error>>, h2::ControlAck, DispatchError>,
155}
156
157impl Service<(), h2::Control<Error<H2Error>>> for ControlService {
158    type Res = h2::ControlAck;
159    type Error = DispatchError;
160
161    async fn ready(&self, _: Ctx<'_, Self, ()>) -> Result<(), Self::Error> {
162        self.inner.ready().await
163    }
164
165    async fn shutdown(&self, _: Ctx<'_, Self, ()>) {
166        self.inner.shutdown().await;
167    }
168
169    async fn call(
170        &self,
171        msg: h2::Control<Error<H2Error>>,
172        _: Ctx<'_, Self, ()>,
173    ) -> Result<Self::Res, Self::Error> {
174        let msg = match msg {
175            h2::Control::Disconnect(h2::control::Reason::Error(err))
176                if matches!(**err.get_ref(), H2Error::EmptyDataFrames) =>
177            {
178                h2::Control::Disconnect(h2::control::Reason::Error(
179                    err.reason(h2::frame::Reason::ENHANCE_YOUR_CALM),
180                ))
181            }
182            msg => msg,
183        };
184        self.inner.call(msg).await
185    }
186}
187
188struct PublishService<Err> {
189    id: usize,
190    io: IoRef,
191    svc: Pipeline<Request, Response, Err>,
192    control: PipelineBinding<h2::Control<Error<H2Error>>, h2::ControlAck, DispatchError>,
193    streams: Rc<RefCell<HashMap<StreamId, StreamPayload>>>,
194    /// Consecutive empty non-final `DATA` frames
195    empty_data: Cell<u8>,
196}
197
198/// Maximum number of consecutive empty non-final `DATA` frames,
199/// such frames are not flow controlled
200const MAX_EMPTY_DATA_FRAMES: u8 = 10;
201
202/// Request payload of a stream.
203struct StreamPayload {
204    sender: PayloadSender,
205    /// The response is complete
206    complete: bool,
207}
208
209impl<Err> PublishService<Err>
210where
211    Err: ResponseError,
212{
213    fn new(
214        id: usize,
215        io: IoRef,
216        svc: Pipeline<Request, Response, Err>,
217        control: PipelineBinding<h2::Control<Error<H2Error>>, h2::ControlAck, DispatchError>,
218    ) -> Self {
219        Self {
220            id,
221            io,
222            svc,
223            control,
224            streams: Rc::new(RefCell::new(HashMap::default())),
225            empty_data: Cell::new(0),
226        }
227    }
228}
229
230impl<Err> Service<(), h2::Message> for PublishService<Err>
231where
232    Err: ResponseError + 'static,
233{
234    type Res = ();
235    type Error = Error<H2Error>;
236
237    async fn shutdown(&self, _: Ctx<'_, Self, ()>) {
238        self.svc.shutdown().await;
239    }
240
241    async fn call(&self, msg: h2::Message, _: Ctx<'_, Self, ()>) -> Result<Self::Res, Self::Error> {
242        let h2::Message { stream, kind } = msg;
243        let (io, pseudo, headers, eof, payload) = match kind {
244            h2::MessageKind::Headers {
245                pseudo,
246                headers,
247                eof,
248            } => {
249                let pl = if eof {
250                    None
251                } else {
252                    #[cfg(feature = "trace")]
253                    log::debug!(
254                        "{}: Creating local payload stream for {:?}",
255                        self.io.tag(),
256                        stream.id()
257                    );
258                    let (sender, payload) = Payload::create(stream.empty_capacity());
259                    self.streams.borrow_mut().insert(
260                        stream.id(),
261                        StreamPayload {
262                            sender,
263                            complete: false,
264                        },
265                    );
266                    Some(payload)
267                };
268                (self.io.clone(), pseudo, headers, eof, pl)
269            }
270            h2::MessageKind::Data(data, cap) => {
271                #[cfg(feature = "trace")]
272                log::debug!(
273                    "{}: Got data chunk for {:?}: {:?}",
274                    self.io.tag(),
275                    stream.id(),
276                    data.len()
277                );
278                // the limit is sticky, frames buffered after it are not delivered
279                let count = if data.is_empty() {
280                    self.empty_data.get().saturating_add(1)
281                } else if self.empty_data.get() >= MAX_EMPTY_DATA_FRAMES {
282                    MAX_EMPTY_DATA_FRAMES
283                } else {
284                    0
285                };
286                self.empty_data.set(count);
287                if count >= MAX_EMPTY_DATA_FRAMES {
288                    log::debug!(
289                        "{}: Too many consecutive empty DATA frames, closing connection",
290                        self.io.tag()
291                    );
292                    return Err(H2Error::EmptyDataFrames.into());
293                }
294                let mut streams = self.streams.borrow_mut();
295                if let Some(pl) = streams.get(&stream.id()) {
296                    if pl.complete && pl.sender.is_dropped() {
297                        // the response is complete, the request body is dropped unread
298                        streams.remove(&stream.id());
299                        drop(streams);
300                        stream.reset(h2::frame::Reason::NO_ERROR);
301                    } else {
302                        pl.sender.feed_data(data, cap);
303                    }
304                } else {
305                    log::error!(
306                        "{}: Payload stream does not exists for {:?}",
307                        self.io.tag(),
308                        stream.id()
309                    );
310                }
311                return Ok(());
312            }
313            h2::MessageKind::Eof(item) => {
314                log::debug!(
315                    "{}: Got payload eof for {:?}: {item:?}",
316                    self.io.tag(),
317                    stream.id()
318                );
319                if self.empty_data.get() >= MAX_EMPTY_DATA_FRAMES {
320                    return Err(H2Error::EmptyDataFrames.into());
321                }
322                if matches!(item, h2::StreamEof::Data(ref data, _) if !data.is_empty()) {
323                    self.empty_data.set(0);
324                }
325                if let Some(StreamPayload { sender, .. }) =
326                    self.streams.borrow_mut().remove(&stream.id())
327                {
328                    match item {
329                        h2::StreamEof::Data(data, cap) => {
330                            sender.feed_eof(data, Some(cap));
331                        }
332                        h2::StreamEof::Trailers(hdrs) => {
333                            sender.feed_trailers(hdrs);
334                        }
335                        h2::StreamEof::Error(err) => {
336                            sender.set_error(err.into_error().into());
337                        }
338                    }
339                }
340                return Ok(());
341            }
342            h2::MessageKind::Disconnect(err) => {
343                log::debug!("{}: Connection is disconnected {err:?}", self.io.tag());
344                if let Some(pl) = self.streams.borrow_mut().remove(&stream.id()) {
345                    pl.sender
346                        .set_error(io::Error::new(io::ErrorKind::UnexpectedEof, err).into());
347                }
348                return Ok(());
349            }
350        };
351
352        // a malformed request is a stream error, the connection stays open,
353        // see RFC 9113 section 8.1.1
354        let Some((method, uri)) = request_uri(&pseudo) else {
355            log::debug!(
356                "{}: Malformed request on {:?}: {pseudo:?}",
357                self.io.tag(),
358                stream.id()
359            );
360            self.streams.borrow_mut().remove(&stream.id());
361
362            let mut res = Response::new(StatusCode::BAD_REQUEST).drop_body();
363            let head = res.head_mut();
364            prepare_response(head, &mut BodySize::Empty);
365            let _ = stream.send_response(StatusCode::BAD_REQUEST, &head.headers, true);
366            if !eof {
367                // the request body is not needed
368                stream.reset(h2::frame::Reason::NO_ERROR);
369            }
370            return Ok(());
371        };
372
373        // the client waits for `100 Continue` before sending the request body,
374        // see RFC 9110 section 10.1.1
375        let (pseudo, headers) = if !eof && expect_continue(&headers) {
376            let msg = h2::Control::expect(stream.clone(), pseudo, headers);
377            match self
378                .control
379                .call(msg)
380                .await
381                .map(h2::ControlAck::into_expect)
382            {
383                Ok(Some(ExpectResult::Continue(expect))) => {
384                    if stream
385                        .send_informational(StatusCode::CONTINUE, HeaderMap::new())
386                        .is_err()
387                    {
388                        // the stream is closed
389                        self.streams.borrow_mut().remove(&stream.id());
390                        return Ok(());
391                    }
392                    let (_, pseudo, headers) = expect.into_parts();
393                    (pseudo, headers)
394                }
395                Ok(Some(ExpectResult::Failed(_, status, headers))) => {
396                    self.streams.borrow_mut().remove(&stream.id());
397
398                    let mut res = Response::new(status).drop_body();
399                    let head = res.head_mut();
400                    h2::recycle_header_map(mem::replace(&mut head.headers, headers));
401                    prepare_response(head, &mut BodySize::Empty);
402                    let _ = stream.send_response(status, &head.headers, true);
403
404                    // the request body is not needed
405                    stream.reset(h2::frame::Reason::NO_ERROR);
406                    return Ok(());
407                }
408                result => {
409                    log::error!(
410                        "{}: Control service failed to handle expect for {:?}: {:?}",
411                        self.io.tag(),
412                        stream.id(),
413                        result.map(|_| ())
414                    );
415                    self.streams.borrow_mut().remove(&stream.id());
416                    stream.reset(h2::frame::Reason::INTERNAL_ERROR);
417                    return Ok(());
418                }
419            }
420        } else {
421            (pseudo, headers)
422        };
423
424        log::trace!(
425            "{}: {:?} got request (eof: {eof}): {pseudo:#?}\nheaders: {headers:#?}",
426            self.io.tag(),
427            stream.id()
428        );
429        let mut req = if let Some(pl) = payload {
430            Request::with_payload(crate::http::Payload::H2(pl))
431        } else {
432            Request::new()
433        };
434
435        let is_head_req = method == Method::HEAD;
436        let head = req.head_mut();
437        head.uri = uri;
438        head.version = Version::HTTP_2;
439        head.method = method;
440        // the pooled head map is empty, the h2 decoder reuses its allocation
441        h2::recycle_header_map(mem::replace(&mut head.headers, headers));
442        head.io = CurrentIo::Ref(io);
443        head.id = self.id;
444
445        let result = self.svc.call(req).await;
446        let (mut res, mut body) = Response::from(result).into_parts();
447
448        // an interim response cannot complete the request and `101` is not
449        // supported in HTTP/2, see RFC 9113 section 8.1 and 8.6
450        let status = res.status();
451        if status.is_informational() {
452            log::error!(
453                "{}: Informational response {status} is not supported, sending 500",
454                self.io.tag()
455            );
456            res = Response::new(StatusCode::INTERNAL_SERVER_ERROR).drop_body();
457            body = ResponseBody::Other(Body::Empty);
458        }
459
460        let head = res.head_mut();
461        let mut size = body.size();
462        prepare_response(head, &mut size);
463
464        #[cfg(feature = "trace")]
465        log::debug!(
466            "{}: Received service response: {head:?} payload: {size:?}",
467            self.io.tag()
468        );
469
470        // borrowed, the pooled response head keeps its header map capacity
471        let hdrs = &head.headers;
472        // `Err(Some(_))` is a body error, `Err(None)` is a closed stream
473        let sent = async {
474            if size.is_eof() || is_head_req {
475                stream
476                    .send_response(head.status, hdrs, true)
477                    .map_err(|_| None)?;
478                return Ok(());
479            }
480            stream
481                .send_response(head.status, hdrs, false)
482                .map_err(|_| None)?;
483
484            // the last chunk of a sized body ends the stream, no empty eof frame
485            let mut remaining = if let BodySize::Sized(len) = size {
486                Some(len)
487            } else {
488                None
489            };
490            loop {
491                match poll_fn(|cx| body.poll_next_chunk(cx)).await {
492                    None => {
493                        #[cfg(feature = "trace")]
494                        log::debug!(
495                            "{}: {:?} closing payload stream",
496                            self.io.tag(),
497                            stream.id()
498                        );
499                        return stream
500                            .send_payload(Bytes::new(), true)
501                            .await
502                            .map_err(|_| None);
503                    }
504                    Some(Ok(chunk)) => {
505                        #[cfg(feature = "trace")]
506                        log::debug!(
507                            "{}: {:?} sending data chunk {:?} bytes",
508                            self.io.tag(),
509                            stream.id(),
510                            chunk.len()
511                        );
512                        if !chunk.is_empty() {
513                            let eof = remaining.as_mut().is_some_and(|rem| {
514                                *rem = rem.saturating_sub(chunk.len() as u64);
515                                *rem == 0
516                            });
517                            stream.send_payload(chunk, eof).await.map_err(|_| None)?;
518                            if eof {
519                                return Ok(());
520                            }
521                        }
522                    }
523                    Some(Err(e)) => return Err(Some(e)),
524                }
525            }
526        }
527        .await;
528
529        match sent {
530            Ok(()) => (),
531            Err(Some(e)) => {
532                // only the stream fails, the connection stays open
533                log::error!(
534                    "{}: Response payload stream error for {:?}: {e:?}",
535                    self.io.tag(),
536                    stream.id()
537                );
538                stream.reset(h2::frame::Reason::INTERNAL_ERROR);
539                return Ok(());
540            }
541            Err(None) => {
542                // the stream is reset or the connection is closed
543                self.streams.borrow_mut().remove(&stream.id());
544                return Ok(());
545            }
546        }
547
548        // the response is complete, an unread request body must not keep the stream
549        let id = stream.id();
550        let mut streams = self.streams.borrow_mut();
551        if let Some(pl) = streams.get_mut(&id) {
552            if pl.sender.is_dropped() {
553                streams.remove(&id);
554                drop(streams);
555                stream.reset(h2::frame::Reason::NO_ERROR);
556            } else {
557                // the app still holds the request body, release the stream once it is dropped
558                pl.complete = true;
559                let streams = Rc::downgrade(&self.streams);
560                pl.sender.on_drop(move || {
561                    if let Some(streams) = streams.upgrade()
562                        && let Ok(mut streams) = streams.try_borrow_mut()
563                    {
564                        streams.remove(&id);
565                    }
566                    stream.reset(h2::frame::Reason::NO_ERROR);
567                });
568            }
569        }
570        Ok(())
571    }
572}
573
574#[allow(clippy::declare_interior_mutable_const)]
575const ZERO_CONTENT_LENGTH: HeaderValue = HeaderValue::from_static("0");
576#[allow(clippy::declare_interior_mutable_const)]
577const KEEP_ALIVE: HeaderName = HeaderName::from_static("keep-alive");
578#[allow(clippy::declare_interior_mutable_const)]
579const PROXY_CONNECTION: HeaderName = HeaderName::from_static("proxy-connection");
580
581/// Builds the request method and uri from the pseudo headers,
582/// returns `None` for a malformed request.
583fn request_uri(pseudo: &h2::frame::PseudoHeaders) -> Option<(Method, Url)> {
584    let method = pseudo.method.clone()?;
585    // userinfo is deprecated, see RFC 9113 section 8.3.1
586    let authority = match pseudo.authority {
587        Some(ref authority) => {
588            // An empty host is "" or ":port"; IPv6 hosts are bracketed.
589            if authority.is_empty() || authority.starts_with(':') || authority.contains('@') {
590                return None;
591            }
592            Some(Authority::new(authority.as_str()).ok()?)
593        }
594        None => None,
595    };
596    let path = pseudo.path.as_ref().map(crate::util::ByteString::as_str);
597    // the same request-target checks as http/1, `Url` parsing is lenient
598    if path.is_some_and(|p| !is_valid_target(p.as_bytes())) {
599        return None;
600    }
601
602    let uri = match (authority, path) {
603        // CONNECT request uses the authority form
604        (Some(authority), None) if method == Method::CONNECT => {
605            Url::parse(authority.as_str()).ok()?
606        }
607        (_, Some("*")) if method == Method::OPTIONS => Url::from_static("*"),
608        (Some(authority), Some(path)) if path.starts_with('/') => {
609            let scheme = Scheme::new(pseudo.scheme.as_ref()?.as_str()).ok()?;
610            Url::try_from(concat(&[scheme.as_str(), "://", authority.as_str(), path])).ok()?
611        }
612        (None, Some(path)) if path.starts_with('/') => {
613            let uri = if path.starts_with("//") {
614                // origin-form path, not a network-path reference
615                Url::try_from(format!("/.{path}")).ok()?
616            } else {
617                // reuses the buffer if the path is normalized
618                Url::try_from(pseudo.path.as_ref()?).ok()?
619            };
620            if uri.authority().is_some() {
621                return None;
622            }
623            uri
624        }
625        _ => return None,
626    };
627    Some((method, uri))
628}
629
630/// Concatenates `pieces` into a buffer that `Url` reuses if the result is
631/// normalized, short strings are stored inline without allocation.
632fn concat(pieces: &[&str]) -> Bytes {
633    const INLINE: usize = 23;
634
635    let len = pieces.iter().map(|p| p.len()).sum::<usize>();
636    if len <= INLINE {
637        let mut buf = [0u8; INLINE];
638        let mut pos = 0;
639        for p in pieces {
640            buf[pos..pos + p.len()].copy_from_slice(p.as_bytes());
641            pos += p.len();
642        }
643        Bytes::copy_from_slice(&buf[..len])
644    } else {
645        let mut buf = BytesMut::with_capacity(len);
646        for p in pieces {
647            buf.extend_from_slice(p.as_bytes());
648        }
649        buf.freeze()
650    }
651}
652
653/// Checks the case-insensitive `100-continue` expectation, see RFC 9110 section 10.1.1
654fn expect_continue(headers: &HeaderMap) -> bool {
655    headers.get_all(header::EXPECT).any(|value| {
656        value
657            .as_bytes()
658            .split(|&b| b == b',')
659            .any(|e| e.trim_ascii().eq_ignore_ascii_case(b"100-continue"))
660    })
661}
662
663fn prepare_response(head: &mut ResponseHead, size: &mut BodySize) {
664    // Content length
665    // `204` and `304` responses never have a body, see RFC 9110 section 15.3.5 and 15.4.5
666    if head.status == StatusCode::NO_CONTENT || head.status == StatusCode::NOT_MODIFIED {
667        *size = BodySize::None;
668    }
669    match size {
670        BodySize::None | BodySize::Stream => head.headers.remove(header::CONTENT_LENGTH),
671        BodySize::Empty => head
672            .headers
673            .insert(header::CONTENT_LENGTH, ZERO_CONTENT_LENGTH),
674        BodySize::Sized(len) => {
675            head.headers
676                .insert(header::CONTENT_LENGTH, HeaderValue::from(*len));
677        }
678    }
679
680    // http2 specific1
681    head.headers.remove(header::CONNECTION);
682    head.headers.remove(header::TRANSFER_ENCODING);
683    head.headers.remove(header::UPGRADE);
684
685    // omit HTTP/1.x only headers according to:
686    // https://datatracker.ietf.org/doc/html/rfc7540#section-8.1.2.2
687    head.headers.remove(KEEP_ALIVE);
688    head.headers.remove(PROXY_CONNECTION);
689
690    // set date header
691    if !head.headers.contains_key(header::DATE) {
692        head.headers
693            .insert(header::DATE, DateService::header_value());
694    }
695}
696
697#[cfg(test)]
698mod tests {
699    use super::*;
700
701    #[test]
702    fn test_request_uri() {
703        let mut pseudo = h2::frame::PseudoHeaders::default();
704        assert!(request_uri(&pseudo).is_none());
705
706        // CONNECT uses the authority form
707        pseudo.method = Some(Method::CONNECT);
708        pseudo.authority = Some("example.com:443".into());
709        let (method, uri) = request_uri(&pseudo).unwrap();
710        assert_eq!(method, Method::CONNECT);
711        assert_eq!(uri.authority().unwrap(), "example.com:443");
712
713        for authority in ["bad authority", "[email protected]:443", ":443"] {
714            pseudo.authority = Some(authority.into());
715            assert!(request_uri(&pseudo).is_none(), "{authority}");
716        }
717
718        // inline and heap buffers
719        for authority in ["a.io:1", "very-long-host-name.example.com:8443"] {
720            pseudo.authority = Some(authority.into());
721            let (_, uri) = request_uri(&pseudo).unwrap();
722            assert_eq!(uri.authority().unwrap(), authority);
723            assert_eq!(uri.to_string(), format!("//{authority}"));
724        }
725
726        // absolute form requires the scheme
727        pseudo.method = Some(Method::GET);
728        pseudo.authority = Some("example.com".into());
729        pseudo.path = Some("/path".into());
730        assert!(request_uri(&pseudo).is_none());
731        pseudo.scheme = Some("https".into());
732        let (_, uri) = request_uri(&pseudo).unwrap();
733        assert_eq!(uri.to_string(), "https://example.com/path");
734        // userinfo and an empty host are rejected
735        for authority in ["u:[email protected]", ""] {
736            pseudo.authority = Some(authority.into());
737            assert!(request_uri(&pseudo).is_none(), "{authority}");
738        }
739        pseudo.authority = Some("example.com".into());
740
741        // 23 bytes are stored inline, 24 bytes on the heap
742        for path in ["/123", "/1234", "/api/v1/users/1?fields=name,email"] {
743            pseudo.path = Some(path.into());
744            let (_, uri) = request_uri(&pseudo).unwrap();
745            assert_eq!(uri.to_string(), format!("https://example.com{path}"));
746        }
747        // the buffer is normalized if needed
748        pseudo.scheme = Some("HTTPS".into());
749        pseudo.authority = Some("Example.COM".into());
750        pseudo.path = Some("/a/./b/../c".into());
751        let (_, uri) = request_uri(&pseudo).unwrap();
752        assert_eq!(uri.to_string(), "https://example.com/a/c");
753        pseudo.path = Some("/a/./b/../c/long-enough-for-the-heap".into());
754        let (_, uri) = request_uri(&pseudo).unwrap();
755        assert_eq!(
756            uri.to_string(),
757            "https://example.com/a/c/long-enough-for-the-heap"
758        );
759        pseudo.authority = Some("example.com".into());
760        pseudo.path = Some("/path".into());
761
762        pseudo.scheme = Some("ht/tp".into());
763        assert!(request_uri(&pseudo).is_none());
764        pseudo.scheme = Some("https".into());
765        pseudo.authority = Some("example.com/x?".into());
766        assert!(request_uri(&pseudo).is_none());
767        pseudo.authority = Some("example.com".into());
768        pseudo.path = Some("x".into());
769        assert!(request_uri(&pseudo).is_none());
770
771        pseudo.authority = None;
772        pseudo.path = Some("/path".into());
773        let (_, uri) = request_uri(&pseudo).unwrap();
774        assert_eq!(uri.path(), "/path");
775        pseudo.path = Some("//example.com/path".into());
776        let (_, uri) = request_uri(&pseudo).unwrap();
777        assert_eq!(uri.path(), "//example.com/path");
778        assert!(uri.authority().is_none());
779        pseudo.path = Some("/a b".into());
780        assert!(request_uri(&pseudo).is_none());
781        pseudo.path = Some("/a\x7f".into());
782        assert!(request_uri(&pseudo).is_none());
783        for path in ["/a#f", "/a%zz", "/a\"", "/a%00"] {
784            pseudo.path = Some(path.into());
785            assert!(request_uri(&pseudo).is_none(), "{path}");
786        }
787        pseudo.path = Some("/a%3b/%2e%2e/b?%00".into());
788        let (_, uri) = request_uri(&pseudo).unwrap();
789        assert_eq!(uri.path_and_query(), "/b?%00");
790
791        pseudo.path = Some("*".into());
792        assert!(request_uri(&pseudo).is_none());
793        pseudo.method = Some(Method::OPTIONS);
794        let (_, uri) = request_uri(&pseudo).unwrap();
795        assert_eq!(uri.path(), "*");
796    }
797
798    #[test]
799    fn test_request_uri_authority() {
800        for (method, path) in [
801            (Method::GET, Some("/path")),
802            (Method::CONNECT, None),
803            (Method::OPTIONS, Some("*")),
804        ] {
805            let mut pseudo = h2::frame::PseudoHeaders {
806                method: Some(method),
807                scheme: path.map(|_| "https".into()),
808                path: path.map(Into::into),
809                ..Default::default()
810            };
811            for authority in [
812                "",
813                ":",
814                ":443",
815                "@example.com",
816                "u:[email protected]:443",
817                "example.com@",
818                "[email protected]",
819                "bad authority",
820                "exa\tmple.com",
821                "m\u{fc}nchen.de",
822                "example.com/path",
823                "example.com?query",
824                "example.com#fragment",
825                "example.com%zz",
826                "example.com:abc",
827                "example.com:+443",
828                "example.com:65536",
829                "127.0.0.256",
830                "::1",
831                "[::1",
832                "[::g]",
833                "[::1]:65536",
834            ] {
835                pseudo.authority = Some(authority.into());
836                assert!(request_uri(&pseudo).is_none(), "{pseudo:?}");
837            }
838            for (authority, normalized) in [
839                ("example.com", "example.com"),
840                ("Example.COM:00080", "example.com:80"),
841                ("example.com:", "example.com"),
842                ("example.com:0", "example.com:0"),
843                ("example.com:65535", "example.com:65535"),
844                ("%45xample.com:443", "example.com:443"),
845                ("127.0.0.1:8443", "127.0.0.1:8443"),
846                ("[::1]", "[::1]"),
847                ("[0:0:0:0:0:0:0:1]:00443", "[::1]:443"),
848            ] {
849                pseudo.authority = Some(authority.into());
850                let (_, uri) = request_uri(&pseudo).unwrap();
851                let expected = match path {
852                    None => format!("//{normalized}"),
853                    Some("*") => "*".to_owned(),
854                    _ => format!("https://{normalized}/path"),
855                };
856                assert_eq!(uri.as_str(), expected, "{pseudo:?}");
857            }
858        }
859    }
860
861    #[test]
862    fn test_concat() {
863        assert_eq!(concat(&[]), "");
864        assert_eq!(concat(&["//", "a.io"]), "//a.io");
865        let inline = "x".repeat(23);
866        assert_eq!(concat(&[&inline[..20], &inline[20..]]), inline.as_str());
867        let heap = "y".repeat(24);
868        assert_eq!(concat(&[&heap[..1], "", &heap[1..]]), heap.as_str());
869    }
870}
ntex/http/h2/service.rs

ntex/http/h2/
service.rs
