feat: allow empty HTTP/2 DATA frames to be throttled

http-core/src/main/resources/reference.conf

@@ -332,8 +332,14 @@ pekko.http {
       frame-type-throttle {
         # Configure the throttle for non-data frame types (https://github.com/apache/pekko-http/issues/332).
         # The supported frame-types for throttling are:
-        # reset, headers, continuation, go-away, priority, ping, push-promise, window-update
+        # reset, headers, continuation, go-away, priority, ping, push-promise, window-update, empty-data
         # By default, RST_STREAM frames are throttled to mitigate HTTP/2 Rapid Reset attacks (CVE-2023-44487).
+        #
+        # "empty-data" covers DATA frames that carry no payload. Those consume no flow-control window, so unlike
+        # data-carrying frames their number is not bounded by flow control and a peer can send them continuously.
+        # Only the empty ones can be throttled: doing the same for data-carrying frames would throttle legitimate
+        # throughput along with them.
+        #
         # Set to [] to disable throttling.
         frame-types = ["reset"]
         cost = 100

http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2Blueprint.scala

@@ -206,9 +206,15 @@ private[http] object Http2Blueprint {
 
   private def rapidResetMitigation(settings: Http2ServerSettings,
       frameTypesForThrottle: Set[String]): BidiFlow[FrameEvent, FrameEvent, FrameEvent, FrameEvent, NotUsed] = {
-    def frameCost(event: FrameEvent): Int = {
-      if (frameTypesForThrottle.contains(event.frameTypeName)) 1 else 0
-    }
+    def frameCost(event: FrameEvent): Int =
+      if (frameTypesForThrottle.contains(event.frameTypeName)) 1
+      else event match {
+        // A DATA frame with no payload consumes no flow-control window, so unlike a data-carrying one its number is
+        // not bounded by flow control at all. It is matched separately because throttling every DATA frame would
+        // throttle legitimate throughput along with it.
+        case d: DataFrame if d.payload.isEmpty && frameTypesForThrottle.contains(EmptyDataFrameThrottleName) => 1
+        case _                                                                                               => 0
+      }
 
     BidiFlow.fromFlows(
       Flow[FrameEvent],
@@ -225,8 +231,15 @@ private[http] object Http2Blueprint {
     }
   }
 
+  /**
+   * Not a real `frameTypeName`, so it never matches one directly: `frameCost` recognises it and charges DATA frames
+   * that carry no payload.
+   */
+  private[http2] val EmptyDataFrameThrottleName = "EmptyDataFrame"
+
   private[http2] def frameTypeAliasToFrameTypeName(frameType: String): Option[String] = {
     toRootLowerCase(frameType) match {
+      case "empty-data"    => Some(EmptyDataFrameThrottleName)
       case "reset"         => Some("RstStreamFrame")
       case "headers"       => Some("HeadersFrame")
       case "continuation"  => Some("ContinuationFrame")

http-core/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2BlueprintSpec.scala

@@ -26,6 +26,10 @@ import org.scalatest.wordspec.AnyWordSpec
 
 class Http2BlueprintSpec extends AnyWordSpec with Matchers {
   "Http2Blueprint" should {
+    "match frame type alias (empty-data)" in {
+      Http2Blueprint.frameTypeAliasToFrameTypeName("empty-data") shouldEqual
+      Some(Http2Blueprint.EmptyDataFrameThrottleName)
+    }
     "match frame type alias (reset)" in {
       Http2Blueprint.frameTypeAliasToFrameTypeName("reset") shouldEqual
       Some(RstStreamFrame(0, ErrorCode.PROTOCOL_ERROR).frameTypeName)

http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ServerEmptyDataThrottleSpec.scala

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.http.impl.engine.http2
+
+import org.apache.pekko
+import pekko.http.impl.engine.http2.Http2Protocol.FrameType
+import pekko.http.impl.engine.http2.framing.FrameRenderer
+import pekko.util.ByteString
+
+/**
+ * Tests that DATA frames carrying no payload can be throttled. They consume no flow-control window, so unlike
+ * data-carrying frames their number is not bounded by flow control.
+ */
+class Http2ServerEmptyDataThrottleSpec extends Http2SpecWithMaterializer("""
+    pekko.http.server.http2.log-frames = on
+    pekko.http.server.http2.frame-type-throttle.frame-types = ["empty-data"]
+  """) {
+  override val expectSevereLogsOnlyToMatch: Option[String] = Some(
+    "HTTP2 connection failed with error [Maximum throttle throughput exceeded.]. Sending INTERNAL_ERROR and closing connection.")
+
+  "The Http/2 server implementation" should {
+    "cancel connection when flooded with empty DATA frames".inAssertAllStagesStopped(
+      new TestSetup with RequestResponseProbes {
+        val emptyDataFrame = FrameRenderer.renderFrame(FrameType.DATA, ByteFlag.Zero, 1, ByteString.empty)
+        network.sendBytes(Seq.fill(1000)(emptyDataFrame).reduce(_ ++ _))
+      })
+  }
+}