HTTP/2 rapid reset mitigation

http-bench-jmh/src/main/scala/org/apache/pekko/http/impl/engine/http2/H2ClientServerBenchmark.scala

@@ -24,6 +24,7 @@ import pekko.http.scaladsl.settings.{ ClientConnectionSettings, ServerSettings }
 import pekko.stream.TLSProtocol.{ SslTlsInbound, SslTlsOutbound }
 import pekko.stream.scaladsl.{ BidiFlow, Flow, Keep, Sink, Source }
 import pekko.util.ByteString
+import com.typesafe.config.ConfigFactory
 import org.openjdk.jmh.annotations._
 
 import java.util.concurrent.{ CountDownLatch, TimeUnit }
@@ -40,6 +41,9 @@ class H2ClientServerBenchmark extends CommonBenchmark with H2RequestResponseBenc
 
   val numRequests = 1000
 
+  @Param(Array("1s", "0s"))
+  var resetFrameThrottleInterval: String = _
+
   @Benchmark
   @OperationsPerInvocation(1000) // should be same as numRequest
   def benchRequestProcessing(): Unit = {
@@ -69,7 +73,9 @@ class H2ClientServerBenchmark extends CommonBenchmark with H2RequestResponseBenc
   def setup(): Unit = {
     initRequestResponse()
 
-    system = ActorSystem("PekkoHttpBenchmarkSystem", config)
+    val resetFrameConfig = ConfigFactory.parseString(
+      s"pekko.http.server.http2.reset-frame.throttle-interval=$resetFrameThrottleInterval")
+    system = ActorSystem("PekkoHttpBenchmarkSystem", resetFrameConfig.withFallback(config))
     val settings = implicitly[ServerSettings]
     val log = system.log
     implicit val ec = system.dispatcher

http-core/src/main/mima-filters/1.0.x.backwards.excludes/http2-rapid-reset-configs.backwards.excludes

@@ -0,0 +1,21 @@
+# 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.
+
+# New configs added to support throttling HTTP/2 reset frames
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.scaladsl.settings.Http2ServerSettings.resetFrameThrottleCost")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.scaladsl.settings.Http2ServerSettings.resetFrameThrottleBurst")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.scaladsl.settings.Http2ServerSettings.resetFrameThrottleInterval")

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

@@ -307,6 +307,15 @@ pekko.http {
       # Fail the connection if a sent ping is not acknowledged within this timeout.
       # When zero the ping-interval is used, if set the value must be evenly divisible by less than or equal to the ping-interval.
       ping-timeout = 0s
+
+      reset-frame {
+        # Configure the throttle for Reset Frames (https://github.com/apache/incubator-pekko-http/issues/332)
+        throttle-cost = 100
+        throttle-burst = 100
+        # setting pekko.http.server.http2.reset-frame.throttle-interval to 0s will disable the throttle
+        # try changing it to 1s if you want to apply a meaningful limit (the cost and burst apply to the interval time)
+        throttle-interval = 0s
+      }
     }
 
     websocket {

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

@@ -35,7 +35,7 @@ import pekko.http.scaladsl.settings.{
   ParserSettings,
   ServerSettings
 }
-import pekko.stream.{ BidiShape, Graph, StreamTcpException }
+import pekko.stream.{ BidiShape, Graph, StreamTcpException, ThrottleMode }
 import pekko.stream.TLSProtocol._
 import pekko.stream.scaladsl.{ BidiFlow, Flow, Keep, Source }
 import pekko.util.ByteString
@@ -122,12 +122,17 @@ private[http] object Http2Blueprint {
       telemetry: TelemetrySpi,
     dateHeaderRendering: DateHeaderRendering): BidiFlow[HttpResponse, ByteString, ByteString, HttpRequest, ServerTerminator] = {
     val masterHttpHeaderParser = HttpHeaderParser(settings.parserSettings, log) // FIXME: reuse for framing
-    telemetry.serverConnection atop
+    val flow0 = telemetry.serverConnection atop
       httpLayer(settings, log, dateHeaderRendering) atopKeepRight
       serverDemux(settings.http2Settings, initialDemuxerSettings, upgraded) atop
       FrameLogger.logFramesIfEnabled(settings.http2Settings.logFrames) atop // enable for debugging
-      hpackCoding(masterHttpHeaderParser, settings.parserSettings) atop
-      framing(log) atop
+      hpackCoding(masterHttpHeaderParser, settings.parserSettings)
+
+    val flow1 = if (settings.http2Settings.resetFrameThrottleInterval.toMillis > 0) {
+      flow0 atop rapidResetMitigation(settings.http2Settings) atopKeepLeft framing(log)
+    } else flow0 atop framing(log)
+
+    flow1 atop
       errorHandling(log) atop
       idleTimeoutIfConfigured(settings.idleTimeout)
   }
@@ -198,6 +203,19 @@ private[http] object Http2Blueprint {
       Flow[FrameEvent].map(FrameRenderer.render).prepend(Source.single(Http2Protocol.ClientConnectionPreface)),
       Flow[ByteString].via(new Http2FrameParsing(shouldReadPreface = false, log)))
 
+  private def rapidResetMitigation(
+      settings: Http2ServerSettings): BidiFlow[FrameEvent, FrameEvent, FrameEvent, FrameEvent, NotUsed] = {
+    def frameCost(event: FrameEvent): Int = event match {
+      case _: FrameEvent.RstStreamFrame => 1
+      case _                            => 0
+    }
+
+    BidiFlow.fromFlows(
+      Flow[FrameEvent],
+      Flow[FrameEvent].throttle(settings.resetFrameThrottleCost, settings.resetFrameThrottleInterval,
+        settings.resetFrameThrottleBurst, frameCost, ThrottleMode.Enforcing))
+  }
+
   /**
    * Runs hpack encoding and decoding. Incoming frames that are processed are HEADERS and CONTINUATION.
    * Outgoing frame is ParsedHeadersFrame.
@@ -290,5 +308,9 @@ private[http] object Http2Blueprint {
     def atopKeepRight[OO1, II2, Mat2](
         other: Graph[BidiShape[O1, OO1, II2, I2], Mat2]): BidiFlow[I1, OO1, II2, O2, Mat2] =
       bidi.atopMat(other)(Keep.right)
+
+    def atopKeepLeft[OO1, II2, Mat2](
+        other: Graph[BidiShape[O1, OO1, II2, I2], Mat2]): BidiFlow[I1, OO1, II2, O2, Mat] =
+      bidi.atopMat(other)(Keep.left)
   }
 }

http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/Http2ServerSettings.scala

@@ -51,6 +51,14 @@ trait Http2ServerSettings {
 
   def getPingTimeout: Duration = Duration.ofMillis(pingTimeout.toMillis)
   def withPingTimeout(timeout: Duration): Http2ServerSettings = withPingTimeout(timeout.toMillis.millis)
+
+  def getResetThrottleCost(): Int = resetFrameThrottleCost
+  def getResetThrottleBurst(): Int = resetFrameThrottleBurst
+
+  def getResetThrottleInterval: Duration = Duration.ofMillis(resetFrameThrottleInterval.toMillis)
+
+  def withResetThrottleInterval(interval: Duration): Http2ServerSettings =
+    withResetThrottleInterval(interval.toMillis.millis)
 }
 object Http2ServerSettings extends SettingsCompanion[Http2ServerSettings] {
   def create(config: Config): Http2ServerSettings = scaladsl.settings.Http2ServerSettings(config)

http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/Http2ServerSettings.scala

@@ -102,6 +102,15 @@ trait Http2ServerSettings extends javadsl.settings.Http2ServerSettings with Http
   def pingTimeout: FiniteDuration
   def withPingTimeout(timeout: FiniteDuration): Http2ServerSettings = copy(pingTimeout = timeout)
 
+  def resetFrameThrottleCost: Int
+  def withResetThrottleCost(cost: Int) = copy(resetFrameThrottleCost = cost)
+
+  def resetFrameThrottleBurst: Int
+  def withResetThrottleBurst(burst: Int) = copy(resetFrameThrottleBurst = burst)
+
+  def resetFrameThrottleInterval: FiniteDuration
+  def withResetThrottleInterval(interval: FiniteDuration) = copy(resetFrameThrottleInterval = interval)
+
   @InternalApi
   private[http] def internalSettings: Option[Http2InternalServerSettings]
   @InternalApi
@@ -124,6 +133,9 @@ object Http2ServerSettings extends SettingsCompanion[Http2ServerSettings] {
       logFrames: Boolean,
       pingInterval: FiniteDuration,
       pingTimeout: FiniteDuration,
+      resetFrameThrottleCost: Int,
+      resetFrameThrottleBurst: Int,
+      resetFrameThrottleInterval: FiniteDuration,
       internalSettings: Option[Http2InternalServerSettings])
       extends Http2ServerSettings {
     require(maxConcurrentStreams >= 0, "max-concurrent-streams must be >= 0")
@@ -151,6 +163,9 @@ object Http2ServerSettings extends SettingsCompanion[Http2ServerSettings] {
       logFrames = c.getBoolean("log-frames"),
       pingInterval = c.getFiniteDuration("ping-interval"),
       pingTimeout = c.getFiniteDuration("ping-timeout"),
+      resetFrameThrottleCost = c.getInt("reset-frame.throttle-cost"),
+      resetFrameThrottleBurst = c.getInt("reset-frame.throttle-burst"),
+      resetFrameThrottleInterval = c.getFiniteDuration("reset-frame.throttle-interval"),
       None // no possibility to configure internal settings with config
     )
   }

http-core/src/test/scala/org/apache/pekko/http/impl/util/WithLogCapturing.scala

@@ -34,6 +34,12 @@ trait WithLogCapturing extends SuiteMixin { this: TestSuite =>
    */
   protected def failOnSevereMessages: Boolean = false
 
+  /**
+   * We expect a severe message but the message should contain this text. If there are any other severe messages,
+   * the test will fail.
+   */
+  protected val expectSevereLogsOnlyToMatch: Option[String] = None
+
   /**
    * Can be overridden to adapt which events should be considered as severe if `failOnSevereMessages` is
    * enabled.
@@ -86,6 +92,19 @@ trait WithLogCapturing extends SuiteMixin { this: TestSuite =>
       Failed(new AssertionError(
         s"No severe log messages should be emitted during test run but got [${stats(
             Logging.WarningLevel)}] warnings and [${stats(Logging.ErrorLevel)}] errors (see marked lines above)"))
+    } else if (expectSevereLogsOnlyToMatch.nonEmpty) {
+      val severeEvents = events.filter(isSevere(_))
+      val matchingEvents = severeEvents.filter(_.message.toString.contains(expectSevereLogsOnlyToMatch.get))
+      if (severeEvents.isEmpty || matchingEvents != severeEvents) {
+        val stats = events.groupBy(_.level).mapValues(_.size).toMap.withDefaultValue(0)
+        flushLog()
+
+        Failed(new AssertionError(
+          s"Expected an error during test run but got unexpected results - got [${
+              stats(
+                Logging.WarningLevel)
+            }] warnings and [${stats(Logging.ErrorLevel)}] errors (see marked lines above)"))
+      } else res
     } else res
 
   }

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

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2018-2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+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.ByteStringBuilder
+
+import java.nio.ByteOrder
+
+/**
+ * This tests the http2 server throttle support for rapid resets is disabled by default.
+ */
+class Http2ServerDisableResetThrottleSpec extends Http2SpecWithMaterializer("""
+    pekko.http.server.remote-address-header = on
+    pekko.http.server.http2.log-frames = on
+  """) {
+  override def failOnSevereMessages: Boolean = true
+
+  "The Http/2 server implementation" should {
+    "not cancel connection during rapid reset attack (throttle disabled)".inAssertAllStagesStopped(
+      new TestSetup with RequestResponseProbes {
+        implicit val bigEndian: ByteOrder = ByteOrder.BIG_ENDIAN
+        val bb = new ByteStringBuilder
+        bb.putInt(0)
+        val rstFrame = FrameRenderer.renderFrame(FrameType.RST_STREAM, ByteFlag.Zero, 1, bb.result())
+        val longFrame = Seq.fill(1000)(rstFrame).reduce(_ ++ _)
+        network.sendBytes(longFrame)
+      })
+  }
+}

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

@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+/*
+ * Copyright (C) 2018-2022 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+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.ByteStringBuilder
+
+import java.nio.ByteOrder
+
+/**
+ * This tests the http2 server throttle support for rapid resets.
+ */
+class Http2ServerEnableResetThrottleSpec extends Http2SpecWithMaterializer("""
+    pekko.http.server.remote-address-header = on
+    pekko.http.server.http2.log-frames = on
+    pekko.http.server.http2.reset-frame.throttle-interval = 1s
+  """) {
+  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 during rapid reset attack".inAssertAllStagesStopped(new TestSetup with RequestResponseProbes {
+      implicit val bigEndian: ByteOrder = ByteOrder.BIG_ENDIAN
+      val bb = new ByteStringBuilder
+      bb.putInt(0)
+      val rstFrame = FrameRenderer.renderFrame(FrameType.RST_STREAM, ByteFlag.Zero, 1, bb.result())
+      val longFrame = Seq.fill(1000)(rstFrame).reduce(_ ++ _)
+      network.sendBytes(longFrame)
+    })
+  }
+}