Sync upstream (hyperium#15)

* chore: allow matching infallible (hyperium#796)

* v0.4.6

* chore(ci): use tokio-util 0.7.11 in MSRV check

* style: replace `split_to` and `split_off` with better alternatives

This removes `let _ = ` from in front of `split_to` and `split_off`
and mostly follows the suggestions from the `#[must_use]` impls.
One of the uses of `split_to` is instead replaced with `take`.

* improve ci/h2spec.sh (macOS compat, /tmp dir and overwrite) (hyperium#809)

- detect if run on MacOS, so we download h2spec macos build in that case
- support overwriting h2spec detection so we anyway download new file,
  useful in case you switch to new version for example
- move h2spec, archive and log all to /tmp dir as to not polute
  the repo dir

* ci: pin hashbrown for msrv job (hyperium#814)

* fix: HEADERS frame with non-zero content-length and END_STREAM is malformed (hyperium#813)

Before this change, content-length underflow is only checked when
receiving date frames. The underflow error was never triggered if
data frames are never received.

This change adds similar check for headers frames.

* fix: notify_recv after send_reset() in reset_on_recv_stream_err() to ensure local stream is released properly (hyperium#816) (hyperium#818)

Similar to what have been done in fn send_reset<B>(), we should notify RecvStream that is parked after send_reset().

Co-authored-by: Jiahao Liang <[email protected]>

---------

Co-authored-by: Sean McArthur <[email protected]>
Co-authored-by: tottoto <[email protected]>
Co-authored-by: Paolo Barbolini <[email protected]>
Co-authored-by: Glen De Cauwsemaecker <[email protected]>
Co-authored-by: Yuchen Wu <[email protected]>
Co-authored-by: Jiahao Liang <[email protected]>

Author: 7 people

.github/workflows/CI.yml

@@ -49,9 +49,11 @@ jobs:
         with:
           toolchain: ${{ steps.msrv.outputs.version }}
 
-      - name: Make sure tokio 1.38.1 is used for MSRV
+      - name: Pin some dependencies for MSRV
         run: |
-          cargo update
           cargo update --package tokio --precise 1.38.1
+          cargo update --package tokio-util --precise 0.7.11
+          cargo update --package hashbrown --precise 0.15.0
+
 
       - run: cargo check -p rh2

CHANGELOG.md

@@ -1,3 +1,10 @@
+# 0.4.6 (August 19, 2024)
+
+* Add `current_max_send_streams()` and `current_max_recv_streams()` to `client::SendRequest`.
+* Fix sending a PROTOCOL_ERROR instead of REFUSED_STREAM when receiving oversized headers.
+* Fix notifying a PushPromise task properly.
+* Fix notifying a stream task when reset.
+
 # 0.4.5 (May 17, 2024)
 
 * Fix race condition that sometimes hung connections during shutdown.

Cargo.toml

@@ -3,7 +3,7 @@ name = "rh2"
 # When releasing to crates.io:
 # - Update CHANGELOG.md.
 # - Create git tag
-version = "0.4.5"
+version = "0.4.6"
 license = "MIT"
 authors = ["0x676e67 <[email protected]>"]
 description = "An HTTP/2 client and server"

ci/h2spec.sh

@@ -1,10 +1,25 @@
 #!/bin/bash
-LOGFILE="h2server.log"
+LOGFILE="/tmp/h2server.log"
 
-if ! [ -e "h2spec" ] ; then
+override_h2spec=false
+
+# Check for optional flag
+while getopts "F" opt; do
+    case $opt in
+        F) override_h2spec=true ;;
+        *) echo "Usage: $0 [-o]"; exit 1 ;;
+    esac
+done
+
+if ! [ -e "/tmp/h2spec" ] || $override_h2spec ; then
     # if we don't already have a h2spec executable, wget it from github
-    wget https://github.com/summerwind/h2spec/releases/download/v2.1.1/h2spec_linux_amd64.tar.gz
-    tar xf h2spec_linux_amd64.tar.gz
+    if [[ "$OSTYPE" == "darwin"* ]]; then
+        curl -L -o /tmp/h2spec_darwin_amd64.tar.gz https://github.com/summerwind/h2spec/releases/download/v2.1.1/h2spec_darwin_amd64.tar.gz \
+            && tar xf /tmp/h2spec_darwin_amd64.tar.gz -C /tmp
+    else
+        curl -L -o /tmp/h2spec_linux_amd64.tar.gz https://github.com/summerwind/h2spec/releases/download/v2.1.1/h2spec_linux_amd64.tar.gz \
+            && tar xf /tmp/h2spec_linux_amd64.tar.gz -C /tmp
+    fi
 fi
 
 cargo build --example server
@@ -16,7 +31,7 @@ SERVER_PID=$!
 sed '/listening on Ok(127.0.0.1:5928)/q' <&3 ; cat <&3 > "${LOGFILE}" &
 
 # run h2spec against the server, printing the server log if h2spec failed
-./h2spec -p 5928
+/tmp/h2spec -p 5928
 H2SPEC_STATUS=$?
 if [ "${H2SPEC_STATUS}" -eq 0 ]; then
     echo "h2spec passed!"

src/codec/framed_read.rs

@@ -8,7 +8,7 @@ use crate::hpack;
 
 use futures_core::Stream;
 
-use bytes::BytesMut;
+use bytes::{Buf, BytesMut};
 
 use std::io;
 
@@ -146,8 +146,7 @@ fn decode_frame(
     macro_rules! header_block {
         ($frame:ident, $head:ident, $bytes:ident) => ({
             // Drop the frame header
-            // TODO: Change to drain: carllerche/bytes#130
-            let _ = $bytes.split_to(frame::HEADER_LEN);
+            $bytes.advance(frame::HEADER_LEN);
 
             // Parse the header frame w/o parsing the payload
             let (mut frame, mut payload) = match frame::$frame::load($head, $bytes) {
@@ -227,7 +226,7 @@ fn decode_frame(
             .into()
         }
         Kind::Data => {
-            let _ = bytes.split_to(frame::HEADER_LEN);
+            bytes.advance(frame::HEADER_LEN);
             let res = frame::Data::load(head, bytes.freeze());
 
             // TODO: Should this always be connection level? Probably not...

src/frame/headers.rs

@@ -3,10 +3,11 @@ use crate::ext::Protocol;
 use crate::frame::{Error, Frame, Head, Kind};
 use crate::hpack::{self, BytesStr};
 
-use bytes::{BufMut, Bytes, BytesMut};
 use http::header::{self, HeaderName, HeaderValue};
 use http::{uri, HeaderMap, Method, Request, StatusCode, Uri};
 
+use bytes::{Buf, BufMut, Bytes, BytesMut};
+
 use std::fmt;
 use std::io::Cursor;
 
@@ -181,7 +182,7 @@ impl Headers {
             pad = src[0] as usize;
 
             // Drop the padding
-            let _ = src.split_to(1);
+            src.advance(1);
         }
 
         // Read the stream dependency
@@ -196,7 +197,7 @@ impl Headers {
             }
 
             // Drop the next 5 bytes
-            let _ = src.split_to(5);
+            src.advance(5);
 
             Some(stream_dep)
         } else {
@@ -269,6 +270,10 @@ impl Headers {
         &mut self.header_block.pseudo
     }
 
+    pub(crate) fn pseudo(&self) -> &Pseudo {
+        &self.header_block.pseudo
+    }
+
     /// Whether it has status 1xx
     pub(crate) fn is_informational(&self) -> bool {
         self.header_block.pseudo.is_informational()
@@ -445,7 +450,7 @@ impl PushPromise {
             pad = src[0] as usize;
 
             // Drop the padding
-            let _ = src.split_to(1);
+            src.advance(1);
         }
 
         if src.len() < 5 {
@@ -454,7 +459,7 @@ impl PushPromise {
 
         let (promised_id, _) = StreamId::parse(&src[..4]);
         // Drop promised_id bytes
-        let _ = src.split_to(4);
+        src.advance(4);
 
         if pad > 0 {
             if pad > src.len() {
@@ -684,7 +689,7 @@ impl EncodingHeaderBlock {
 
         // Now, encode the header payload
         let continuation = if self.hpack.len() > dst.remaining_mut() {
-            dst.put_slice(&self.hpack.split_to(dst.remaining_mut()));
+            dst.put((&mut self.hpack).take(dst.remaining_mut()));
 
             Some(Continuation {
                 stream_id: head.stream_id(),

src/frame/util.rs

@@ -1,7 +1,7 @@
 use std::fmt;
 
 use super::Error;
-use bytes::Bytes;
+use bytes::{Buf, Bytes};
 
 /// Strip padding from the given payload.
 ///
@@ -32,8 +32,8 @@ pub fn strip_padding(payload: &mut Bytes) -> Result<u8, Error> {
         return Err(Error::TooMuchPadding);
     }
 
-    let _ = payload.split_to(1);
-    let _ = payload.split_off(payload_len - pad_len - 1);
+    payload.advance(1);
+    payload.truncate(payload_len - pad_len - 1);
 
     Ok(pad_len as u8)
 }

src/proto/streams/recv.rs

@@ -185,6 +185,18 @@ impl Recv {
                 };
 
                 stream.content_length = ContentLength::Remaining(content_length);
+                // END_STREAM on headers frame with non-zero content-length is malformed.
+                // https://datatracker.ietf.org/doc/html/rfc9113#section-8.1.1
+                if frame.is_end_stream()
+                    && content_length > 0
+                    && frame
+                        .pseudo()
+                        .status
+                        .map_or(true, |status| status != 204 && status != 304)
+                {
+                    proto_err!(stream: "recv_headers with END_STREAM: content-length is not zero; stream={:?};", stream.id);
+                    return Err(Error::library_reset(stream.id, Reason::PROTOCOL_ERROR).into());
+                }
             }
         }
 

src/proto/streams/store.rs

@@ -137,6 +137,7 @@ impl Store {
             Ok::<_, Infallible>(())
         }) {
             Ok(()) => (),
+            #[allow(unused)]
             Err(infallible) => match infallible {},
         }
     }

src/proto/streams/streams.rs

@@ -1599,6 +1599,9 @@ impl Actions {
                 // Reset the stream.
                 self.send
                     .send_reset(reason, initiator, buffer, stream, counts, &mut self.task);
+                self.recv.enqueue_reset_expiration(stream, counts);
+                // if a RecvStream is parked, ensure it's notified
+                stream.notify_recv();
                 Ok(())
             } else {
                 tracing::warn!(

tests/h2-tests/tests/client_request.rs

@@ -1261,6 +1261,49 @@ async fn allow_empty_data_for_head() {
     join(srv, h2).await;
 }
 
+#[tokio::test]
+async fn reject_none_zero_content_length_header_with_end_stream() {
+    h2_support::trace_init!();
+    let (io, mut srv) = mock::new();
+
+    let srv = async move {
+        let settings = srv.assert_client_handshake().await;
+        assert_default_settings!(settings);
+        srv.recv_frame(
+            frames::headers(1)
+                .request("GET", "https://example.com/")
+                .eos(),
+        )
+        .await;
+        srv.send_frame(
+            frames::headers(1)
+                .response(200)
+                .field("content-length", 100)
+                .eos(),
+        )
+        .await;
+    };
+
+    let h2 = async move {
+        let (mut client, h2) = client::Builder::new()
+            .handshake::<_, Bytes>(io)
+            .await
+            .unwrap();
+        tokio::spawn(async {
+            h2.await.expect("connection failed");
+        });
+        let request = Request::builder()
+            .method(Method::GET)
+            .uri("https://example.com/")
+            .body(())
+            .unwrap();
+        let (response, _) = client.send_request(request, true).unwrap();
+        let _ = response.await.unwrap_err();
+    };
+
+    join(srv, h2).await;
+}
+
 #[tokio::test]
 async fn early_hints() {
     h2_support::trace_init!();

tests/h2-tests/tests/stream_states.rs

@@ -536,7 +536,12 @@ async fn recv_next_stream_id_updated_by_malformed_headers() {
         client.recv_frame(frames::go_away(1).protocol_error()).await;
     };
     let srv = async move {
-        let mut srv = server::handshake(io).await.expect("handshake");
+        let mut srv = server::Builder::new()
+            // forget the bad stream immediately
+            .max_concurrent_reset_streams(0)
+            .handshake::<_, Bytes>(io)
+            .await
+            .expect("handshake");
         let res = srv.next().await.unwrap();
         let err = res.unwrap_err();
         assert_eq!(err.reason(), Some(h2::Reason::PROTOCOL_ERROR));