fix: do not send a response body to HTTP/2 HEAD requests

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

@@ -63,6 +63,19 @@ private[http2] case class Http2SubStream(
   def withCorrelationAttributes(newAttributes: Map[AttributeKey[?], ?]): Http2SubStream =
     copy(correlationAttributes = newAttributes)
 
+  /**
+   * Returns a copy that carries no data and ends the stream with its initial headers. The header pairs are kept as
+   * they are, so that a response keeps advertising the `content-length` the peer would have received otherwise.
+   *
+   * Used for responses to HEAD requests, which must not carry content (RFC 9110 section 9.3.2). The caller is
+   * responsible for cancelling `data` if it is a stream.
+   */
+  def withoutData: Http2SubStream =
+    copy(
+      initialHeaders = initialHeaders.copy(endStream = true),
+      trailingHeaders = OptionVal.None,
+      data = Left(ByteString.empty))
+
   /**
    * Create the request entity (when we're the server) or response entity (when we're the client) for this substream
    */

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

@@ -18,7 +18,7 @@ import pekko.annotation.InternalApi
 import pekko.http.impl.engine.http2.FrameEvent._
 import pekko.http.impl.engine.http2.Http2Protocol.ErrorCode
 import pekko.http.impl.engine.rendering.DateHeaderRendering
-import pekko.http.scaladsl.model.{ AttributeKey, HttpEntity }
+import pekko.http.scaladsl.model.{ AttributeKey, HttpEntity, HttpMethods }
 import pekko.http.scaladsl.model.http2.PeerClosedStreamException
 import pekko.http.scaladsl.settings.Http2CommonSettings
 import pekko.macros.LogHelper
@@ -54,6 +54,9 @@ private[http2] trait Http2StreamHandling extends GraphStageLogic with LogHelper
 
   def flowController: IncomingFlowController = IncomingFlowController.default(settings)
 
+  private def isHeadRequest(frame: ParsedHeadersFrame): Boolean =
+    frame.keyValuePairs.exists { case (name, value) => name == ":method" && (value eq HttpMethods.HEAD) }
+
   /**
    * Tries to generate demand of SubStreams on the inlet from the user handler. The
    * attemp to demand will succeed if the inlet is open and has no pending pull, and,
@@ -65,6 +68,10 @@ private[http2] trait Http2StreamHandling extends GraphStageLogic with LogHelper
   def tryPullSubStreams(): Unit
 
   private val streamStates = new mutable.LongMap[StreamState](settings.maxConcurrentStreams)
+  // Stream ids of incoming HEAD requests. The response to a HEAD request must not carry content
+  // (RFC 9110 section 9.3.2), but the response path only ever sees an HttpResponse and so cannot know the request
+  // method. Entries are removed when the response is created or when the stream is closed.
+  private val headRequestStreamIds = mutable.Set.empty[Int]
   private var largestIncomingStreamId = 0
   private var outstandingConnectionLevelWindow = Http2Protocol.InitialWindowSize
   private var totalBufferedData = 0
@@ -118,7 +125,15 @@ private[http2] trait Http2StreamHandling extends GraphStageLogic with LogHelper
     updateState(e.streamId, _.handle(e), "handleStreamEvent", e.frameTypeName)
 
   /** Called by Http2ServerDemux when a stream comes in from the user-handler */
-  def handleOutgoingCreated(stream: Http2SubStream): Unit = {
+  def handleOutgoingCreated(outgoing: Http2SubStream): Unit = {
+    // a response to a HEAD request must not carry content (RFC 9110 section 9.3.2); the headers are sent unchanged
+    // so that the peer still learns the `content-length` it would have received for a GET
+    val stream =
+      if (headRequestStreamIds.remove(outgoing.streamId) && outgoing.hasEntity) {
+        outgoing.data.foreach(_.runWith(Sink.cancelled)(subFusingMaterializer))
+        outgoing.withoutData
+      } else outgoing
+
     stream.initialHeaders.priorityInfo.foreach(multiplexer.updatePriority)
     if (streamFor(stream.streamId) != Closed) {
       multiplexer.pushControlFrame(stream.initialHeaders)
@@ -188,6 +203,7 @@ private[http2] trait Http2StreamHandling extends GraphStageLogic with LogHelper
     newState match {
       case Closed =>
         streamStates.remove(streamId)
+        headRequestStreamIds -= streamId
         if (streamStates.isEmpty) onAllStreamsClosed()
         tryPullSubStreams()
       case newState => streamStates.put(streamId, newState)
@@ -300,6 +316,7 @@ private[http2] trait Http2StreamHandling extends GraphStageLogic with LogHelper
         correlationAttributes: Map[AttributeKey[?], ?] = Map.empty): StreamState =
       event match {
         case frame @ ParsedHeadersFrame(streamId, endStream, _, _, _) =>
+          if (isServer && isHeadRequest(frame)) headRequestStreamIds += streamId
           if (endStream) {
             dispatchSubstream(frame, Left(ByteString.empty), correlationAttributes)
             nextStateEmpty
@@ -621,6 +638,7 @@ private[http2] trait Http2StreamHandling extends GraphStageLogic with LogHelper
       multiplexer.pushControlFrame(RstStreamFrame(streamId, ErrorCode.CANCEL))
       // FIXME: go through state machine and don't manipulate vars directly here
       streamStates.remove(streamId)
+      headRequestStreamIds -= streamId
       wasClosed = true
       buffer = ByteString.empty
       trailingHeaders = None

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

@@ -51,6 +51,13 @@ private[http2] class ResponseRendering(settings: ServerSettings, val log: Loggin
 
   override lazy val peerIdHeader: Option[(String, String)] = settings.serverHeader.map(h => h.lowercaseName -> h.value)
 
+  // Mirrors the status based part of the rules the HTTP/1.1 renderer applies through
+  // HttpMethod.contentLengthAllowed: no content-length for 1xx, 204 or 304 (RFC 9110 sections 8.6 and 15.4.5, where a
+  // 304 is supposed to carry the content-length a 200 would have had rather than a made up zero). GET is used here
+  // only because it carries the rules that are common to all methods; the request method is not known on this path.
+  protected override def contentLengthAllowed(response: HttpResponse): Boolean =
+    HttpMethods.GET.contentLengthAllowed(response.status)
+
 }
 
 /** INTERNAL API */
@@ -86,10 +93,16 @@ private[http2] sealed abstract class MessageRendering[R <: HttpMessage] extends
   protected def peerIdHeader: Option[(String, String)]
   protected def dateHeaderRendering: DateHeaderRendering
 
+  /**
+   * Whether a `content-length` may be rendered for this message. Always true for requests, where the length simply
+   * describes the request body.
+   */
+  protected def contentLengthAllowed(r: R): Boolean = true
+
   def apply(r: R): Http2SubStream = {
     val headerPairs = initialHeaderPairs(r)
 
-    HttpMessageRendering.addContentHeaders(headerPairs, r.entity)
+    HttpMessageRendering.addContentHeaders(headerPairs, r.entity, contentLengthAllowed(r))
     HttpMessageRendering.renderHeaders(r.headers, headerPairs, peerIdHeader, log, isServer = r.isResponse,
       shouldRenderAutoHeaders = true, dateHeaderRendering)
 
@@ -114,10 +127,12 @@ private[http2] object HttpMessageRendering {
   /**
    * Mutates `headerPairs` adding headers related to content (type and length).
    */
-  def addContentHeaders(headerPairs: VectorBuilder[(String, String)], entity: HttpEntity): Unit = {
+  def addContentHeaders(headerPairs: VectorBuilder[(String, String)], entity: HttpEntity,
+      contentLengthAllowed: Boolean): Unit = {
     if (entity.contentType ne ContentTypes.NoContentType)
       headerPairs += "content-type" -> entity.contentType.toString
-    entity.contentLengthOption.foreach(headerPairs += "content-length" -> _.toString)
+    if (contentLengthAllowed)
+      entity.contentLengthOption.foreach(headerPairs += "content-length" -> _.toString)
   }
 
   def renderHeaders(

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

@@ -97,12 +97,17 @@ private[http2] object RequestParsing {
 
         val entity = subStream.createEntity(contentLength, contentType)
 
+        // mirrors HttpServerBluePrint's ControllerStage for HTTP/1.1: the handler sees a GET, while the substream
+        // handling in Http2StreamHandling still knows the request was a HEAD and strips the response body
+        val effectiveMethod =
+          if (method == HttpMethods.HEAD && serverSettings.transparentHeadRequests) HttpMethods.GET else method
+
         val (path, rawQueryString) = pathAndRawQuery
         val authorityOrDefault: Uri.Authority = if (authority == null) Uri.Authority.Empty else authority
         val uri = Uri(scheme, authorityOrDefault, path, rawQueryString)
         val attributes = baseAttributes.updated(Http2.streamId, subStream.streamId)
 
-        new HttpRequest(method, uri, headers.result(), attributes, entity, HttpProtocols.`HTTP/2.0`)
+        new HttpRequest(effectiveMethod, uri, headers.result(), attributes, entity, HttpProtocols.`HTTP/2.0`)
       }
 
       @tailrec

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

@@ -411,16 +411,32 @@ class Http2ServerSpec extends Http2SpecWithMaterializer("""
         network.expectDecodedResponseHEADERSPairs(streamId = TheStreamId).toMap should contain(":status" -> "200")
       })
 
-      "render content-length for a strict response entity".inAssertAllStagesStopped(new HeadRequestSetup {
+      // RFC 9110 section 9.3.2: the server MUST NOT send content in a response to a HEAD request. The headers are
+      // still rendered as they would be for a GET, so that the peer learns the size of the resource.
+      "not send the response entity as DATA frames".inAssertAllStagesStopped(new HeadRequestSetup {
         sendHeadRequest()
         user.emitResponse(TheStreamId,
           HttpResponse(entity = HttpEntity(ContentTypes.`application/octet-stream`, ByteString("abcde"))))
 
-        val pairs = network.expectDecodedResponseHEADERSPairs(streamId = TheStreamId, endStream = false).toMap
+        val pairs = network.expectDecodedResponseHEADERSPairs(streamId = TheStreamId, endStream = true).toMap
         pairs should contain("content-length" -> "5")
         pairs should contain("content-type" -> "application/octet-stream")
+        network.expectNoBytes(100.millis)
       })
 
+      "not send DATA frames for a streamed response entity and cancel it".inAssertAllStagesStopped(
+        new HeadRequestSetup {
+          sendHeadRequest()
+          val entityDataOut = TestPublisher.probe[ByteString]()
+          user.emitResponse(TheStreamId,
+            HttpResponse(entity = HttpEntity(ContentTypes.`application/octet-stream`,
+              Source.fromPublisher(entityDataOut))))
+
+          network.expectDecodedResponseHEADERSPairs(streamId = TheStreamId, endStream = true)
+          entityDataOut.expectCancellation()
+          network.expectNoBytes(100.millis)
+        })
+
       // the HTTP/1.1 equivalent is "to a HEAD request setting a custom Content-Type and Content-Length
       // (default response entity)" in ResponseRendererSpec: it lets a handler answer a HEAD request with the
       // size of the hypothetical GET response without producing the bytes
@@ -430,49 +446,71 @@ class Http2ServerSpec extends Http2SpecWithMaterializer("""
           user.emitResponse(TheStreamId,
             HttpResponse(entity = HttpEntity.Default(ContentTypes.`application/octet-stream`, 100, Source.empty)))
 
-          network.expectDecodedResponseHEADERSPairs(streamId = TheStreamId, endStream = false).toMap should contain(
+          network.expectDecodedResponseHEADERSPairs(streamId = TheStreamId, endStream = true).toMap should contain(
             "content-length" -> "100")
+          network.expectNoBytes(100.millis)
         })
 
-      // FIXME: RFC 9110 section 9.3.2 says the server MUST NOT send content in a response to a HEAD request, but
-      // the HTTP/2 engine has no notion of the request method on the response path (ResponseRendering only ever
-      // sees the HttpResponse plus its stream id), so the entity is emitted as DATA frames. HTTP/1.1 strips it in
-      // HttpResponseRendererFactory. This test pins the current wire behaviour so that a fix has to flip it
-      // deliberately rather than silently.
-      "send the response entity as DATA frames (should not, see RFC 9110 section 9.3.2)".inAssertAllStagesStopped(
+      "translate HEAD to GET when transparent-head-requests is enabled and still strip the body".inAssertAllStagesStopped(
         new HeadRequestSetup {
-          sendHeadRequest()
+          override def settings: ServerSettings = super.settings.withTransparentHeadRequests(true)
+
+          sendHeadRequest().method shouldBe HttpMethods.GET
+
           user.emitResponse(TheStreamId,
             HttpResponse(entity = HttpEntity(ContentTypes.`application/octet-stream`, ByteString("abcde"))))
-
-          network.expectDecodedResponseHEADERSPairs(streamId = TheStreamId, endStream = false)
-          network.expectDATA(TheStreamId, endStream = true, ByteString("abcde"))
+          network.expectDecodedResponseHEADERSPairs(streamId = TheStreamId, endStream = true).toMap should contain(
+            "content-length" -> "5")
+          network.expectNoBytes(100.millis)
         })
 
-      // FIXME: `transparent-head-requests` is only applied by HttpServerBluePrint, so it has no effect over
-      // HTTP/2 and the handler always sees a HEAD request. Pinned here so the divergence from HTTP/1.1 is visible.
-      "ignore transparent-head-requests and pass HEAD through to the handler".inAssertAllStagesStopped(
-        new HeadRequestSetup {
-          override def settings: ServerSettings = super.settings.withTransparentHeadRequests(true)
+      "keep HEAD when transparent-head-requests is disabled".inAssertAllStagesStopped(new HeadRequestSetup {
+        sendHeadRequest().method shouldBe HttpMethods.HEAD
 
-          sendHeadRequest().method shouldBe HttpMethods.HEAD
+        user.emitResponse(TheStreamId, HttpResponse())
+        network.expectDecodedResponseHEADERSPairs(streamId = TheStreamId).toMap should contain(":status" -> "200")
+      })
+    }
 
-          user.emitResponse(TheStreamId, HttpResponse())
-          network.expectDecodedResponseHEADERSPairs(streamId = TheStreamId).toMap should contain(":status" -> "200")
-        })
+    // The status based rules are the ones PR #962 introduced for HTTP/1.1 in HttpMethod.contentLengthAllowed. They
+    // do not depend on the request method, so they are exercised here with a plain GET.
+    "render content-length according to the response status" should {
+      abstract class GetRequestSetup extends TestSetup with RequestResponseProbes {
+        val TheStreamId = 1
+        def responsePairs(response: HttpResponse): Map[String, String] = {
+          network.sendRequest(TheStreamId,
+            HttpRequest(HttpMethods.GET, "https://www.example.com/", protocol = HttpProtocols.`HTTP/2.0`))
+          user.expectRequest()
+          user.emitResponse(TheStreamId, response)
+          network.expectDecodedResponseHEADERSPairs(streamId = TheStreamId).toMap
+        }
+      }
 
-      // FIXME: HttpMessageRendering.addContentHeaders renders content-length straight from the entity and never
-      // consults HttpMethod.contentLengthAllowed, so a 304 gets `content-length: 0` where HTTP/1.1 omits the
-      // header entirely (RFC 9110 section 15.4.5: a 304 should carry the Content-Length a 200 would have had).
-      "render content-length 0 for a 304 response (HTTP/1.1 omits it)".inAssertAllStagesStopped(
-        new HeadRequestSetup {
-          sendHeadRequest()
-          user.emitResponse(TheStreamId, HttpResponse(StatusCodes.NotModified))
+      "render it for a 200 with an empty entity".inAssertAllStagesStopped(new GetRequestSetup {
+        responsePairs(HttpResponse()) should contain("content-length" -> "0")
+      })
 
-          val pairs = network.expectDecodedResponseHEADERSPairs(streamId = TheStreamId).toMap
-          pairs should contain(":status" -> "304")
-          pairs should contain("content-length" -> "0")
-        })
+      "not render it for a 204".inAssertAllStagesStopped(new GetRequestSetup {
+        val pairs = responsePairs(HttpResponse(StatusCodes.NoContent))
+        pairs should contain(":status" -> "204")
+        pairs.keySet should not contain "content-length"
+      })
+
+      // 205 is deliberately not exempt: RFC 9112 section 6.3 only lets 1xx, 204 and 304 be self delimiting, so a 205
+      // has to be framed. This is the case PR #962 fixed for HTTP/1.1 and the one http4s/http4s#7919 reports.
+      "render it for a 205".inAssertAllStagesStopped(new GetRequestSetup {
+        val pairs = responsePairs(HttpResponse(StatusCodes.ResetContent))
+        pairs should contain(":status" -> "205")
+        pairs should contain("content-length" -> "0")
+      })
+
+      // RFC 9110 section 15.4.5: a 304 is supposed to carry the content-length a 200 would have had, so rendering a
+      // zero here would be actively misleading.
+      "not render it for a 304".inAssertAllStagesStopped(new GetRequestSetup {
+        val pairs = responsePairs(HttpResponse(StatusCodes.NotModified))
+        pairs should contain(":status" -> "304")
+        pairs.keySet should not contain "content-length"
+      })
     }
 
     def requestTests(minCollectStrictEntityBytes: Int) = {

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

@@ -132,6 +132,17 @@ class RequestParsingSpec extends PekkoSpecWithMaterializer with Inside with Insp
       // pseudo-header field that appears in a header block after a regular
       // header field MUST be treated as malformed...
 
+      // Neither the CONNECT method of RFC 9113 section 8.5 nor the extended CONNECT of RFC 8441 is supported: a
+      // CONNECT request omits ":scheme" and ":path", which the parser rejects as mandatory. Pinned here so that the
+      // CONNECT specific rule in HttpMethods.contentLengthAllowed is understood to be unreachable over HTTP/2.
+      "not accept a CONNECT request" in {
+        val ex = parseExpectProtocolError(
+          keyValuePairs = Vector(
+            ":method" -> "CONNECT",
+            ":authority" -> "www.example.com:443"))
+        ex.getMessage should ===("Malformed request: Mandatory pseudo-header ':scheme' missing")
+      }
+
       "not accept pseudo-header fields after regular headers" in {
         val pseudoHeaders = Vector(
           ":method" -> "GET",