Respond 400 + Reset to malformed request (hyperium#747)

`h2` returns `RST_STREAM` frames with the `PROTOCOL_ERROR` bit set as a
response to many types of errors in client requests. Many of those cases,
when handled by an HTTP/1 server such as the one used in `hyper`, would
result in an HTTP 400 Bad Request response returned to the client rather
than a TCP reset (the HTTP/1 equivalent of a `RST_STREAM`). As a
consequence, a client will observe differences in behaviour between
HTTP/1 and HTTP/2: a malformed request will result in a 400 response when
HTTP/1 is used whereas the same request will result in a reset stream
with `h2`.

Make `h2` reply a `HEADERS`+`400`+`END_STREAM` frame followed by a
`RST_STREAM`+`PROTOCOL_ERROR` frame to all the `malformed!()` macro
invocations in `Peer::convert_poll_message()` in `src/server.rs`.

The `Reset` variant in the `h2::proto::error::Error` Enum now contains an
`Option<http::StatusCode>` value that is set by the `malformed!()` macro
with a `Some(400)`. That value is propagated all the way until
`h2::proto::streams::send::Send::send_reset()` where, if not `None`, a
`h2::frame::headers::Headers` frame with the status code and the
`END_STREAM` flag set will now be sent to the client before the
`RST_STREAM` frame.

Some of the parameters passed to `send_reset()` have been grouped into a
new `SendResetContext` Struct in order to avoid a
`clippy::too_many_arguments` lint.

Tests where malformed requests were sent have been updated to check that
an additional  `HEADERS`+`400`+`END_STREAM` frame is now received before
the `RST_STREAM + PROTOCOL_ERROR` frame.

This change has been validated with  other clients like `curl`, a custom
ad-hoc client written with `python3+httpx` and `varnishtest`.

src/error.rs

@@ -3,6 +3,7 @@ use crate::frame::StreamId;
 use crate::proto::{self, Initiator};
 
 use bytes::Bytes;
+use http::StatusCode;
 use std::{error, fmt, io};
 
 pub use crate::frame::Reason;
@@ -26,7 +27,7 @@ pub struct Error {
 enum Kind {
     /// A RST_STREAM frame was received or sent.
     #[allow(dead_code)]
-    Reset(StreamId, Reason, Initiator),
+    Reset(StreamId, Reason, Initiator, Option<StatusCode>),
 
     /// A GO_AWAY frame was received or sent.
     GoAway(Bytes, Reason, Initiator),
@@ -51,7 +52,7 @@ impl Error {
     /// action taken by the peer (i.e. a protocol error).
     pub fn reason(&self) -> Option<Reason> {
         match self.kind {
-            Kind::Reset(_, reason, _) | Kind::GoAway(_, reason, _) | Kind::Reason(reason) => {
+            Kind::Reset(_, reason, _, _) | Kind::GoAway(_, reason, _) | Kind::Reason(reason) => {
                 Some(reason)
             }
             _ => None,
@@ -101,7 +102,7 @@ impl Error {
     pub fn is_remote(&self) -> bool {
         matches!(
             self.kind,
-            Kind::GoAway(_, _, Initiator::Remote) | Kind::Reset(_, _, Initiator::Remote)
+            Kind::GoAway(_, _, Initiator::Remote) | Kind::Reset(_, _, Initiator::Remote, _)
         )
     }
 
@@ -111,7 +112,7 @@ impl Error {
     pub fn is_library(&self) -> bool {
         matches!(
             self.kind,
-            Kind::GoAway(_, _, Initiator::Library) | Kind::Reset(_, _, Initiator::Library)
+            Kind::GoAway(_, _, Initiator::Library) | Kind::Reset(_, _, Initiator::Library, _)
         )
     }
 }
@@ -122,7 +123,9 @@ impl From<proto::Error> for Error {
 
         Error {
             kind: match src {
-                Reset(stream_id, reason, initiator) => Kind::Reset(stream_id, reason, initiator),
+                Reset(stream_id, reason, initiator, status_code) => {
+                    Kind::Reset(stream_id, reason, initiator, status_code)
+                }
                 GoAway(debug_data, reason, initiator) => {
                     Kind::GoAway(debug_data, reason, initiator)
                 }
@@ -162,13 +165,13 @@ impl From<UserError> for Error {
 impl fmt::Display for Error {
     fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
         let debug_data = match self.kind {
-            Kind::Reset(_, reason, Initiator::User) => {
+            Kind::Reset(_, reason, Initiator::User, _) => {
                 return write!(fmt, "stream error sent by user: {}", reason)
             }
-            Kind::Reset(_, reason, Initiator::Library) => {
+            Kind::Reset(_, reason, Initiator::Library, _) => {
                 return write!(fmt, "stream error detected: {}", reason)
             }
-            Kind::Reset(_, reason, Initiator::Remote) => {
+            Kind::Reset(_, reason, Initiator::Remote, _) => {
                 return write!(fmt, "stream error received: {}", reason)
             }
             Kind::GoAway(ref debug_data, reason, Initiator::User) => {

src/proto/connection.rs

@@ -447,7 +447,7 @@ where
             // Attempting to read a frame resulted in a stream level error.
             // This is handled by resetting the frame then trying to read
             // another frame.
-            Err(Error::Reset(id, reason, initiator)) => {
+            Err(Error::Reset(id, reason, initiator, _)) => {
                 debug_assert_eq!(initiator, Initiator::Library);
                 tracing::trace!(?id, ?reason, "stream error");
                 self.streams.send_reset(id, reason);

src/proto/error.rs

@@ -2,13 +2,14 @@ use crate::codec::SendError;
 use crate::frame::{Reason, StreamId};
 
 use bytes::Bytes;
+use http::StatusCode;
 use std::fmt;
 use std::io;
 
 /// Either an H2 reason  or an I/O error
 #[derive(Clone, Debug)]
 pub enum Error {
-    Reset(StreamId, Reason, Initiator),
+    Reset(StreamId, Reason, Initiator, Option<StatusCode>),
     GoAway(Bytes, Reason, Initiator),
     Io(io::ErrorKind, Option<String>),
 }
@@ -23,7 +24,7 @@ pub enum Initiator {
 impl Error {
     pub(crate) fn is_local(&self) -> bool {
         match *self {
-            Self::Reset(_, _, initiator) | Self::GoAway(_, _, initiator) => initiator.is_local(),
+            Self::Reset(_, _, initiator, _) | Self::GoAway(_, _, initiator) => initiator.is_local(),
             Self::Io(..) => true,
         }
     }
@@ -33,7 +34,15 @@ impl Error {
     }
 
     pub(crate) fn library_reset(stream_id: StreamId, reason: Reason) -> Self {
-        Self::Reset(stream_id, reason, Initiator::Library)
+        Self::Reset(stream_id, reason, Initiator::Library, None)
+    }
+
+    pub(crate) fn library_reset_with_status_code(
+        stream_id: StreamId,
+        reason: Reason,
+        status_code: StatusCode,
+    ) -> Self {
+        Self::Reset(stream_id, reason, Initiator::Library, Some(status_code))
     }
 
     pub(crate) fn library_go_away(reason: Reason) -> Self {
@@ -45,7 +54,7 @@ impl Error {
     }
 
     pub(crate) fn remote_reset(stream_id: StreamId, reason: Reason) -> Self {
-        Self::Reset(stream_id, reason, Initiator::Remote)
+        Self::Reset(stream_id, reason, Initiator::Remote, None)
     }
 
     pub(crate) fn remote_go_away(debug_data: Bytes, reason: Reason) -> Self {
@@ -65,7 +74,7 @@ impl Initiator {
 impl fmt::Display for Error {
     fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
         match *self {
-            Self::Reset(_, reason, _) | Self::GoAway(_, reason, _) => reason.fmt(fmt),
+            Self::Reset(_, reason, _, _) | Self::GoAway(_, reason, _) => reason.fmt(fmt),
             Self::Io(_, Some(ref inner)) => inner.fmt(fmt),
             Self::Io(kind, None) => io::Error::from(kind).fmt(fmt),
         }

src/proto/mod.rs

@@ -11,7 +11,7 @@ pub use self::error::{Error, Initiator};
 pub(crate) use self::peer::{Dyn as DynPeer, Peer};
 pub(crate) use self::ping_pong::UserPings;
 pub(crate) use self::streams::{DynStreams, OpaqueStreamRef, StreamRef, Streams};
-pub(crate) use self::streams::{Open, PollReset, Prioritized};
+pub(crate) use self::streams::{Open, PollReset, Prioritized, SendResetContext};
 
 use crate::codec::Codec;
 

src/proto/streams/mod.rs

@@ -12,7 +12,7 @@ mod streams;
 
 pub(crate) use self::prioritize::Prioritized;
 pub(crate) use self::recv::Open;
-pub(crate) use self::send::PollReset;
+pub(crate) use self::send::{PollReset, SendResetContext};
 pub(crate) use self::streams::{DynStreams, OpaqueStreamRef, StreamRef, Streams};
 
 use self::buffer::Buffer;

src/proto/streams/send.rs

@@ -47,6 +47,33 @@ pub(crate) enum PollReset {
     Streaming,
 }
 
+/// Context for `send_reset`.
+pub(crate) struct SendResetContext {
+    reason: Reason,
+    initiator: Initiator,
+    status_code: Option<http::StatusCode>,
+}
+
+impl SendResetContext {
+    /// Create a new `SendResetContext` with and optional `http::StatusCode`.
+    pub(crate) fn with_status_code(
+        reason: Reason,
+        initiator: Initiator,
+        status_code: Option<http::StatusCode>,
+    ) -> Self {
+        Self {
+            reason,
+            initiator,
+            status_code,
+        }
+    }
+
+    /// Create a new `SendResetContext`
+    pub(crate) fn new(reason: Reason, initiator: Initiator) -> Self {
+        Self::with_status_code(reason, initiator, None)
+    }
+}
+
 impl Send {
     /// Create a new `Send`
     pub fn new(config: &Config) -> Self {
@@ -170,8 +197,7 @@ impl Send {
     /// Send an explicit RST_STREAM frame
     pub fn send_reset<B>(
         &mut self,
-        reason: Reason,
-        initiator: Initiator,
+        context: SendResetContext,
         buffer: &mut Buffer<Frame<B>>,
         stream: &mut store::Ptr,
         counts: &mut Counts,
@@ -182,18 +208,25 @@ impl Send {
         let is_empty = stream.pending_send.is_empty();
         let stream_id = stream.id;
 
+        let SendResetContext {
+            reason,
+            initiator,
+            status_code,
+        } = context;
+
         tracing::trace!(
             "send_reset(..., reason={:?}, initiator={:?}, stream={:?}, ..., \
              is_reset={:?}; is_closed={:?}; pending_send.is_empty={:?}; \
-             state={:?} \
+             state={:?}; status_code={:?} \
              ",
             reason,
             initiator,
             stream_id,
             is_reset,
             is_closed,
             is_empty,
-            stream.state
+            stream.state,
+            status_code
         );
 
         if is_reset {
@@ -225,6 +258,19 @@ impl Send {
         // `reclaim_all_capacity`.
         self.prioritize.clear_queue(buffer, stream);
 
+        // For malformed requests, a server may send an HTTP response prior to resetting the stream.
+        if let Some(status_code) = status_code {
+            tracing::trace!("send_reset -- sending response with status code: {status_code}");
+            let pseudo = frame::Pseudo::response(status_code);
+            let fields = http::HeaderMap::default();
+            let mut frame = frame::Headers::new(stream.id, pseudo, fields);
+            frame.set_end_stream();
+
+            tracing::trace!("send_reset -- queueing response; frame={:?}", frame);
+            self.prioritize
+                .queue_frame(frame.into(), buffer, stream, task);
+        }
+
         let frame = frame::Reset::new(stream.id, reason);
 
         tracing::trace!("send_reset -- queueing; frame={:?}", frame);
@@ -378,8 +424,7 @@ impl Send {
             tracing::debug!("recv_stream_window_update !!; err={:?}", e);
 
             self.send_reset(
-                Reason::FLOW_CONTROL_ERROR,
-                Initiator::Library,
+                SendResetContext::new(Reason::FLOW_CONTROL_ERROR, Initiator::Library),
                 buffer,
                 stream,
                 counts,

src/proto/streams/state.rs

@@ -333,7 +333,9 @@ impl State {
 
     /// Set the stream state to reset locally.
     pub fn set_reset(&mut self, stream_id: StreamId, reason: Reason, initiator: Initiator) {
-        self.inner = Closed(Cause::Error(Error::Reset(stream_id, reason, initiator)));
+        self.inner = Closed(Cause::Error(Error::Reset(
+            stream_id, reason, initiator, None,
+        )));
     }
 
     /// Set the stream state to a scheduled reset.
@@ -364,7 +366,7 @@ impl State {
     pub fn is_remote_reset(&self) -> bool {
         matches!(
             self.inner,
-            Closed(Cause::Error(Error::Reset(_, _, Initiator::Remote)))
+            Closed(Cause::Error(Error::Reset(_, _, Initiator::Remote, _)))
         )
     }
 
@@ -446,7 +448,7 @@ impl State {
     /// Returns a reason if the stream has been reset.
     pub(super) fn ensure_reason(&self, mode: PollReset) -> Result<Option<Reason>, crate::Error> {
         match self.inner {
-            Closed(Cause::Error(Error::Reset(_, reason, _)))
+            Closed(Cause::Error(Error::Reset(_, reason, _, _)))
             | Closed(Cause::Error(Error::GoAway(_, reason, _)))
             | Closed(Cause::ScheduledLibraryReset(reason)) => Ok(Some(reason)),
             Closed(Cause::Error(ref e)) => Err(e.clone().into()),

src/proto/streams/streams.rs

@@ -1537,8 +1537,7 @@ impl Actions {
     ) {
         counts.transition(stream, |counts, stream| {
             self.send.send_reset(
-                reason,
-                initiator,
+                proto::SendResetContext::new(reason, initiator),
                 send_buffer,
                 stream,
                 counts,
@@ -1557,15 +1556,20 @@ impl Actions {
         counts: &mut Counts,
         res: Result<(), Error>,
     ) -> Result<(), Error> {
-        if let Err(Error::Reset(stream_id, reason, initiator)) = res {
+        if let Err(Error::Reset(stream_id, reason, initiator, status_code)) = res {
             debug_assert_eq!(stream_id, stream.id);
 
             if counts.can_inc_num_local_error_resets() {
                 counts.inc_num_local_error_resets();
 
                 // Reset the stream.
-                self.send
-                    .send_reset(reason, initiator, buffer, stream, counts, &mut self.task);
+                self.send.send_reset(
+                    proto::SendResetContext::with_status_code(reason, initiator, status_code),
+                    buffer,
+                    stream,
+                    counts,
+                    &mut self.task,
+                );
                 Ok(())
             } else {
                 tracing::warn!(

src/server.rs

@@ -121,7 +121,7 @@ use crate::proto::{self, Config, Error, Prioritized};
 use crate::{FlowControl, PingPong, RecvStream, SendStream};
 
 use bytes::{Buf, Bytes};
-use http::{HeaderMap, Method, Request, Response};
+use http::{HeaderMap, Method, Request, Response, StatusCode};
 use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
@@ -1515,7 +1515,7 @@ impl proto::Peer for Peer {
         macro_rules! malformed {
             ($($arg:tt)*) => {{
                 tracing::debug!($($arg)*);
-                return Err(Error::library_reset(stream_id, Reason::PROTOCOL_ERROR));
+                return Err(Error::library_reset_with_status_code(stream_id, Reason::PROTOCOL_ERROR, StatusCode::BAD_REQUEST));
             }}
         }
 

tests/h2-tests/tests/server.rs

@@ -504,6 +504,9 @@ async fn recv_invalid_authority() {
         let settings = client.assert_server_handshake().await;
         assert_default_settings!(settings);
         client.send_frame(bad_headers).await;
+        client
+            .recv_frame(frames::headers(1).status(StatusCode::BAD_REQUEST).eos())
+            .await;
         client.recv_frame(frames::reset(1).protocol_error()).await;
     };
 
@@ -1290,6 +1293,9 @@ async fn reject_pseudo_protocol_on_non_connect_request() {
             )
             .await;
 
+        client
+            .recv_frame(frames::headers(1).status(StatusCode::BAD_REQUEST).eos())
+            .await;
         client.recv_frame(frames::reset(1).protocol_error()).await;
     };
 
@@ -1329,6 +1335,9 @@ async fn reject_authority_target_on_extended_connect_request() {
             )
             .await;
 
+        client
+            .recv_frame(frames::headers(1).status(StatusCode::BAD_REQUEST).eos())
+            .await;
         client.recv_frame(frames::reset(1).protocol_error()).await;
     };
 
@@ -1364,6 +1373,9 @@ async fn reject_non_authority_target_on_connect_request() {
             .send_frame(frames::headers(1).request("CONNECT", "https://bread/baguette"))
             .await;
 
+        client
+            .recv_frame(frames::headers(1).status(StatusCode::BAD_REQUEST).eos())
+            .await;
         client.recv_frame(frames::reset(1).protocol_error()).await;
     };
 

tests/h2-tests/tests/stream_states.rs

@@ -524,6 +524,9 @@ async fn recv_next_stream_id_updated_by_malformed_headers() {
         assert_default_settings!(settings);
         // bad headers -- should error.
         client.send_frame(bad_headers).await;
+        client
+            .recv_frame(frames::headers(1).status(StatusCode::BAD_REQUEST).eos())
+            .await;
         client.recv_frame(frames::reset(1).protocol_error()).await;
         // this frame is good, but the stream id should already have been incr'd
         client