Allow reading the request body in both security and main server logic

client/http4s-client/src/main/scala/sttp/tapir/client/http4s/EndpointToHttp4sClient.scala

@@ -100,9 +100,10 @@ private[http4s] class EndpointToHttp4sClient(clientOptions: Http4sClientOptions)
           currentUri.withQueryParam(key, values)
         }
         req.withUri(uri)
-      case EndpointIO.Empty(_, _)              => req
-      case EndpointIO.Body(bodyType, codec, _) => setBody(value, bodyType, codec, req)
-      case ob: EndpointIO.OneOfBody[_, _]      =>
+      case EndpointIO.Empty(_, _)                        => req
+      case b @ EndpointIO.Body(_, _, _) if b.isSecondary => req
+      case EndpointIO.Body(bodyType, codec, _)           => setBody(value, bodyType, codec, req)
+      case ob: EndpointIO.OneOfBody[_, _]                =>
         ob.headVariantBodyWithAppliedMapping match {
           case Some(Left(body))                                                                 => setInputParams(body, params, req)
           case Some(Right(EndpointIO.StreamBodyWrapper(StreamBodyIO(streams, codec, _, _, _)))) =>

client/play-client/src/main/scala/sttp/tapir/client/play/EndpointToPlayClient.scala

@@ -115,8 +115,9 @@ private[play] class EndpointToPlayClient(clientOptions: PlayClientOptions, ws: S
       case EndpointInput.QueryParams(codec, _) =>
         val mqp = codec.encode(value)
         req.addQueryStringParameters(mqp.toSeq*)
-      case EndpointIO.Empty(_, _)              => req
-      case EndpointIO.Body(bodyType, codec, _) =>
+      case EndpointIO.Empty(_, _)                        => req
+      case b @ EndpointIO.Body(_, _, _) if b.isSecondary => req
+      case EndpointIO.Body(bodyType, codec, _)           =>
         val req2 = setBody(value, bodyType, codec, req)
         req2
       case ob: EndpointIO.OneOfBody[_, _] =>

client/play29-client/src/main/scala/sttp/tapir/client/play/EndpointToPlayClient.scala

@@ -115,8 +115,9 @@ private[play] class EndpointToPlayClient(clientOptions: PlayClientOptions, ws: S
       case EndpointInput.QueryParams(codec, _) =>
         val mqp = codec.encode(value)
         req.addQueryStringParameters(mqp.toSeq*)
-      case EndpointIO.Empty(_, _)              => req
-      case EndpointIO.Body(bodyType, codec, _) =>
+      case EndpointIO.Empty(_, _)                        => req
+      case b @ EndpointIO.Body(_, _, _) if b.isSecondary => req
+      case EndpointIO.Body(bodyType, codec, _)           =>
         val req2 = setBody(value, bodyType, codec, req)
         req2
       case ob: EndpointIO.OneOfBody[_, _] =>

client/sttp-client/src/main/scala/sttp/tapir/client/sttp/EndpointToSttpClient.scala

@@ -91,8 +91,9 @@ private[sttp] class EndpointToSttpClient[R](clientOptions: SttpClientOptions, ws
         val mqp = codec.encode(value)
         val uri2 = uri.addParams(mqp.toSeq*)
         (uri2, req)
-      case EndpointIO.Empty(_, _)              => (uri, req)
-      case EndpointIO.Body(bodyType, codec, _) =>
+      case EndpointIO.Empty(_, _)                        => (uri, req)
+      case b @ EndpointIO.Body(_, _, _) if b.isSecondary => (uri, req)
+      case EndpointIO.Body(bodyType, codec, _)           =>
         val req2 = setBody(value, bodyType, codec, req)
         (uri, req2)
       case ob: EndpointIO.OneOfBody[_, _] =>

client/sttp-client4/src/main/scala/sttp/tapir/client/sttp4/EndpointToSttpClientBase.scala

@@ -108,8 +108,9 @@ private[sttp4] trait EndpointToSttpClientBase {
         val mqp = codec.encode(value)
         val uri2 = uri.addParams(mqp.toSeq*)
         (uri2, req, streamBody)
-      case EndpointIO.Empty(_, _)              => (uri, req, streamBody)
-      case EndpointIO.Body(bodyType, codec, _) =>
+      case EndpointIO.Empty(_, _)                        => (uri, req, streamBody)
+      case b @ EndpointIO.Body(_, _, _) if b.isSecondary => (uri, req, streamBody)
+      case EndpointIO.Body(bodyType, codec, _)           =>
         val req2 = setBody(value, bodyType, codec, req)
         (uri, req2, streamBody)
       case ob: EndpointIO.OneOfBody[_, _] =>

client/tests/src/main/scala/sttp/tapir/client/tests/ClientBasicTests.scala

@@ -36,6 +36,7 @@ trait ClientBasicTests { this: ClientTests[Any] =>
       Right("orange/grapefruit 10 None")
     )
     testClient(in_string_out_string, (), "delicious", Right("delicious"))
+    testClient(in_security_string_secondary_string_out_string, "sent", "ignored", Right("sent"))
     testClient(in_json_out_json, (), FruitAmount("orange", 11), Right(FruitAmount("orange", 11)))
     testClient(in_byte_array_out_byte_array, (), "banana kiwi".getBytes(), Right("banana kiwi".getBytes()))
     testClient(in_byte_buffer_out_byte_buffer, (), ByteBuffer.wrap("mango".getBytes), Right(ByteBuffer.wrap("mango".getBytes)))

core/src/main/scala/sttp/tapir/EndpointIO.scala

@@ -12,8 +12,10 @@ import sttp.tapir.model.{ServerRequest, StatusCodeRange}
 import sttp.tapir.typelevel.ParamConcat
 import sttp.ws.WebSocketFrame
 
+import java.io.InputStream
+import java.nio.ByteBuffer
 import java.nio.charset.{Charset, StandardCharsets}
-import scala.annotation.StaticAnnotation
+import scala.annotation.{StaticAnnotation, implicitNotFound}
 import scala.collection.immutable.{ListMap, Seq}
 import scala.concurrent.duration.FiniteDuration
 
@@ -486,13 +488,51 @@ object EndpointIO {
     override private[tapir] type L = R
     override private[tapir] type CF = CodecFormat
     override private[tapir] def copyWith[U](c: Codec[R, U, CodecFormat], i: Info[U]): Body[R, U] = copy(codec = c, info = i)
+
+    /** Marks this as a secondary body definition: still decoded on the server, but not part of the API contract - excluded from the
+      * documentation and ignored by client interpreters. Lets the request body be decoded a second time, e.g. once in `serverSecurityLogic`
+      * and once in the main logic. Only bodies which can be re-read from buffered bytes may be secondary.
+      */
+    def asSecondary(implicit ev: Body.ReplayableRawBody[R]): Body[R, T] = {
+      val _ = ev
+      attribute(SecondaryBody.Attribute, SecondaryBody())
+    }
+
+    /** Whether this body is marked with [[asSecondary]]. */
+    def isSecondary: Boolean =
+      // attribute keys are equal by type name, so an unrelated key with the same name mustn't mark the body as secondary
+      info.attribute(SecondaryBody.Attribute).contains(SecondaryBody())
+
     override def show: String = {
       val charset = bodyType.asInstanceOf[RawBodyType[?]] match {
         case RawBodyType.StringBody(charset) => s" (${charset.toString})"
         case _                               => ""
       }
       val format = codec.format.mediaType
-      s"{body as $format$charset}"
+      val secondary = if (isSecondary) "secondary " else ""
+      s"{${secondary}body as $format$charset}"
+    }
+  }
+
+  object Body {
+
+    /** Evidence that a raw body type can be re-read from buffered bytes, and is therefore usable as a secondary body. */
+    @implicitNotFound(
+      "Cannot use a body with raw type ${R} as a secondary body. Only bodies which can be re-read from buffered bytes " +
+        "are supported: string, byte array, byte buffer, input stream, input stream range. File and multipart bodies " +
+        "cannot be read twice."
+    )
+    sealed trait ReplayableRawBody[R]
+
+    object ReplayableRawBody {
+      private val instance: ReplayableRawBody[Any] = new ReplayableRawBody[Any] {}
+      private def of[R]: ReplayableRawBody[R] = instance.asInstanceOf[ReplayableRawBody[R]]
+
+      implicit val forString: ReplayableRawBody[String] = of
+      implicit val forByteArray: ReplayableRawBody[Array[Byte]] = of
+      implicit val forByteBuffer: ReplayableRawBody[ByteBuffer] = of
+      implicit val forInputStream: ReplayableRawBody[InputStream] = of
+      implicit val forInputStreamRange: ReplayableRawBody[InputStreamRange] = of
     }
   }
 

core/src/main/scala/sttp/tapir/internal/EndpointBodyVerifier.scala

@@ -0,0 +1,121 @@
+package sttp.tapir.internal
+
+import sttp.tapir.{AnyEndpoint, EndpointIO, EndpointInput}
+
+private[tapir] case class EndpointBodyProblems(errors: List[String], warnings: List[String])
+
+/** Finds request body definitions which can't be served (errors), or which can but probably aren't what was intended (warnings). Errors are
+  * thrown by server interpreters when routes are constructed; both are reported by `EndpointVerifier`.
+  */
+private[tapir] object EndpointBodyVerifier {
+  def verifyOrThrow(endpoints: List[AnyEndpoint]): Unit = {
+    val errors = endpoints.flatMap(verifyOne(_).errors)
+    if (errors.nonEmpty) throw new IllegalArgumentException(errors.mkString("\n"))
+  }
+
+  def verifyOne(endpoint: AnyEndpoint): EndpointBodyProblems = {
+    val securityInputs = endpoint.securityInput.asVectorOfBasicInputs()
+    val inInputs = endpoint.input.asVectorOfBasicInputs()
+    val inputs = securityInputs ++ inInputs
+
+    val secondary = inputs.collect { case b: EndpointIO.Body[?, ?] if b.isSecondary => b }
+    def bodiesOf(basics: Vector[EndpointInput.Basic[?]]): Vector[EndpointInput.Basic[?]] = basics.collect {
+      case b: EndpointIO.Body[?, ?]              => b
+      case b: EndpointIO.OneOfBody[?, ?]         => b
+      case b: EndpointIO.StreamBodyWrapper[?, ?] => b
+    }
+    def isPrimary(body: EndpointInput.Basic[?]): Boolean = body match {
+      case b: EndpointIO.Body[?, ?] => !b.isSecondary
+      case _                        => true
+    }
+    val securityBodies = bodiesOf(securityInputs)
+    val inBodies = bodiesOf(inInputs)
+    val securityPrimaryBodies = securityBodies.filter(isPrimary)
+    val inPrimaryBodies = inBodies.filter(isPrimary)
+    val primaryBodies = securityPrimaryBodies ++ inPrimaryBodies
+    def asAtoms(body: EndpointInput.Basic[?]): Vector[EndpointInput.Basic[?]] = body match {
+      case ob: EndpointIO.OneOfBody[?, ?] => ob.variants.map(_.bodyAsAtom).toVector
+      case other                          => Vector(other)
+    }
+    val primaryBodyAtoms: Vector[EndpointInput.Basic[?]] = primaryBodies.flatMap(asAtoms)
+    val streamingPrimary = primaryBodyAtoms.exists(_.isInstanceOf[EndpointIO.StreamBodyWrapper[?, ?]])
+    val fileOrMultipartPrimary = primaryBodyAtoms.exists {
+      case b: EndpointIO.Body[?, ?] => !isReplayable(b.bodyType)
+      case _                        => false
+    }
+    lazy val shown = endpoint.showShort
+
+    // asSecondary can be called on a variant, as oneOfBody takes bodies, but the server interpreters only look for
+    // the marker on a top-level body input - so accepting it here would silently fall back to reading the body once
+    val secondaryInsideOneOfBody: List[String] =
+      inputs.collect {
+        case ob: EndpointIO.OneOfBody[?, ?] if asAtoms(ob).exists { case b: EndpointIO.Body[?, ?] => b.isSecondary; case _ => false } =>
+          s"Endpoint $shown marks a `oneOfBody` variant as secondary. Only a body input used on its own can be " +
+            s"secondary; a `oneOfBody` is always part of the API contract."
+      }.toList
+
+    val secondaryOutputs: List[String] =
+      List(endpoint.output, endpoint.errorOutput)
+        .flatMap(_.traverseOutputs[EndpointIO.Body[?, ?]] {
+          case b: EndpointIO.Body[?, ?] if b.isSecondary => Vector(b)
+          case ob: EndpointIO.OneOfBody[?, ?]            =>
+            ob.variants.map(_.bodyAsAtom).collect { case b: EndpointIO.Body[?, ?] if b.isSecondary => b }.toVector
+        })
+        .map(b => s"Endpoint $shown marks the response body ${b.show} as secondary. Only request bodies can be secondary.")
+
+    val hiddenSchemaHint =
+      if (primaryBodies.exists { case b: EndpointIO.Body[?, ?] => b.codec.schema.hidden; case _ => false })
+        " A hidden schema doesn't make a body secondary."
+      else ""
+
+    val tooManyBodies: List[String] =
+      if (securityPrimaryBodies.nonEmpty && inPrimaryBodies.nonEmpty)
+        List(
+          s"Endpoint $shown declares a request body in both `securityIn` and `in`. Only one may be part of the API " +
+            s"contract. If both should decode the same request body, mark one of them with `.asSecondary`." +
+            hiddenSchemaHint
+        )
+      else if (securityBodies.size > 1)
+        List(
+          s"Endpoint $shown declares more than one request body in `securityIn`. Each of `securityIn` and `in` may " +
+            s"declare at most one body, secondary or not.$hiddenSchemaHint"
+        )
+      else if (inBodies.size > 1)
+        List(
+          s"Endpoint $shown declares more than one request body in `in`. Each of `securityIn` and `in` may declare at " +
+            s"most one body, secondary or not.$hiddenSchemaHint"
+        )
+      else Nil
+
+    val nonReplayablePrimaryKinds = List(streamingPrimary -> "streaming", fileOrMultipartPrimary -> "file or multipart").collect {
+      case (true, kind) => kind
+    }
+    val nonReplayablePrimaryWithSecondary =
+      if (secondary.nonEmpty)
+        nonReplayablePrimaryKinds.map { kind =>
+          s"Endpoint $shown combines a $kind body with a secondary body. The request body is buffered to decode the " +
+            s"secondary body, so it can't also be read as a $kind body."
+        }
+      else Nil
+
+    val secondaryWithoutPrimary =
+      if (secondary.nonEmpty && primaryBodies.isEmpty)
+        List(
+          s"Endpoint $shown reads a secondary request body, but no request body is part of the API contract: it " +
+            s"will be absent from the documentation and clients will not send it. Either declare an ordinary body " +
+            s"as well, or drop `.asSecondary` and use the body input directly."
+        )
+      else Nil
+
+    val uselessMetadata =
+      secondary.filter(b => b.info.description.isDefined || b.info.examples.nonEmpty).map { b =>
+        s"Endpoint $shown sets a description or example on the secondary body ${b.show}, which never reaches the " +
+          s"documentation, as secondary bodies are excluded from it."
+      }
+
+    EndpointBodyProblems(
+      errors = secondaryInsideOneOfBody ++ secondaryOutputs ++ tooManyBodies ++ nonReplayablePrimaryWithSecondary,
+      warnings = secondaryWithoutPrimary ++ uselessMetadata
+    )
+  }
+}

core/src/main/scala/sttp/tapir/internal/SecondaryBody.scala

@@ -0,0 +1,9 @@
+package sttp.tapir.internal
+
+import sttp.tapir.AttributeKey
+
+private[tapir] case class SecondaryBody()
+
+private[tapir] object SecondaryBody {
+  val Attribute: AttributeKey[SecondaryBody] = new AttributeKey[SecondaryBody]("sttp.tapir.internal.SecondaryBody")
+}

core/src/main/scala/sttp/tapir/internal/package.scala

@@ -360,4 +360,11 @@ package object internal {
     case null          => true
     case _             => false
   }
+
+  def isReplayable(bodyType: RawBodyType[?]): Boolean = bodyType match {
+    case RawBodyType.StringBody(_) | RawBodyType.ByteArrayBody | RawBodyType.ByteBufferBody | RawBodyType.InputStreamBody |
+        RawBodyType.InputStreamRangeBody =>
+      true
+    case RawBodyType.FileBody | _: RawBodyType.MultipartBody => false
+  }
 }

core/src/test/scala/sttp/tapir/SecondaryBodyTest.scala

@@ -0,0 +1,22 @@
+package sttp.tapir
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class SecondaryBodyTest extends AnyFlatSpec with Matchers {
+  it should "not treat a key built from the same name as a secondary marker" in {
+    fileBody.attribute(new AttributeKey[Unit]("sttp.tapir.internal.SecondaryBody"), ()).isSecondary shouldBe false
+  }
+
+  it should "not compile for file bodies" in {
+    assertDoesNotCompile("fileBody.asSecondary")
+  }
+
+  it should "not compile for multipart bodies" in {
+    assertDoesNotCompile("multipartBody.asSecondary")
+  }
+
+  it should "render a secondary body distinctly in show" in {
+    stringBody.asSecondary.show shouldBe "{secondary body as text/plain (UTF-8)}"
+  }
+}