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,7 +100,9 @@ private[http4s] class EndpointToHttp4sClient(clientOptions: Http4sClientOptions)
           currentUri.withQueryParam(key, values)
         }
         req.withUri(uri)
-      case EndpointIO.Empty(_, _)              => req
+      case EndpointIO.Empty(_, _)                        => req
+      case b @ EndpointIO.Body(_, _, _) if b.isSecondary =>
+        req // decoded server-side only; not part of the request the client sends
       case EndpointIO.Body(bodyType, codec, _) => setBody(value, bodyType, codec, req)
       case ob: EndpointIO.OneOfBody[_, _]      =>
         ob.headVariantBodyWithAppliedMapping match {

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

@@ -115,7 +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.Empty(_, _)                        => req
+      case b @ EndpointIO.Body(_, _, _) if b.isSecondary =>
+        req // decoded server-side only; not part of the request the client sends
       case EndpointIO.Body(bodyType, codec, _) =>
         val req2 = setBody(value, bodyType, codec, req)
         req2

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

@@ -115,7 +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.Empty(_, _)                        => req
+      case b @ EndpointIO.Body(_, _, _) if b.isSecondary =>
+        req // decoded server-side only; not part of the request the client sends
       case EndpointIO.Body(bodyType, codec, _) =>
         val req2 = setBody(value, bodyType, codec, req)
         req2

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

@@ -91,7 +91,10 @@ 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.Empty(_, _)                        => (uri, req)
+      case b @ EndpointIO.Body(_, _, _) if b.isSecondary =>
+        // decoded server-side only; not part of the request the client sends
+        (uri, req)
       case EndpointIO.Body(bodyType, codec, _) =>
         val req2 = setBody(value, bodyType, codec, req)
         (uri, req2)

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

@@ -108,7 +108,10 @@ 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.Empty(_, _)                        => (uri, req, streamBody)
+      case b @ EndpointIO.Body(_, _, _) if b.isSecondary =>
+        // decoded server-side only; not part of the request the client sends
+        (uri, req, streamBody)
       case EndpointIO.Body(bodyType, codec, _) =>
         val req2 = setBody(value, bodyType, codec, req)
         (uri, req2, streamBody)

client/sttp-client4/src/test/scala/sttp/tapir/client/sttp4/SecondaryBodyClientTest.scala

@@ -0,0 +1,24 @@
+package sttp.tapir.client.sttp4
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import sttp.client4.Request
+import sttp.model.Uri._
+import sttp.tapir._
+
+class SecondaryBodyClientTest extends AnyFlatSpec with Matchers {
+  it should "send only the primary body, ignoring the secondary one" in {
+    // the secondary body is on `in`, processed after `securityIn` - so an unskipped one would overwrite the primary
+    val e = endpoint.post
+      .in("people")
+      .securityIn(stringBody)
+      .in(stringBody.asSecondary)
+      .out(stringBody)
+
+    val request: Request[_] =
+      SttpClientInterpreter().toSecureRequestThrowDecodeFailures(e, Some(uri"http://example.com"))("sent")("ignored")
+
+    request.body.show should include("sent")
+    request.body.show should not include ("ignored")
+  }
+}

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

@@ -486,13 +486,26 @@ 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: ReplayableRawBody[R]): Body[R, T] = {
+      val _ = ev
+      attribute(SecondaryBody.attributeKey, SecondaryBody())
+    }
+
+    def isSecondary: Boolean = info.attribute(SecondaryBody.attributeKey).isDefined
+
     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}"
     }
   }
 

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

