http/2: bound incoming header blocks with max-header-list-size

http-core/src/main/mima-filters/2.0.x.backwards.excludes/http2-max-header-list-size.excludes

@@ -0,0 +1,24 @@
+# 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 max-header-list-size setting for HTTP/2
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.javadsl.settings.Http2ClientSettings.maxHeaderListSize")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.javadsl.settings.Http2ClientSettings.withMaxHeaderListSize")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.javadsl.settings.Http2ServerSettings.getMaxHeaderListSize")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.javadsl.settings.Http2ServerSettings.withMaxHeaderListSize")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.scaladsl.settings.Http2ClientSettings.maxHeaderListSize")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.scaladsl.settings.Http2ServerSettings.maxHeaderListSize")

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

@@ -239,6 +239,21 @@ pekko.http {
       # the connection was established but before it received our SETTINGS.
       max-concurrent-streams = 256
 
+      # The maximum size of the header list (the sum of the sizes of the decompressed header names and values) that
+      # this endpoint is prepared to accept, in bytes. The value is advertised to the peer using the
+      # SETTINGS_MAX_HEADER_LIST_SIZE setting. A header block that decompresses to more than this amount is rejected
+      # with a GOAWAY(ENHANCE_YOUR_CALM) frame instead of being buffered.
+      #
+      # The same limit is applied to the accumulated header block fragments carried by a HEADERS frame and its
+      # subsequent CONTINUATION frames, so that the memory used for a header block that the peer never completes
+      # (END_HEADERS is never set) stays bounded. Each fragment is accounted with its frame header size on top of
+      # its payload size, which also bounds the number of empty fragments that are accepted for one header block.
+      #
+      # Note that peers calculate the header list size with an extra overhead of 32 octets per header field (see
+      # RFC 9113, section 6.5.2) while this implementation only counts the actual name and value bytes, so the
+      # effective limit for a well-behaved peer is somewhat stricter than the configured value.
+      max-header-list-size = 64 KiB
+
       # The maximum number of bytes to receive from a request entity in a single chunk.
       #
       # The reasoning to limit that amount (instead of delivering all buffered data for a stream) is that
@@ -473,6 +488,21 @@ pekko.http {
       # the connection was established but before it received our SETTINGS.
       max-concurrent-streams = 256
 
+      # The maximum size of the header list (the sum of the sizes of the decompressed header names and values) that
+      # this endpoint is prepared to accept, in bytes. The value is advertised to the peer using the
+      # SETTINGS_MAX_HEADER_LIST_SIZE setting. A header block that decompresses to more than this amount is rejected
+      # with a GOAWAY(ENHANCE_YOUR_CALM) frame instead of being buffered.
+      #
+      # The same limit is applied to the accumulated header block fragments carried by a HEADERS frame and its
+      # subsequent CONTINUATION frames, so that the memory used for a header block that the peer never completes
+      # (END_HEADERS is never set) stays bounded. Each fragment is accounted with its frame header size on top of
+      # its payload size, which also bounds the number of empty fragments that are accepted for one header block.
+      #
+      # Note that peers calculate the header list size with an extra overhead of 32 octets per header field (see
+      # RFC 9113, section 6.5.2) while this implementation only counts the actual name and value bytes, so the
+      # effective limit for a well-behaved peer is somewhat stricter than the configured value.
+      max-header-list-size = 64 KiB
+
       # The maximum number of bytes to receive from a request entity in a single chunk.
       #
       # The reasoning to limit that amount (instead of delivering all buffered data for a stream) is that

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

@@ -127,7 +127,7 @@ private[http] object Http2Blueprint {
       httpLayer(settings, log, dateHeaderRendering) atopKeepRight
       serverDemux(settings.http2Settings, initialDemuxerSettings, upgraded) atop
       FrameLogger.logFramesIfEnabled(settings.http2Settings.logFrames) atop // enable for debugging
-      hpackCoding(masterHttpHeaderParser, settings.parserSettings)
+      hpackCoding(masterHttpHeaderParser, settings.parserSettings, settings.http2Settings.maxHeaderListSize)
 
     val frameTypesForThrottle = getFrameTypesForThrottle(settings.http2Settings)
 
@@ -153,7 +153,7 @@ private[http] object Http2Blueprint {
       httpLayerClient(masterHttpHeaderParser, settings, log)).atop(
       clientDemux(settings.http2Settings, masterHttpHeaderParser)).atop(
       FrameLogger.logFramesIfEnabled(settings.http2Settings.logFrames)).atop( // enable for debugging
-      hpackCoding(masterHttpHeaderParser, settings.parserSettings)).atop(
+      hpackCoding(masterHttpHeaderParser, settings.parserSettings, settings.http2Settings.maxHeaderListSize)).atop(
       framingClient(log)).atop(
       errorHandling(log)).atop(
       idleTimeoutIfConfigured(settings.idleTimeout))
@@ -247,11 +247,11 @@ private[http] object Http2Blueprint {
    * TODO: introduce another FrameEvent type that exclude HeadersFrame and ContinuationFrame from
    * reaching the higher-level.
    */
-  def hpackCoding(masterHttpHeaderParser: HttpHeaderParser, parserSettings: ParserSettings)
+  def hpackCoding(masterHttpHeaderParser: HttpHeaderParser, parserSettings: ParserSettings, maxHeaderListSize: Int)
       : BidiFlow[FrameEvent, FrameEvent, FrameEvent, FrameEvent, NotUsed] =
     BidiFlow.fromFlows(
       Flow[FrameEvent].via(HeaderCompression),
-      Flow[FrameEvent].via(new HeaderDecompression(masterHttpHeaderParser, parserSettings)))
+      Flow[FrameEvent].via(new HeaderDecompression(masterHttpHeaderParser, parserSettings, maxHeaderListSize)))
 
   /**
    * Creates substreams for every stream and manages stream state machines

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

@@ -295,7 +295,8 @@ private[http2] abstract class Http2Demux(http2Settings: Http2CommonSettings,
       // enforced immediately even before the acknowledgement is received.
       // Reminder: the receiver of a SETTINGS frame must process them in the order they are received.
       val initialLocalSettings: immutable.Seq[Setting] = immutable.Seq(
-        Setting(SettingIdentifier.SETTINGS_MAX_CONCURRENT_STREAMS, http2Settings.maxConcurrentStreams)) ++
+        Setting(SettingIdentifier.SETTINGS_MAX_CONCURRENT_STREAMS, http2Settings.maxConcurrentStreams),
+        Setting(SettingIdentifier.SETTINGS_MAX_HEADER_LIST_SIZE, http2Settings.maxHeaderListSize)) ++
         immutable.Seq(Setting(SettingIdentifier.SETTINGS_ENABLE_PUSH, 0)).filter(_ => !isServer) // only on client
 
       override def preStart(): Unit = {

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

@@ -36,13 +36,23 @@ import scala.collection.immutable.VectorBuilder
  * INTERNAL API
  *
  * Can be used on server and client side.
+ *
+ * @param maxHeaderListSize the maximum size of a decoded header list. The same limit is applied to the accumulated
+ *                          header block fragments of a HEADERS frame and its CONTINUATION frames, so that the memory
+ *                          used for a header block that the peer never completes stays bounded
+ *                          (see RFC 9113, section 10.5).
  */
 @InternalApi
-private[http2] final class HeaderDecompression(masterHeaderParser: HttpHeaderParser, parserSettings: ParserSettings)
+private[http2] final class HeaderDecompression(masterHeaderParser: HttpHeaderParser, parserSettings: ParserSettings,
+    maxHeaderListSize: Int)
     extends GraphStage[FlowShape[FrameEvent, FrameEvent]] {
   val UTF8 = StandardCharsets.UTF_8
   val US_ASCII = StandardCharsets.US_ASCII
 
+  // Each fragment is accounted with the size of its frame header on top of its payload size. Without that, empty
+  // CONTINUATION frames would never add to the accumulated header block and their number would be unbounded.
+  private val FrameHeaderSize = 9
+
   val eventsIn = Inlet[FrameEvent]("HeaderDecompression.eventsIn")
   val eventsOut = Outlet[FrameEvent]("HeaderDecompression.eventsOut")
 
@@ -51,8 +61,8 @@ private[http2] final class HeaderDecompression(masterHeaderParser: HttpHeaderPar
   def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
     new HandleOrPassOnStage[FrameEvent, FrameEvent](shape) {
       val httpHeaderParser = masterHeaderParser.createShallowCopy()
-      val decoder = new pekko.http.shaded.com.twitter.hpack.Decoder(Http2Protocol.InitialMaxHeaderListSize,
-        Http2Protocol.InitialMaxHeaderTableSize)
+      val decoder =
+        new pekko.http.shaded.com.twitter.hpack.Decoder(maxHeaderListSize, Http2Protocol.InitialMaxHeaderTableSize)
 
       become(Idle)
 
@@ -99,9 +109,11 @@ private[http2] final class HeaderDecompression(masterHeaderParser: HttpHeaderPar
         val stream = payload.compact.asInputStream
         try {
           decoder.decode(stream, Receiver) // only compact ByteString supports InputStream with mark/reset
-          decoder.endHeaderBlock() // TODO: do we have to check the result here?
+          // the decoder stops emitting headers as soon as the limit is exceeded and reports that here
+          val truncated = decoder.endHeaderBlock()
 
-          push(eventsOut, ParsedHeadersFrame(streamId, endStream, headers.result(), prioInfo, None))
+          if (truncated) headerListSizeExceeded(streamId)
+          else push(eventsOut, ParsedHeadersFrame(streamId, endStream, headers.result(), prioInfo, None))
         } catch {
           case ex: ParsingException =>
             // push details further and let RequestErrorFlow handle responding with bad request
@@ -119,6 +131,7 @@ private[http2] final class HeaderDecompression(masterHeaderParser: HttpHeaderPar
         val handleEvent: PartialFunction[FrameEvent, Unit] = {
           case HeadersFrame(streamId, endStream, endHeaders, fragment, prioInfo) =>
             if (endHeaders) parseAndEmit(streamId, endStream, fragment, prioInfo)
+            else if (exceedsMaxHeaderListSize(0, fragment)) headerListSizeExceeded(streamId)
             else {
               become(new ReceivingHeaders(streamId, endStream, fragment, prioInfo))
               pull(eventsIn)
@@ -132,21 +145,37 @@ private[http2] final class HeaderDecompression(masterHeaderParser: HttpHeaderPar
       class ReceivingHeaders(streamId: Int, endStream: Boolean, initiallyReceivedData: ByteString,
           priorityInfo: Option[PriorityFrame]) extends State {
         var receivedData = initiallyReceivedData
+        // includes the frame headers of the fragments received so far, see `FrameHeaderSize`
+        var accountedSize: Long = FrameHeaderSize + initiallyReceivedData.size
 
         val handleEvent: PartialFunction[FrameEvent, Unit] = {
           case ContinuationFrame(`streamId`, endHeaders, payload) =>
-            if (endHeaders) {
+            if (exceedsMaxHeaderListSize(accountedSize, payload))
+              // Neither the HPACK decoder nor any of the checks further down the line run before the header block
+              // is complete, so this is the only place where the size of an unfinished header block is bounded.
+              headerListSizeExceeded(streamId)
+            else if (endHeaders) {
               parseAndEmit(streamId, endStream, receivedData ++ payload, priorityInfo)
               become(Idle)
             } else {
               receivedData ++= payload
+              accountedSize += FrameHeaderSize + payload.size
               pull(eventsIn)
             }
           case x =>
             protocolError(s"While waiting for CONTINUATION frame on stream $streamId received unexpected frame $x")
         }
       }
 
+      def exceedsMaxHeaderListSize(accountedSize: Long, payload: ByteString): Boolean =
+        accountedSize + FrameHeaderSize + payload.size > maxHeaderListSize
+
+      def headerListSizeExceeded(streamId: Int): Unit =
+        fail(eventsOut,
+          new Http2ProtocolException(
+            ErrorCode.ENHANCE_YOUR_CALM,
+            s"Header block of stream $streamId exceeded the configured max-header-list-size of $maxHeaderListSize bytes"))
+
       def protocolError(msg: String): Unit = failStage(new Http2ProtocolException(msg))
     }
 }

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

@@ -34,6 +34,20 @@ trait Http2ClientSettings { self: scaladsl.settings.Http2ClientSettings.Http2Cli
   def maxConcurrentStreams: Int
   def withMaxConcurrentStreams(newValue: Int): Http2ClientSettings = copy(maxConcurrentStreams = newValue)
 
+  /**
+   * The maximum size of a decoded header list that this endpoint is prepared to accept, in bytes. The value is
+   * advertised to the peer via SETTINGS_MAX_HEADER_LIST_SIZE and the same limit is applied to the accumulated
+   * header block fragments of a HEADERS frame and its CONTINUATION frames.
+   *
+   * @since 2.0.0
+   */
+  def maxHeaderListSize: Int
+
+  /**
+   * @since 2.0.0
+   */
+  def withMaxHeaderListSize(newValue: Int): Http2ClientSettings = copy(maxHeaderListSize = newValue)
+
   def outgoingControlFrameBufferSize: Int
   def withOutgoingControlFrameBufferSize(newValue: Int): Http2ClientSettings =
     copy(outgoingControlFrameBufferSize = newValue)

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

@@ -42,6 +42,20 @@ trait Http2ServerSettings {
   def getMaxConcurrentStreams: Int = maxConcurrentStreams
   def withMaxConcurrentStreams(newValue: Int): Http2ServerSettings
 
+  /**
+   * The maximum size of a decoded header list that this endpoint is prepared to accept, in bytes. The value is
+   * advertised to the peer via SETTINGS_MAX_HEADER_LIST_SIZE and the same limit is applied to the accumulated
+   * header block fragments of a HEADERS frame and its CONTINUATION frames.
+   *
+   * @since 2.0.0
+   */
+  def getMaxHeaderListSize: Int = maxHeaderListSize
+
+  /**
+   * @since 2.0.0
+   */
+  def withMaxHeaderListSize(newValue: Int): Http2ServerSettings
+
   def getOutgoingControlFrameBufferSize: Int = outgoingControlFrameBufferSize
   def withOutgoingControlFrameBufferSize(newValue: Int): Http2ServerSettings