give ParsingErrorHandler access to the request that was rejected

http-core/src/main/mima-filters/2.0.x.backwards.excludes/illegal-request-context.excludes

@@ -0,0 +1,25 @@
+# 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.
+
+# internal API: MessageStartError carries what is known about the request that was rejected
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.http.impl.engine.parsing.ParserOutput#MessageStartError.copy")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.http.impl.engine.parsing.ParserOutput#MessageStartError.this")
+ProblemFilters.exclude[MissingTypesProblem]("org.apache.pekko.http.impl.engine.parsing.ParserOutput$MessageStartError$")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.http.impl.engine.parsing.ParserOutput#MessageStartError.apply")
+ProblemFilters.exclude[IncompatibleSignatureProblem]("org.apache.pekko.http.impl.engine.parsing.ParserOutput#MessageStartError.unapply")
+
+# internal API: replaced by an instance level completion handling that can report the same context

http-core/src/main/scala/org/apache/pekko/http/ParsingErrorHandler.scala

@@ -13,19 +13,115 @@
 
 package org.apache.pekko.http
 
+import java.util.Optional
+
+import scala.jdk.OptionConverters._
+
 import org.apache.pekko
 import pekko.event.LoggingAdapter
 import pekko.http.javadsl.{ model => jm }
-import pekko.http.scaladsl.model.{ ErrorInfo, HttpResponse, StatusCode }
+import pekko.http.scaladsl.model.{ ErrorInfo, HttpMethod, HttpProtocol, HttpResponse, StatusCode }
 import pekko.http.scaladsl.settings.ServerSettings
 
