Repository: bearcove/loona (language: rust)

Issue:
h2: Send GOAWAY (connection error) on HPACK errors
```
just h2spec hpack/6.1
(cut)
Failures: 

HPACK: Header Compression for HTTP/2
  6. Binary Format
    6.1. Indexed Header Field Representation
      × 1: Sends a indexed header field representation with index 0
        -> The endpoint MUST treat this as a decoding error.
           Expected: GOAWAY Frame (Error Code: COMPRESSION_ERROR)
                     Connection closed
             Actual: Timeout

Finished in 1.0043 seconds
1 tests, 0 passed, 0 skipped, 1 failed
```


Discussion / hints:

Fix patch:
diff --git a/src/h2/encode.rs b/src/h2/encode.rs
index ea028cb2..ab99d0f6 100644
--- a/src/h2/encode.rs
+++ b/src/h2/encode.rs
@@ -6,13 +6,18 @@ use tracing::warn;
 
 use crate::{h1::body::BodyWriteMode, Encoder, Piece, Response, Roll};
 
-use super::parse::{Frame, StreamId};
+use super::parse::{Frame, KnownErrorCode, StreamId};
 
 pub(crate) enum H2ConnEvent {
     Ping(Roll),
     ClientFrame(Frame, Option<Roll>),
     ServerEvent(H2Event),
     AcknowledgeSettings,
+    GoAway {
+        error_code: KnownErrorCode,
+        last_stream_id: StreamId,
+        additional_debug_data: Piece,
+    },
 }
 
 #[derive(Debug)]
diff --git a/src/h2/parse.rs b/src/h2/parse.rs
index 79d3646d..9ca02a9b 100644
--- a/src/h2/parse.rs
+++ b/src/h2/parse.rs
@@ -28,16 +28,16 @@ pub fn preface(i: Roll) -> IResult<Roll, ()> {
 #[EnumRepr(type = "u8")]
 #[derive(Debug, Clone, Copy)]
 pub enum RawFrameType {
-    Data = 0,
-    Headers = 1,
-    Priority = 2,
-    RstStream = 3,
-    Settings = 4,
-    PushPromise = 5,
-    Ping = 6,
-    GoAway = 7,
-    WindowUpdate = 8,
-    Continuation = 9,
+    Data = 0x00,
+    Headers = 0x01,
+    Priority = 0x02,
+    RstStream = 0x03,
+    Settings = 0x04,
+    PushPromise = 0x05,
+    Ping = 0x06,
+    GoAway = 0x07,
+    WindowUpdate = 0x08,
+    Continuation = 0x09,
 }
 
 /// Typed flags for various frame types
@@ -158,8 +158,8 @@ impl FrameType {
     }
 }
 
