feat: throttle empty HTTP/2 DATA frames that do not end their stream

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

@@ -332,10 +332,25 @@ 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
-        # By default, RST_STREAM frames are throttled to mitigate HTTP/2 Rapid Reset attacks (CVE-2023-44487).
+        # reset, headers, continuation, go-away, priority, ping, push-promise, window-update,
+        # empty-data-no-end-stream, empty-data
+        # By default, RST_STREAM frames are throttled to mitigate HTTP/2 Rapid Reset attacks (CVE-2023-44487), as
+        # are empty DATA frames that do not end their stream.
+        #
+        # A DATA frame that carries no payload consumes no flow-control window, so unlike a data-carrying frame its
+        # number is not bounded by flow control at all and a peer can send them continuously. Two aliases cover them:
+        #  - "empty-data-no-end-stream" charges only the ones that do not carry END_STREAM. A peer has no reason to
+        #    send those at all, so this is throttled by default.
+        #  - "empty-data" charges every empty DATA frame, including the END_STREAM one that a client sends to close a
+        #    request body whose length it did not know up front - that is one frame per such request, so enabling this
+        #    can throttle legitimate traffic on a busy connection. Off by default.
+        # Data-carrying DATA frames cannot be throttled here, because doing so would throttle legitimate throughput.
+        #
+        # Note that all throttled frame types share one budget, so adding a frame type that legitimate traffic also
+        # produces reduces the headroom left for the others.
+        #
         # Set to [] to disable throttling.
-        frame-types = ["reset"]
+        frame-types = ["reset", "empty-data-no-end-stream"]
         cost = 100
         burst = 100
         # interval must be a positive duration

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

@@ -205,16 +205,11 @@ private[http] object Http2Blueprint {
       Flow[ByteString].via(new Http2FrameParsing(shouldReadPreface = false, log)))
 
   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
-    }
-
+      frameTypesForThrottle: Set[String]): BidiFlow[FrameEvent, FrameEvent, FrameEvent, FrameEvent, NotUsed] =
     BidiFlow.fromFlows(
       Flow[FrameEvent],
       Flow[FrameEvent].throttle(settings.frameTypeThrottleCost, settings.frameTypeThrottleInterval,
-        settings.frameTypeThrottleBurst, frameCost, ThrottleMode.Enforcing))
-  }
+        settings.frameTypeThrottleBurst, frameCost(frameTypesForThrottle, _), ThrottleMode.Enforcing))
 
   private def getFrameTypesForThrottle(settings: Http2ServerSettings): Set[String] = {
     val set = settings.frameTypeThrottleFrameTypes
@@ -225,17 +220,48 @@ private[http] object Http2Blueprint {
     }
   }
 
+  /**
+   * Not a real `frameTypeName`, so it never matches one directly: [[frameCost]] recognises it and charges every DATA
+   * frame that carries no payload.
+   *
+   * Such a frame consumes no flow-control window, so unlike a data-carrying one its number is not bounded by flow
+   * control at all and a peer can send them continuously. Data-carrying frames are deliberately not covered, because
+   * throttling those would throttle legitimate throughput along with them.
+   */
+  private[http2] val EmptyDataFrameThrottleName = "EmptyDataFrame"
+
+  /**
+   * As [[EmptyDataFrameThrottleName]], but only for the empty DATA frames that do not carry END_STREAM. An empty
+   * DATA frame with END_STREAM is how a client closes a request body whose length it did not know up front, so it is
+   * both legitimate and self-limiting: one per stream, after which the stream is half-closed. An empty DATA frame
+   * that does not end its stream has no such use, which is why this is the alias that is throttled by default.
+   */
+  private[http2] val EmptyDataFrameNoEndStreamThrottleName = "EmptyDataFrameNoEndStream"
+
+  private[http2] def frameCost(frameTypesForThrottle: Set[String], event: FrameEvent): Int =
+    if (frameTypesForThrottle.contains(event.frameTypeName)) 1
+    else event match {
+      case d: DataFrame if d.payload.isEmpty && isThrottledEmptyDataFrame(frameTypesForThrottle, d) => 1
+      case _                                                                                        => 0
+    }
+
+  private def isThrottledEmptyDataFrame(frameTypesForThrottle: Set[String], frame: DataFrame): Boolean =
+    frameTypesForThrottle.contains(EmptyDataFrameThrottleName) ||
+    (!frame.endStream && frameTypesForThrottle.contains(EmptyDataFrameNoEndStreamThrottleName))
+
   private[http2] def frameTypeAliasToFrameTypeName(frameType: String): Option[String] = {
     toRootLowerCase(frameType) match {
-      case "reset"         => Some("RstStreamFrame")
-      case "headers"       => Some("HeadersFrame")
-      case "continuation"  => Some("ContinuationFrame")
-      case "go-away"       => Some("GoAwayFrame")
-      case "priority"      => Some("PriorityFrame")
-      case "ping"          => Some("PingFrame")
-      case "push-promise"  => Some("PushPromiseFrame")
-      case "window-update" => Some("WindowUpdateFrame")
-      case _               => None
+      case "empty-data"               => Some(EmptyDataFrameThrottleName)
+      case "empty-data-no-end-stream" => Some(EmptyDataFrameNoEndStreamThrottleName)
+      case "reset"                    => Some("RstStreamFrame")
+      case "headers"                  => Some("HeadersFrame")
+      case "continuation"             => Some("ContinuationFrame")
+      case "go-away"                  => Some("GoAwayFrame")
+      case "priority"                 => Some("PriorityFrame")
+      case "ping"                     => Some("PingFrame")
+      case "push-promise"             => Some("PushPromiseFrame")
+      case "window-update"            => Some("WindowUpdateFrame")
+      case _                          => None
     }
   }
 

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

@@ -26,6 +26,14 @@ 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 (empty-data-no-end-stream)" in {
+      Http2Blueprint.frameTypeAliasToFrameTypeName("empty-data-no-end-stream") shouldEqual
+      Some(Http2Blueprint.EmptyDataFrameNoEndStreamThrottleName)
+    }
     "match frame type alias (reset)" in {
       Http2Blueprint.frameTypeAliasToFrameTypeName("reset") shouldEqual
       Some(RstStreamFrame(0, ErrorCode.PROTOCOL_ERROR).frameTypeName)
@@ -67,5 +75,40 @@ class Http2BlueprintSpec extends AnyWordSpec with Matchers {
     "not match unknown frame type alias" in {
       Http2Blueprint.frameTypeAliasToFrameTypeName("unknown") shouldEqual None
     }
+
+    "charge nothing when no frame type is throttled" in {
+      Http2Blueprint.frameCost(Set.empty, emptyDataFrame(endStream = false)) shouldEqual 0
+      Http2Blueprint.frameCost(Set.empty, rstStreamFrame) shouldEqual 0
+    }
+    "charge a frame matched by its own frame type name" in {
+      val throttled = Set(rstStreamFrame.frameTypeName)
+      Http2Blueprint.frameCost(throttled, rstStreamFrame) shouldEqual 1
+      Http2Blueprint.frameCost(throttled, emptyDataFrame(endStream = false)) shouldEqual 0
+    }
+    "charge only the empty DATA frames that do not end the stream (empty-data-no-end-stream)" in {
+      val throttled = Set(Http2Blueprint.EmptyDataFrameNoEndStreamThrottleName)
+      Http2Blueprint.frameCost(throttled, emptyDataFrame(endStream = false)) shouldEqual 1
+      // an empty DATA frame with END_STREAM is how a client closes a request body, so it is left uncharged
+      Http2Blueprint.frameCost(throttled, emptyDataFrame(endStream = true)) shouldEqual 0
+      Http2Blueprint.frameCost(throttled, dataFrame(endStream = false)) shouldEqual 0
+      Http2Blueprint.frameCost(throttled, dataFrame(endStream = true)) shouldEqual 0
+    }
+    "charge every empty DATA frame (empty-data)" in {
+      val throttled = Set(Http2Blueprint.EmptyDataFrameThrottleName)
+      Http2Blueprint.frameCost(throttled, emptyDataFrame(endStream = false)) shouldEqual 1
+      Http2Blueprint.frameCost(throttled, emptyDataFrame(endStream = true)) shouldEqual 1
+      Http2Blueprint.frameCost(throttled, dataFrame(endStream = false)) shouldEqual 0
+      Http2Blueprint.frameCost(throttled, dataFrame(endStream = true)) shouldEqual 0
+    }
+    "charge an empty DATA frame once when both empty DATA aliases are throttled" in {
+      val throttled =
+        Set(Http2Blueprint.EmptyDataFrameThrottleName, Http2Blueprint.EmptyDataFrameNoEndStreamThrottleName)
+      Http2Blueprint.frameCost(throttled, emptyDataFrame(endStream = false)) shouldEqual 1
+      Http2Blueprint.frameCost(throttled, emptyDataFrame(endStream = true)) shouldEqual 1
+    }
   }
+
+  private def emptyDataFrame(endStream: Boolean) = DataFrame(1, endStream, ByteString.empty)
+  private def dataFrame(endStream: Boolean) = DataFrame(1, endStream, ByteString("payload"))
+  private def rstStreamFrame = RstStreamFrame(1, ErrorCode.PROTOCOL_ERROR)
 }

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

@@ -0,0 +1,45 @@
+/*
+ * 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 which carry no payload and do not end their stream are throttled by default. They consume
+ * no flow-control window, so unlike data-carrying frames their number is not bounded by flow control, and unlike the
+ * empty DATA frame that carries END_STREAM they have no legitimate use.
+ *
+ * The config deliberately does not set `frame-type-throttle.frame-types`, so this covers the default.
+ */
+class Http2ServerEmptyDataNoEndStreamThrottleSpec extends Http2SpecWithMaterializer("""
+    pekko.http.server.http2.log-frames = on
+  """) {
+  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 that do not end the stream".inAssertAllStagesStopped(
+      new TestSetup with RequestResponseProbes {
+        val emptyDataFrame = FrameRenderer.renderFrame(FrameType.DATA, ByteFlag.Zero, 1, ByteString.empty)
+        network.sendBytes(Seq.fill(1000)(emptyDataFrame).reduce(_ ++ _))
+      })
+  }
+}

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

@@ -0,0 +1,44 @@
+/*
+ * 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 the opt-in "empty-data" throttle, which charges every DATA frame carrying no payload - including the one
+ * with END_STREAM that a client sends to close a request body. The default throttles only the empty DATA frames
+ * that do not end their stream, see [[Http2ServerEmptyDataNoEndStreamThrottleSpec]].
+ */
+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(_ ++ _))
+      })
+  }
+}