Harden HTTP/2 client response flow control for large bodies

crates/rttp-client/src/http2.rs

@@ -369,9 +369,12 @@ fn reject_goaway_before_opening_request_stream(
       (FRAME_PING, _) => {
         return Err(error::bad_response("invalid HTTP/2 PING frame"));
       }
-      (FRAME_WINDOW_UPDATE, _) => {
+      (FRAME_WINDOW_UPDATE, 0) => {
         window_update_increment(&frame)?;
       }
+      (FRAME_WINDOW_UPDATE, _) => {
+        return Err(idle_window_update_error());
+      }
       (FRAME_PRIORITY, _) => {
         validate_priority_frame(&frame)?;
       }
@@ -386,6 +389,9 @@ fn reject_goaway_before_opening_request_stream(
           "unexpected HTTP/2 CONTINUATION frame without header block",
         ));
       }
+      (FRAME_DATA, 0) => {
+        return Err(connection_data_error());
+      }
       (_, 0) => {}
       _ => {
         return Err(error::bad_response(
@@ -1061,14 +1067,13 @@ fn read_until_send_window_available(
   loop {
     let frame = read_frame(stream, local_settings)?;
     match (frame.frame_type, frame.stream_id) {
-      (FRAME_WINDOW_UPDATE, 0) => {
-        connection_send_window.increase(window_update_increment(&frame)?)?;
-      }
-      (FRAME_WINDOW_UPDATE, id) if id == stream_id => {
-        stream_send_window.increase(window_update_increment(&frame)?)?;
-      }
       (FRAME_WINDOW_UPDATE, _) => {
-        window_update_increment(&frame)?;
+        apply_send_window_update(
+          &frame,
+          connection_send_window,
+          stream_send_window,
+          stream_id,
+        )?;
       }
       (FRAME_SETTINGS, _) => {
         handle_settings_while_sending(
@@ -1121,6 +1126,9 @@ fn read_until_send_window_available(
         )
         .map(Some);
       }
+      (FRAME_DATA, 0) => {
+        return Err(connection_data_error());
+      }
       _ => {}
     }
 
@@ -1581,14 +1589,13 @@ fn read_single_stream_response_with_first_frame(
           break;
         }
       }
-      (FRAME_WINDOW_UPDATE, 0) => {
-        connection_send_window.increase(window_update_increment(&frame)?)?;
-      }
-      (FRAME_WINDOW_UPDATE, id) if id == stream_id => {
-        stream_send_window.increase(window_update_increment(&frame)?)?;
-      }
       (FRAME_WINDOW_UPDATE, _) => {
-        window_update_increment(&frame)?;
+        apply_send_window_update(
+          &frame,
+          &mut connection_send_window,
+          &mut stream_send_window,
+          stream_id,
+        )?;
       }
       (FRAME_PRIORITY, _) => {
         validate_priority_frame(&frame)?;
@@ -1630,6 +1637,9 @@ fn read_single_stream_response_with_first_frame(
           "unexpected HTTP/2 CONTINUATION frame without header block",
         ));
       }
+      (FRAME_DATA, 0) => {
+        return Err(connection_data_error());
+      }
       (_, 0) => {}
       _ => {}
     }
@@ -1770,6 +1780,30 @@ fn window_update_increment(frame: &Frame) -> error::Result<u32> {
   Ok(increment)
 }
 
+fn apply_send_window_update(
+  frame: &Frame,
+  connection_send_window: &mut SendWindow,
+  stream_send_window: &mut SendWindow,
+  stream_id: u32,
+) -> error::Result<()> {
+  let increment = window_update_increment(frame)?;
+  if frame.stream_id == 0 {
+    connection_send_window.increase(increment)
+  } else if frame.stream_id == stream_id {
+    stream_send_window.increase(increment)
+  } else {
+    Err(idle_window_update_error())
+  }
+}
+
+fn idle_window_update_error() -> error::Error {
+  error::bad_response("HTTP/2 WINDOW_UPDATE for idle stream")
+}
+
+fn connection_data_error() -> error::Error {
+  error::bad_response("invalid HTTP/2 DATA frame on stream 0")
+}
+
 fn validate_priority_frame(frame: &Frame) -> error::Result<()> {
   if frame.stream_id == 0 || frame.payload.len() != 5 {
     return Err(error::bad_response("invalid HTTP/2 PRIORITY frame"));

crates/rttp-client/tests/http2_prior_knowledge.rs

@@ -4323,6 +4323,7 @@ fn prior_knowledge_rejects_invalid_window_update_frames() {
     ("stream-zero", 1, 0, "WINDOW_UPDATE"),
     ("connection-overflow", 0, 0x7fff_ffff, "overflow"),
     ("stream-overflow", 1, 0x7fff_ffff, "overflow"),
+    ("idle-stream", 3, 1, "idle stream"),
   ] {
     let (addr, handle) = spawn_window_update_peer(stream_id, vec![increment]);
     let error = HttpClient::new()
@@ -4338,6 +4339,90 @@ fn prior_knowledge_rejects_invalid_window_update_frames() {
   }
 }
 
+#[test]
+fn prior_knowledge_rejects_window_update_for_other_streams_during_response() {
+  let listener = TcpListener::bind("127.0.0.1:0").expect("bind h2 peer");
+  let addr = listener.local_addr().expect("h2 peer addr");
+
+  let handle = thread::spawn(move || {
+    let (mut stream, _) = listener.accept().expect("accept h2 client");
+    complete_h2_request_handshake(&mut stream);
+    write_frame(&mut stream, FRAME_HEADERS, FLAG_END_HEADERS, 1, &[0x88]);
+    write_frame(&mut stream, FRAME_WINDOW_UPDATE, 0, 3, &1_u32.to_be_bytes());
+    write_frame(&mut stream, FRAME_DATA, FLAG_END_STREAM, 1, b"ignored");
+  });
+
+  let error = HttpClient::new()
+    .get()
+    .url(format!("http://{addr}/other-stream-window-update"))
+    .emit_http2_prior_knowledge()
+    .expect_err("WINDOW_UPDATE for another stream must fail");
+  assert!(
+    error.to_string().contains("idle stream"),
+    "unexpected other-stream WINDOW_UPDATE error: {error}"
+  );
+  handle
+    .join()
+    .expect("other-stream window update peer thread");
+}
+
+#[test]
+fn prior_knowledge_rejects_connection_level_data_frames() {
+  let listener = TcpListener::bind("127.0.0.1:0").expect("bind h2 peer");
+  let addr = listener.local_addr().expect("h2 peer addr");
+
+  let handle = thread::spawn(move || {
+    let (mut stream, _) = listener.accept().expect("accept h2 client");
+    complete_h2_request_handshake(&mut stream);
+    write_frame(&mut stream, FRAME_HEADERS, FLAG_END_HEADERS, 1, &[0x88]);
+    write_frame(&mut stream, FRAME_DATA, 0, 0, b"connection-data");
+    write_frame(&mut stream, FRAME_DATA, FLAG_END_STREAM, 1, b"ignored");
+  });
+
+  let error = HttpClient::new()
+    .get()
+    .url(format!("http://{addr}/connection-data"))
+    .emit_http2_prior_knowledge()
+    .expect_err("DATA on stream 0 must fail");
+  assert!(
+    error.to_string().contains("DATA frame on stream 0"),
+    "unexpected connection-level DATA error: {error}"
+  );
+  handle.join().expect("connection-level DATA peer thread");
+}
+
+#[test]
+fn prior_knowledge_rejects_data_exceeding_flow_control_window() {
+  let listener = TcpListener::bind("127.0.0.1:0").expect("bind h2 peer");
+  let addr = listener.local_addr().expect("h2 peer addr");
+
+  let handle = thread::spawn(move || {
+    let (mut stream, _) = listener.accept().expect("accept h2 client");
+    complete_h2_request_handshake(&mut stream);
+    write_frame(&mut stream, FRAME_HEADERS, FLAG_END_HEADERS, 1, &[0x88]);
+    write_frame(
+      &mut stream,
+      FRAME_DATA,
+      FLAG_END_STREAM,
+      1,
+      &vec![b'x'; 65_536],
+    );
+  });
+
+  let error = HttpClient::new()
+    .get()
+    .url(format!("http://{addr}/over-window"))
+    .emit_http2_prior_knowledge()
+    .expect_err("DATA larger than the 65535 window must fail");
+  assert!(
+    error
+      .to_string()
+      .contains("DATA frame exceeds flow-control window"),
+    "unexpected flow-control overflow error: {error}"
+  );
+  handle.join().expect("over-window DATA peer thread");
+}
+
 #[test]
 fn prior_knowledge_acks_ping_before_consuming_final_response() {
   let listener = TcpListener::bind("127.0.0.1:0").expect("bind h2 peer");

tests/http2_client_server_roundtrip.rs

@@ -53,6 +53,54 @@ fn bounded_h2c_prior_knowledge_round_trip_reaches_the_server() {
   handle.join().expect("h2c server thread");
 }
 
+#[test]
+fn bounded_h2c_prior_knowledge_round_trip_receives_response_body_larger_than_initial_window() {
+  let server = HttpServer::bind("127.0.0.1:0")
+    .expect("bind h2c server")
+    .with_read_timeout(Some(Duration::from_secs(2)))
+    .with_write_timeout(Some(Duration::from_secs(2)));
+  let addr = server.local_addr().expect("h2c server address");
+  let response_body = (0..65_535 + 4_096)
+    .map(|idx| b'0' + (idx % 10) as u8)
+    .collect::<Vec<_>>();
+  let expected_body = response_body.clone();
+  let (tx, rx) = mpsc::channel();
+
+  let handle = thread::spawn(move || {
+    server
+      .accept_one(move |request| {
+        tx.send((
+          request.version().to_string(),
+          request.method().to_string(),
+          request.target().to_string(),
+        ))
+        .expect("record large h2c request");
+        HttpResponse::ok(response_body)
+      })
+      .expect("serve large h2c response");
+  });
+
+  let response = HttpClient::new()
+    .get()
+    .url(format!("http://{addr}/workspace/h2c-large-body"))
+    .emit_http2_prior_knowledge()
+    .expect("receive large h2c response");
+
+  assert_eq!(
+    (
+      "HTTP/2".to_string(),
+      "GET".to_string(),
+      "/workspace/h2c-large-body".to_string()
+    ),
+    rx.recv_timeout(Duration::from_secs(2))
+      .expect("recorded large h2c request")
+  );
+  assert_eq!("HTTP/2", response.version());
+  assert_eq!(expected_body.len(), response.body().binary().len());
+  assert_eq!(expected_body.as_slice(), response.body().binary());
+  handle.join().expect("large h2c server thread");
+}
+
 #[test]
 fn h2c_prior_knowledge_round_trip_preserves_accept_charset_metadata() {
   let server = HttpServer::bind("127.0.0.1:0")