@@ -0,0 +1,36 @@
+package sttp.tapir
+
+import java.io.InputStream
+import java.nio.ByteBuffer
+import scala.annotation.implicitNotFound
+
+/** Attribute value marking a body input as a secondary definition: decoded from the request on the server, but not part of the API
+  * contract. Secondary bodies are excluded from documentation and ignored by client interpreters, which allows the request body to be
+  * decoded more than once - e.g. in `serverSecurityLogic` and again in the main logic.
+  *
+  * Set using [[EndpointIO.Body.asSecondary]].
+  */
+case class SecondaryBody()
+
+object SecondaryBody {
+  val attributeKey: AttributeKey[SecondaryBody] = new AttributeKey[SecondaryBody]("sttp.tapir.SecondaryBody")
+}
+
+/** 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. File, multipart and streaming bodies cannot be " +
+    "read twice."
+)
+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/package.scala

@@ -360,4 +360,9 @@ package object internal {
     case null          => true
     case _             => false
   }
+
+  def isSecondaryBodyInput(input: EndpointInput[?]): Boolean = input match {
+    case b: EndpointIO.Body[?, ?] => b.isSecondary
+    case _                        => false
+  }
 }

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

@@ -0,0 +1,127 @@
+package sttp.tapir.server
+
+import sttp.model.Method
+import sttp.tapir.internal._
+import sttp.tapir.{AnyEndpoint, EndpointIO, EndpointInput, RawBodyType}
+
+/** Errors make an endpoint unserveable; warnings describe one that works, but whose published contract probably isn't what was intended. */
+private[tapir] case class EndpointBodyProblems(errors: List[String], warnings: List[String]) {
+  def ++(other: EndpointBodyProblems): EndpointBodyProblems =
+    EndpointBodyProblems(errors ++ other.errors, warnings ++ other.warnings)
+}
+
+private[tapir] object EndpointBodyProblems {
+  val Empty: EndpointBodyProblems = EndpointBodyProblems(Nil, Nil)
+}
+
+/** Verifies that endpoint descriptions are structurally serveable. Run by server interpreters when routes are constructed; warnings are not
+  * logged anywhere, so call this directly to assert on them.
+  */
+private[tapir] object EndpointBodyVerifier {
+  def verify(endpoints: List[AnyEndpoint]): EndpointBodyProblems =
+    endpoints.map(verifyOne).foldLeft(EndpointBodyProblems.Empty)(_ ++ _)
+
+  private[tapir] def throwOnErrors(problems: EndpointBodyProblems): Unit =
+    if (problems.errors.nonEmpty) throw new IllegalArgumentException(problems.errors.mkString("\n"))
+
+  def verifyOne(endpoint: AnyEndpoint): EndpointBodyProblems = {
+    val securityInputs = endpoint.securityInput.asVectorOfBasicInputs()
+    val ordinaryInputs = endpoint.input.asVectorOfBasicInputs()
+    val inputs = securityInputs ++ ordinaryInputs
+
+    val secondary = inputs.collect { case b: EndpointIO.Body[?, ?] if b.isSecondary => b }
+    def primaryBodiesOf(basics: Vector[EndpointInput.Basic[?]]): Vector[EndpointInput.Basic[?]] = basics.collect {
+      case b: EndpointIO.Body[?, ?] if !b.isSecondary => b
+      case b: EndpointIO.OneOfBody[?, ?]              => b
+      case b: EndpointIO.StreamBodyWrapper[?, ?]      => b
+    }
+    val securityPrimaryBodies = primaryBodiesOf(securityInputs)
+    val inPrimaryBodies = primaryBodiesOf(ordinaryInputs)
+    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 nonReplayablePrimary = primaryBodyAtoms.exists {
+      case b: EndpointIO.Body[?, ?] =>
+        b.bodyType match {
+          case RawBodyType.FileBody         => true
+          case _: RawBodyType.MultipartBody => true
+          case _                            => false
+        }
+      case _ => false
+    }
+    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[?, ?] => ob }
+        .collect {
+          case ob if ob.variants.map(_.bodyAsAtom).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 tooManyPrimaries: List[String] =
+      if (secondaryInsideOneOfBody.nonEmpty) Nil
+      else 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 the securityIn one: " +
+            s"stringBody.asSecondary."
+        )
+      else if (securityPrimaryBodies.size > 1)
+        List(
+          s"Endpoint $shown declares more than one request body in securityIn. Only one request body may be part " +
+            s"of the API contract."
+        )
+      else if (inPrimaryBodies.size > 1)
+        List(
+          s"Endpoint $shown declares more than one request body in in. Only one request body may be part of the " +
+            s"API contract."
+        )
+      else Nil
+
+    val streamWithSecondary =
+      if (streamingPrimary && secondary.nonEmpty)
+        List(
+          s"Endpoint $shown combines a streaming body with a secondary body. The request body can either be " +
+            s"streamed lazily or buffered for repeated reads, not both."
+        )
+      else Nil
+
+    val nonReplayableWithSecondary =
+      if (nonReplayablePrimary && secondary.nonEmpty)
+        List(
+          s"Endpoint $shown combines a file or multipart body with a secondary body. Reading the secondary body " +
+            s"consumes the request; the file or multipart body would then be read from an already-drained request."
+        )
+      else Nil
+
+    val bodyCarryingMethod = endpoint.method.exists(m => m == Method.POST || m == Method.PUT || m == Method.PATCH)
+    val secondaryWithoutPrimary =
+      if (secondary.nonEmpty && primaryBodies.isEmpty && bodyCarryingMethod)
+        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 the body in `in` " +
+            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 ++ tooManyPrimaries ++ streamWithSecondary ++ nonReplayableWithSecondary,
+      warnings = (secondaryWithoutPrimary ++ uselessMetadata).toList
+    )
+  }
+}

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

@@ -0,0 +1,57 @@
+package sttp.tapir
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+class SecondaryBodyTest extends AnyFlatSpec with Matchers {
+  it should "mark a string body as secondary" in {
+    stringBody.asSecondary.attribute(SecondaryBody.attributeKey) shouldBe Some(SecondaryBody())
+  }
+
+  it should "mark a json-style string body as secondary" in {
+    val body = stringBodyUtf8AnyFormat(Codec.string)
+    body.asSecondary.attribute(SecondaryBody.attributeKey) shouldBe Some(SecondaryBody())
+  }
+
+  it should "leave a plain body unmarked" in {
+    stringBody.attribute(SecondaryBody.attributeKey) shouldBe None
+    stringBody.isSecondary shouldBe false
+  }
+
+  it should "report a marked body through isSecondary" in {
+    stringBody.asSecondary.isSecondary shouldBe true
+  }
+
+  it should "preserve the codec and body type" in {
+    val secondary = byteArrayBody.asSecondary
+    secondary.bodyType shouldBe RawBodyType.ByteArrayBody
+    secondary.codec shouldBe byteArrayBody.codec
+  }
+
+  it should "not compile for file bodies" in {
+    assertDoesNotCompile("fileBody.asSecondary")
+  }
+
+  it should "not compile for multipart bodies" in {
+    assertDoesNotCompile("multipartBody.asSecondary")
+  }
+
+  it should "not compile for oneOfBody" in {
+    assertDoesNotCompile("""oneOfBody(stringBody, stringBody).asSecondary""")
+  }
+
+  it should "render a secondary body distinctly in show" in {
+    stringBody.asSecondary.show shouldBe "{secondary body as text/plain (UTF-8)}"
+  }
+
+  it should "render a plain body unchanged in show" in {
+    stringBody.show shouldBe "{body as text/plain (UTF-8)}"
+  }
+
+  it should "report secondary bodies through the internal predicate" in {
+    import sttp.tapir.internal._
+    isSecondaryBodyInput(stringBody.asSecondary) shouldBe true
+    isSecondaryBodyInput(stringBody) shouldBe false
+    isSecondaryBodyInput(query[String]("q")) shouldBe false
+  }
+}