+/**
+ * What is known about a request that failed to parse, at the point where parsing gave up.
+ *
+ * Every field is optional because a request can be rejected before that part of it has been read:
+ * a request with an unsupported method fails before the request target is seen, and one with an
+ * unparsable request target fails before the protocol is seen.
+ *
+ * Note that `rawRequestTarget` is unvalidated, attacker-controlled input, by definition malformed
+ * whenever the rejection was caused by the request target itself. Anything that logs or echoes it
+ * has to escape it.
+ *
+ * @since 2.0.0
+ */
+final class IllegalRequestContext private[http] (
+    val method: Option[HttpMethod],
+    val rawRequestTarget: Option[String],
+    val protocol: Option[HttpProtocol]) {
+
+  /**
+   * Java API
+   *
+   * @since 2.0.0
+   */
+  def getMethod: Optional[jm.HttpMethod] = method.map(m => m: jm.HttpMethod).toJava
+
+  /**
+   * Java API
+   *
+   * @since 2.0.0
+   */
+  def getRawRequestTarget: Optional[String] = rawRequestTarget.toJava
+
+  /**
+   * Java API
+   *
+   * @since 2.0.0
+   */
+  def getProtocol: Optional[jm.HttpProtocol] = protocol.map(p => p: jm.HttpProtocol).toJava
+
+  override def toString: String =
+    s"IllegalRequestContext(${method.map(_.value).getOrElse("-")}," +
+    s"${rawRequestTarget.getOrElse("-")},${protocol.map(_.value).getOrElse("-")})"
+}
+
+object IllegalRequestContext {
+
+  /**
+   * A context that knows nothing about the request, used when no information could be recovered.
+   *
+   * @since 2.0.0
+   */
+  val empty: IllegalRequestContext = new IllegalRequestContext(None, None, None)
+
+  private[http] def apply(
+      method: Option[HttpMethod],
+      rawRequestTarget: Option[String],
+      protocol: Option[HttpProtocol]): IllegalRequestContext =
+    if (method.isEmpty && rawRequestTarget.isEmpty && protocol.isEmpty) empty
+    else new IllegalRequestContext(method, rawRequestTarget, protocol)
+}
+
+/**
+ * Produces the response to a request that failed to parse. Selected by the
+ * `pekko.http.server.parsing-error-handler` setting.
+ *
+ * This is also the earliest public symbol that observes a rejected request, so it is what
+ * observability tooling attaches to: the OpenTelemetry Java agent instruments `handle` to emit a
+ * span for a request that never reaches the route handler
+ * (open-telemetry/opentelemetry-java-instrumentation#5139).
+ */
 abstract class ParsingErrorHandler {
   def handle(status: StatusCode, error: ErrorInfo, log: LoggingAdapter, settings: ServerSettings): jm.HttpResponse
+
+  /**
+   * Called by the server for a request that failed to parse, with what is known about that request.
+   *
+   * The default implementation ignores `context` and delegates to the four-argument `handle`, so
+   * existing handlers keep working unchanged; override this method instead to make use of the
+   * context. The parameter is still worth passing for a handler that does not read it, because the
+   * arguments of this method are visible to anything instrumenting it: the OpenTelemetry Java agent
+   * has to name the span it emits for a rejected request `HTTP` and report neither
+   * `http.request.method` nor `url.path`, since the four-argument signature describes only the
+   * failure and never the request that caused it. See
+   * [[https://github.com/apache/pekko-http/issues/1245]].
+   *
+   * Note that `DefaultParsingErrorHandler` deliberately keeps implementing the four-argument method
+   * rather than this one, so that advice matching that signature keeps firing.
+   *
+   * @since 2.0.0
+   */
+  def handle(status: StatusCode, error: ErrorInfo, log: LoggingAdapter, settings: ServerSettings,
+      context: IllegalRequestContext): jm.HttpResponse =
+    handle(status, error, log, settings)
 }
 
 object DefaultParsingErrorHandler extends ParsingErrorHandler {
   import pekko.http.impl.engine.parsing.logParsingError
 
+  // implements the four-argument method on purpose, see the scaladoc of the five-argument one
   override def handle(
       status: StatusCode, info: ErrorInfo, log: LoggingAdapter, settings: ServerSettings): HttpResponse = {
     logParsingError(

http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala

@@ -201,7 +201,7 @@ private[http] object OutgoingConnectionBlueprint {
             push(httpResponseOut, new HttpResponse(statusCode, headers, attributes, entity, protocol))
             completeOnMessageEnd = closeRequested
 
-          case MessageStartError(_, info) =>
+          case MessageStartError(_, info, _) =>
             throw IllegalResponseException(info)
 
           case other =>

http-core/src/main/scala/org/apache/pekko/http/impl/engine/parsing/HttpMessageParser.scala

@@ -22,6 +22,7 @@ import org.parboiled2.CharUtils
 
 import org.apache.pekko
 import pekko.annotation.InternalApi
+import pekko.http.IllegalRequestContext
 import pekko.http.impl.model.parser.CharacterClasses
 import pekko.http.impl.util.HttpConstants._
 import pekko.http.scaladsl.model.{ ParsingException => _, _ }
@@ -62,6 +63,25 @@ private[http] trait HttpMessageParser[Output >: MessageOutput <: ParserOutput] {
 
   /** invoked if the specified protocol is unknown */
   protected def onBadProtocol(input: ByteString): Nothing
+
+  /**
+   * What is known about the message that is currently being parsed, at the point where parsing failed.
+   * Only the request parser has anything to report here.
+   */
+  protected def illegalRequestContext: IllegalRequestContext = IllegalRequestContext.empty
+
+  /** The protocol of the message that is currently being parsed */
+  protected final def currentProtocol: HttpProtocol = protocol
+
+  /**
+   * Completion handling for a message start that was truncated by the connection closing, reporting
+   * what the parser had already read of that message.
+   */
+  protected final val completionIsMessageStartError: CompletionHandling =
+    () =>
+      Some(MessageStartError(StatusCodes.BadRequest, ErrorInfo("Illegal HTTP message start"),
+        illegalRequestContext))
+
   protected def parseMessage(input: ByteString, offset: Int): HttpMessageParser.StateResult
   protected def parseEntity(headers: List[HttpHeader], protocol: HttpProtocol, input: ByteString, bodyStart: Int,
       clh: Option[`Content-Length`], cth: Option[`Content-Type`], isChunked: Boolean,
@@ -122,7 +142,7 @@ private[http] trait HttpMessageParser[Output >: MessageOutput <: ParserOutput] {
   }
 
   protected final def startNewMessage(input: ByteString, offset: Int): StateResult = {
-    if (offset < input.length) setCompletionHandling(CompletionIsMessageStartError)
+    if (offset < input.length) setCompletionHandling(completionIsMessageStartError)
     try parseMessage(input, offset)
     catch { case NotEnoughDataException => continue(input, offset)(startNewMessage) }
   }
@@ -363,7 +383,7 @@ private[http] trait HttpMessageParser[Output >: MessageOutput <: ParserOutput] {
   protected final def failMessageStart(status: StatusCode, summary: String, detail: String = ""): StateResult =
     failMessageStart(status, ErrorInfo(summary, detail))
   protected final def failMessageStart(status: StatusCode, info: ErrorInfo): StateResult = {
-    emit(MessageStartError(status, info))
+    emit(MessageStartError(status, info, illegalRequestContext))
     setCompletionHandling(CompletionOk)
     terminate()
   }
@@ -433,8 +453,6 @@ private[http] object HttpMessageParser {
 
   type CompletionHandling = () => Option[ErrorOutput]
   val CompletionOk: CompletionHandling = () => None
-  val CompletionIsMessageStartError: CompletionHandling =
-    () => Some(ParserOutput.MessageStartError(StatusCodes.BadRequest, ErrorInfo("Illegal HTTP message start")))
   val CompletionIsEntityStreamError: CompletionHandling =
     () =>
       Some(ParserOutput.EntityStreamError(ErrorInfo(

http-core/src/main/scala/org/apache/pekko/http/impl/engine/parsing/HttpRequestParser.scala

@@ -14,12 +14,14 @@
 package org.apache.pekko.http.impl.engine.parsing
 
 import java.lang.{ StringBuilder => JStringBuilder }
+import java.nio.charset.StandardCharsets
 import javax.net.ssl.SSLSession
 
 import scala.annotation.{ switch, tailrec }
 
 import org.apache.pekko
 import pekko.annotation.InternalApi
+import pekko.http.IllegalRequestContext
 import pekko.http.impl.engine.server.HttpAttributes
 import pekko.http.impl.util.ByteStringParserInput
 import pekko.http.impl.util.HttpConstants._
@@ -69,6 +71,8 @@ private[http] final class HttpRequestParser(
       private var method: HttpMethod = null
       private var uri: Uri = null
       private var uriBytes: ByteString = null
+      // whether `protocol` of the underlying parser belongs to the message currently being parsed
+      private var protocolParsed: Boolean = false
 
       override def onPush(): Unit = handleParserOutput(parseSessionBytes(grab(in)))
       override def onPull(): Unit = handleParserOutput(doPull())
@@ -89,9 +93,16 @@ private[http] final class HttpRequestParser(
 
       override def parseMessage(input: ByteString, offset: Int): StateResult =
         if (offset < input.length) {
+          // the fields below are reused for every message on a connection, forget what the previous
+          // one left behind so that a failure cannot report values belonging to another request
+          method = null
+          uri = null
+          uriBytes = null
+          protocolParsed = false
           var cursor = parseMethod(input, offset)
           cursor = parseRequestTarget(input, cursor)
           cursor = parseProtocol(input, cursor)
+          protocolParsed = true
           if (byteAt(input, cursor) == CR_BYTE && byteAt(input, cursor + 1) == LF_BYTE)
             parseHeaderLines(input, cursor + 2)
           else if (byteAt(input, cursor) == LF_BYTE)
@@ -252,6 +263,12 @@ private[http] final class HttpRequestParser(
           }
         } else failMessageStart("Request is missing required `Host` header")
 
+      override protected def illegalRequestContext: IllegalRequestContext =
+        IllegalRequestContext(
+          Option(method),
+          Option.unless(uriBytes eq null)(uriBytes.decodeString(StandardCharsets.US_ASCII)),
+          Option.when(protocolParsed)(currentProtocol))
+
       private def remoteAddressStr: String =
         inheritedAttributes.get[HttpAttributes.RemoteAddress].map(_.address) match {
           case Some(addr) => s" from ${addr.getHostString}:${addr.getPort}"

http-core/src/main/scala/org/apache/pekko/http/impl/engine/parsing/ParserOutput.scala

@@ -16,6 +16,7 @@ package org.apache.pekko.http.impl.engine.parsing
 import org.apache.pekko
 import pekko.NotUsed
 import pekko.annotation.InternalApi
+import pekko.http.IllegalRequestContext
 import pekko.http.impl.util.StreamUtils
 import pekko.http.scaladsl.model._
 import pekko.stream.scaladsl.Source
@@ -62,7 +63,10 @@ private[http] object ParserOutput {
 
   final case class EntityChunk(chunk: HttpEntity.ChunkStreamPart) extends MessageOutput
 
-  final case class MessageStartError(status: StatusCode, info: ErrorInfo) extends MessageStart with ErrorOutput
+  final case class MessageStartError(
+      status: StatusCode,
+      info: ErrorInfo,
+      context: IllegalRequestContext = IllegalRequestContext.empty) extends MessageStart with ErrorOutput
 
   final case class EntityStreamError(info: ErrorInfo) extends ErrorOutput
 

http-core/src/main/scala/org/apache/pekko/http/impl/engine/server/HttpServerBluePrint.scala

@@ -26,7 +26,7 @@ import pekko.stream._
 import pekko.stream.TLSProtocol._
 import pekko.stream.scaladsl._
 import pekko.stream.stage._
-import pekko.http.ParsingErrorHandler
+import pekko.http.{ IllegalRequestContext, ParsingErrorHandler }
 import pekko.http.scaladsl.settings.ServerSettings
 import pekko.http.impl.engine.parsing.ParserOutput._
 import pekko.http.impl.engine.parsing._
@@ -257,21 +257,25 @@ private[http] object HttpServerBluePrint {
     def establishAbsoluteUri(requestOutput: RequestOutput): RequestOutput = requestOutput match {
       case connect: RequestStart if connect.method == HttpMethods.CONNECT =>
         MessageStartError(StatusCodes.BadRequest,
-          ErrorInfo(s"CONNECT requests are not supported", s"Rejecting CONNECT request to '${connect.uri}'"))
+          ErrorInfo(s"CONNECT requests are not supported", s"Rejecting CONNECT request to '${connect.uri}'"),
+          contextOf(connect))
       case start: RequestStart =>
         try {
           val effectiveUri = HttpRequest.effectiveUri(start.uri, start.headers, isSecureConnection, defaultHostHeader)
           start.copy(uri = effectiveUri)
         } catch {
           case e: IllegalUriException =>
-            MessageStartError(StatusCodes.BadRequest, e.info)
+            MessageStartError(StatusCodes.BadRequest, e.info, contextOf(start))
         }
       case x => x
     }
 
     Flow[SessionBytes].via(rootParser).map(establishAbsoluteUri)
   }
 
+  private def contextOf(start: RequestStart): IllegalRequestContext =
+    IllegalRequestContext(Some(start.method), Some(start.uri.toString), Some(start.protocol))
+
   def rendering(settings: ServerSettings, log: LoggingAdapter, dateHeaderRendering: DateHeaderRendering)
       : Flow[ResponseRenderingContext, ResponseRenderingOutput, NotUsed] = {
     import settings._
@@ -464,7 +468,8 @@ private[http] object HttpServerBluePrint {
                 case MessageEnd =>
                   messageEndPending = false
                   push(requestPrepOut, MessageEnd)
-                case MessageStartError(status, info)                                   => finishWithIllegalRequestError(status, info)
+                case MessageStartError(status, info, context) =>
+                  finishWithIllegalRequestError(status, info, context)
                 case x: EntityStreamError if messageEndPending && openRequests.isEmpty =>
                   // client terminated the connection after receiving an early response to 100-continue
                   completeStage()
@@ -567,8 +572,9 @@ private[http] object HttpServerBluePrint {
             }
           })
 
-        def finishWithIllegalRequestError(status: StatusCode, info: ErrorInfo): Unit = {
-          val errorResponse = JavaMapping.toScala(parsingErrorHandler.handle(status, info, log, settings))
+        def finishWithIllegalRequestError(status: StatusCode, info: ErrorInfo,
+            context: IllegalRequestContext = IllegalRequestContext.empty): Unit = {
+          val errorResponse = JavaMapping.toScala(parsingErrorHandler.handle(status, info, log, settings, context))
           emitErrorResponse(errorResponse)
         }
 

http-core/src/main/scala/org/apache/pekko/http/impl/engine/ws/WebSocketClientBlueprint.scala

@@ -139,7 +139,7 @@ private[http] object WebSocketClientBlueprint {
                     result.success(InvalidUpgradeResponse(response, s"WebSocket server at $uri returned $problem"))
                     failStage(new IllegalArgumentException(s"WebSocket upgrade did not finish because of '$problem'"))
                 }
-              case MessageStartError(statusCode, errorInfo) =>
+              case MessageStartError(statusCode, errorInfo, _) =>
                 throw new IllegalStateException(s"Message failed with status code $statusCode; Error info: $errorInfo")
               case other =>
                 throw new IllegalStateException(s"unexpected element of type ${other.getClass}")