-#[derive(Clone, Copy, PartialEq, Eq, Hash)]
-pub struct StreamId(u32);
+#[derive(Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
+pub struct StreamId(pub(crate) u32);
 
 impl StreamId {
     /// Stream ID used for connection control frames
@@ -241,6 +241,8 @@ impl Frame {
 
     /// Write a frame (without payload)
     pub async fn write(&self, w: &impl WriteOwned) -> eyre::Result<()> {
+        // TODO: we need an owned type here, but do we need to re-allocate it
+        // here? can we have a pool of them?
         let mut header = vec![0u8; 9];
         {
             use byteorder::{BigEndian, WriteBytesExt};
@@ -294,3 +296,89 @@ impl PrioritySpec {
         )(i)
     }
 }
+
+#[derive(Clone, Copy)]
+pub struct ErrorCode(u32);
+
+impl fmt::Debug for ErrorCode {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match KnownErrorCode::from_repr(self.0) {
+            Some(e) => fmt::Debug::fmt(&e, f),
+            None => write!(f, "ErrorCode(0x{:02x})", self.0),
+        }
+    }
+}
+
+impl From<KnownErrorCode> for ErrorCode {
+    fn from(e: KnownErrorCode) -> Self {
+        Self(e as u32)
+    }
+}
+
+#[EnumRepr(type = "u32")]
+#[derive(Debug, Clone, Copy)]
+pub enum KnownErrorCode {
+    /// The associated condition is not a result of an error. For example, a
+    /// GOAWAY might include this code to indicate graceful shutdown of a
+    /// connection.
+    NoError = 0x00,
+
+    /// The endpoint detected an unspecific protocol error. This error is for
+    /// use when a more specific error code is not available.
+    ProtocolError = 0x01,
+
+    /// The endpoint encountered an unexpected internal error.
+    InternalError = 0x02,
+
+    /// The endpoint detected that its peer violated the flow-control protocol.
+    FlowControlError = 0x03,
+
+    /// The endpoint sent a SETTINGS frame but did not receive a response in a
+    /// timely manner. See Section 6.5.3 ("Settings Synchronization").
+    /// https://httpwg.org/specs/rfc9113.html#SettingsSync
+    SettingsTimeout = 0x04,
+
+    /// The endpoint received a frame after a stream was half-closed.
+    StreamClosed = 0x05,
+
+    /// The endpoint received a frame with an invalid size.
+    FrameSizeError = 0x06,
+
+    /// The endpoint refused the stream prior to performing any application
+    /// processing (see Section 8.7 for details).
+    /// https://httpwg.org/specs/rfc9113.html#Reliability
+    RefusedStream = 0x07,
+
+    /// The endpoint uses this error code to indicate that the stream is no
+    /// longer needed.
+    Cancel = 0x08,
+
+    /// The endpoint is unable to maintain the field section compression context
+    /// for the connection.
+    CompressionError = 0x09,
+
+    /// The connection established in response to a CONNECT request (Section
+    /// 8.5) was reset or abnormally closed.
+    /// https://httpwg.org/specs/rfc9113.html#CONNECT
+    ConnectError = 0x0a,
+
+    /// The endpoint detected that its peer is exhibiting a behavior that might
+    /// be generating excessive load.
+    EnhanceYourCalm = 0x0b,
+
+    /// The underlying transport has properties that do not meet minimum
+    /// security requirements (see Section 9.2).
+    /// https://httpwg.org/specs/rfc9113.html#TLSUsage
+    InadequateSecurity = 0x0c,
+
+    /// The endpoint requires that HTTP/1.1 be used instead of HTTP/2.
+    Http1_1Required = 0x0d,
+}
+
+impl TryFrom<ErrorCode> for KnownErrorCode {
+    type Error = ();
+
+    fn try_from(e: ErrorCode) -> Result<Self, Self::Error> {
+        KnownErrorCode::from_repr(e.0).ok_or(())
+    }
+}
diff --git a/src/h2/server.rs b/src/h2/server.rs
index 00e1dfdc..f228d915 100644
--- a/src/h2/server.rs
+++ b/src/h2/server.rs
@@ -5,15 +5,15 @@ use http::{header::HeaderName, Version};
 use nom::Finish;
 use pretty_hex::PrettyHex;
 use tokio::sync::mpsc;
-use tracing::{debug, trace};
+use tracing::{debug, trace, warn};
 
 use crate::{
     h2::{
         body::H2Body,
         encode::{EncoderState, H2ConnEvent, H2Encoder, H2EventPayload},
         parse::{
-            self, DataFlags, Frame, FrameType, HeadersFlags, PingFlags, PrioritySpec,
-            SettingsFlags, StreamId,
+            self, DataFlags, Frame, FrameType, HeadersFlags, KnownErrorCode, PingFlags,
+            PrioritySpec, SettingsFlags, StreamId,
         },
     },
     util::read_and_parse,
@@ -100,6 +100,8 @@ pub async fn serve(
 
     let write_task = h2_write_loop(driver, ev_tx, ev_rx, transport, state);
     tokio::try_join!(read_task, write_task)?;
+    debug!("joined read_task / write_task");
+
     Ok(())
 }
 
@@ -256,6 +258,7 @@ async fn h2_write_loop(
     let mut hpack_enc = hpack::Encoder::new();
 
     while let Some(ev) = ev_rx.recv().await {
+        trace!("h2_write_loop: received H2 event");
         match ev {
             H2ConnEvent::AcknowledgeSettings => {
                 debug!("acknowledging new settings");
@@ -296,45 +299,74 @@ async fn h2_write_loop(
                     debug!("receiving headers for stream {}", frame.stream_id);
 
                     if let Some(payload) = payload {
-                        hpack_dec
-                            .decode_with_cb(&payload[..], |key, value| {
-                                debug!(
-                                    "got key: {:?}\nvalue: {:?}",
-                                    key.hex_dump(),
-                                    value.hex_dump()
-                                );
-
-                                if &key[..1] == b":" {
-                                    // it's a pseudo-header!
-                                    match &key[1..] {
-                                        b"path" => {
-                                            let value: PieceStr =
-                                                Piece::from(value.to_vec()).to_string().unwrap();
-                                            path = Some(value);
-                                        }
-                                        b"method" => {
-                                            // TODO: error handling
-                                            let value: PieceStr =
-                                                Piece::from(value.to_vec()).to_string().unwrap();
-                                            method = Some(Method::try_from(value).unwrap());
-                                        }
-                                        other => {
-                                            debug!("ignoring pseudo-header {:?}", other.hex_dump());
-                                        }
+                        if let Err(e) = hpack_dec.decode_with_cb(&payload[..], |key, value| {
+                            debug!(
+                                "got key: {:?}\nvalue: {:?}",
+                                key.hex_dump(),
+                                value.hex_dump()
+                            );
+
+                            if &key[..1] == b":" {
+                                // it's a pseudo-header!
+                                match &key[1..] {
+                                    b"path" => {
+                                        let value: PieceStr =
+                                            Piece::from(value.to_vec()).to_string().unwrap();
+                                        path = Some(value);
+                                    }
+                                    b"method" => {
+                                        // TODO: error handling
+                                        let value: PieceStr =
+                                            Piece::from(value.to_vec()).to_string().unwrap();
+                                        method = Some(Method::try_from(value).unwrap());
+                                    }
+                                    other => {
+                                        debug!("ignoring pseudo-header {:?}", other.hex_dump());
                                     }
-                                } else {
-                                    // TODO: what do we do in case of malformed header names?
-                                    // ignore it? return a 400?
-                                    let name = HeaderName::from_bytes(&key[..])
-                                        .expect("malformed header name");
-                                    let value: Piece = value.to_vec().into();
-                                    headers.append(name, value);
                                 }
-                            })
-                            .map_err(|e| eyre::eyre!("hpack error: {e:?}"))?;
-
-                        // TODO: hpack errors are COMPRESSION_ERROR, cf.
-                        // https://httpwg.org/specs/rfc9113.html#ErrorCodes
+                            } else {
+                                // TODO: what do we do in case of malformed header names?
+                                // ignore it? return a 400?
+                                let name = HeaderName::from_bytes(&key[..])
+                                    .expect("malformed header name");
+                                let value: Piece = value.to_vec().into();
+                                headers.append(name, value);
+                            }
+                        }) {
+                            warn!("hpack error: {e:?}");
+                            let error_code = KnownErrorCode::CompressionError;
+                            debug!("error_code = {error_code:?}");
+                            // FIXME: this is almost definitely wrong. we must separate
+                            // client-initiated streams from server-initiated streams, and
+                            // take stream state into account
+                            let last_stream_id = state
+                                .borrow()
+                                .streams
+                                .keys()
+                                .copied()
+                                .max()
+                                .unwrap_or(StreamId::CONNECTION);
+                            debug!("last_stream_id = {last_stream_id}");
+                            // TODO: is this a good idea?
+                            let additional_debug_data = format!("hpack error: {e:?}").into_bytes();
+
+                            if ev_tx
+                                .send(H2ConnEvent::GoAway {
+                                    error_code,
+                                    last_stream_id,
+                                    additional_debug_data: Piece::Vec(additional_debug_data),
+                                })
+                                .await
+                                .is_err()
+                            {
+                                debug!("error sending goaway");
+                            }
+                            // TODO: move most processing to `h2_read_loop`?
+                            // It's confusing to be doing parsing here.  and we
+                            // can't return because we just sent something to
+                            // `ev_tx` which _we're the ones reading_
+                            continue;
+                        }
                     }
 
                     // TODO: proper error handling (return 400)
@@ -501,7 +533,36 @@ async fn h2_write_loop(
                 let (res, _) = transport.write_all(payload).await;
                 res?;
             }
+            H2ConnEvent::GoAway {
+                error_code,
+                last_stream_id,
+                additional_debug_data,
+            } => {
+                debug!("must send goaway");
+                let mut header = vec![0u8; 8];
+                {
+                    use byteorder::{BigEndian, WriteBytesExt};
+                    let mut header = &mut header[..];
+                    // TODO: do we ever need to write the reserved bit?
+                    header.write_u32::<BigEndian>(last_stream_id.0)?;
+                    header.write_u32::<BigEndian>(error_code.repr())?;
+                }
+
+                debug!("sending goaway frame");
+                let frame = Frame::new(FrameType::GoAway, StreamId::CONNECTION).with_len(
+                    (header.len() + additional_debug_data.len())
+                        .try_into()
+                        .unwrap(),
+                );
+                frame.write(transport.as_ref()).await?;
+                let (res, _) = transport.write_all(header).await;
+                res?;
+                let (res, _) = transport.write_all(additional_debug_data).await;
+                res?;
+            }
         }
     }
+    debug!("h2_write_loop is done");
+
     Ok(())
 }


Test patch: