MaxContentLength support for Netty pt 2

doc/endpoint/security.md

@@ -49,6 +49,34 @@ will show you a password prompt.
 Optional and multiple authentication inputs have some additional rules as to how hey map to documentation, see the
 ["Authentication inputs and security requirements"](../docs/openapi.md) section in the OpenAPI docs for details.
 
+## Limiting request body length
+
+*Supported backends*: 
+Feature enabled only for Netty-based servers. More backends will be added in the near future.
+
+Individual endpoints can be annotated with content length limit:
+
+```scala mdoc:compile-only
+import sttp.tapir._
+import sttp.tapir.server.model.EndpointExtensions._
+
+val limitedEndpoint = endpoint.maxRequestBodyLength(maxBytes = 163484L)
+```
+
+The `EndpointsExtensions` utility is available in `tapir-server` core module. If you can't depend on it where your endpoint 
+definitions are located, you can directly put an attribute:
+
+```scala mdoc:compile-only
+import sttp.tapir._
+import sttp.tapir.server.model.MaxContentLength
+
+val limitedEndpoint = endpoint.attribute(AttributeKey[MaxContentLength], MaxContentLength(16384L))
+```
+Such protection would prevent loading all the input data into memory if it exceeds the limit. Instead, it will result 
+in a `HTTP 413` response to the client. 
+Please note that in case of endpoints with `streamBody` input type, the server logic receives a reference to a lazily
+evaluated stream, so actual length verification will happen only when the logic performs streams processing, not earlier.
+
 ## Next
 
 Read on about [streaming support](streaming.md).

doc/migrating.md

@@ -1,5 +1,9 @@
 # Migrating
 
+## From 1.9.3 to 1.9.4
+
+- `NettyConfig.defaultNoStreaming` has been removed, use `NettyConfig.default`.
+
 ## From 1.4 to 1.5
 
 - `badRequestOnPathErrorIfPathShapeMatches` and `badRequestOnPathInvalidIfPathShapeMatches` have been removed from `DefaultDecodeFailureHandler`. These flags were causing confusion and incosistencies caused by specifics of ZIO and Play backends. Before tapir 1.5, keeping defaults (`false` and `true` respectively for these flags) meant that some path segment decoding failures (specifically, errors - when an exception has been thrown during decoding, but not for e.g. enumeration mismatches) were translated to a "no-match", meaning that the next endpoint was attempted. From 1.5, tapir defaults to a 400 Bad Request response to be sent instead, on all path decoding failures.

doc/server/netty.md

@@ -80,7 +80,7 @@ NettyFutureServer().port(9090).addEndpoints(???)
 NettyFutureServer(NettyFutureServerOptions.customiseInterceptors.serverLog(None).options)
 
 // customise Netty config
-NettyFutureServer(NettyConfig.defaultNoStreaming.socketBacklog(256))
+NettyFutureServer(NettyConfig.default.socketBacklog(256))
 ```
 
 ## Graceful shutdown
@@ -93,9 +93,9 @@ import sttp.tapir.server.netty.NettyConfig
 import scala.concurrent.duration._
 
 // adjust the waiting time to your needs
-val config = NettyConfig.defaultNoStreaming.withGracefulShutdownTimeout(5.seconds)
+val config = NettyConfig.default.withGracefulShutdownTimeout(5.seconds)
 // or if you don't want the server to wait for in-flight requests
-val config2 = NettyConfig.defaultNoStreaming.noGracefulShutdown
+val config2 = NettyConfig.default.noGracefulShutdown
 ```
 
 ## Domain socket support

generated-doc/out/server/netty.md

@@ -80,7 +80,7 @@ NettyFutureServer().port(9090).addEndpoints(???)
 NettyFutureServer(NettyFutureServerOptions.customiseInterceptors.serverLog(None).options)
 
 // customise Netty config
-NettyFutureServer(NettyConfig.defaultNoStreaming.socketBacklog(256))
+NettyFutureServer(NettyConfig.default.socketBacklog(256))
 ```
 
 ## Graceful shutdown
@@ -93,9 +93,9 @@ import sttp.tapir.server.netty.NettyConfig
 import scala.concurrent.duration._
 
 // adjust the waiting time to your needs
-val config = NettyConfig.defaultNoStreaming.withGracefulShutdownTimeout(5.seconds)
+val config = NettyConfig.default.withGracefulShutdownTimeout(5.seconds)
 // or if you don't want the server to wait for in-flight requests
-val config2 = NettyConfig.defaultNoStreaming.noGracefulShutdown
+val config2 = NettyConfig.default.noGracefulShutdown
 ```
 
 ## Domain socket support

integrations/zio/src/test/scala/sttp/tapir/ztapir/ZTapirTest.scala

@@ -28,7 +28,7 @@ object ZTapirTest extends ZIOSpecDefault with ZTapir {
 
   private val exampleRequestBody = new RequestBody[TestEffect, RequestBodyType] {
     override val streams: Streams[RequestBodyType] = null.asInstanceOf[Streams[RequestBodyType]]
-    override def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R]): TestEffect[RawValue[R]] = ???
+    override def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R], maxBytes: Option[Long]): TestEffect[RawValue[R]] = ???
     override def toStream(serverRequest: ServerRequest, maxBytes: Option[Long]): streams.BinaryStream = ???
   }
 

integrations/zio1/src/test/scala/sttp/tapir/ztapir/ZTapirTest.scala

@@ -30,7 +30,7 @@ object ZTapirTest extends DefaultRunnableSpec with ZTapir {
 
   private val exampleRequestBody = new RequestBody[TestEffect, RequestBodyType] {
     override val streams: Streams[RequestBodyType] = null.asInstanceOf[Streams[RequestBodyType]]
-    override def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R]): TestEffect[RawValue[R]] = ???
+    override def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R], maxBytes: Option[Long]): TestEffect[RawValue[R]] = ???
     override def toStream(serverRequest: ServerRequest, maxBytes: Option[Long]): streams.BinaryStream = ???
   }
 

server/akka-grpc-server/src/main/scala/sttp/tapir/server/akkagrpc/AkkaGrpcRequestBody.scala

@@ -22,7 +22,7 @@ private[akkagrpc] class AkkaGrpcRequestBody(serverOptions: AkkaHttpServerOptions
   private val grpcProtocol = GrpcProtocolNative.newReader(Identity)
 
   override val streams: AkkaStreams = AkkaStreams
-  override def toRaw[R](request: ServerRequest, bodyType: RawBodyType[R]): Future[RawValue[R]] =
+  override def toRaw[R](request: ServerRequest, bodyType: RawBodyType[R], maxBytes: Option[Long]): Future[RawValue[R]] =
     toRawFromEntity(request, akkaRequestEntity(request), bodyType)
 
   override def toStream(request: ServerRequest, maxBytes: Option[Long]): streams.BinaryStream = ???

server/akka-http-server/src/main/scala/sttp/tapir/server/akkahttp/AkkaRequestBody.scala

@@ -21,7 +21,7 @@ private[akkahttp] class AkkaRequestBody(serverOptions: AkkaHttpServerOptions)(im
     ec: ExecutionContext
 ) extends RequestBody[Future, AkkaStreams] {
   override val streams: AkkaStreams = AkkaStreams
-  override def toRaw[R](request: ServerRequest, bodyType: RawBodyType[R]): Future[RawValue[R]] =
+  override def toRaw[R](request: ServerRequest, bodyType: RawBodyType[R], maxBytes: Option[Long]): Future[RawValue[R]] =
     toRawFromEntity(request, akkeRequestEntity(request), bodyType)
   override def toStream(request: ServerRequest, maxBytes: Option[Long]): streams.BinaryStream = {
     val stream = akkeRequestEntity(request).dataBytes

server/armeria-server/src/main/scala/sttp/tapir/server/armeria/ArmeriaRequestBody.scala

@@ -29,7 +29,7 @@ private[armeria] final class ArmeriaRequestBody[F[_], S <: Streams[S]](
       .asInstanceOf[streams.BinaryStream]
   }
 
-  override def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R]): F[RawValue[R]] = {
+  override def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R], maxBytes: Option[Long]): F[RawValue[R]] = {
     val ctx = armeriaCtx(serverRequest)
     val request = ctx.request()
 

server/core/src/main/scala/sttp/tapir/server/interceptor/decodefailure/DecodeFailureHandler.scala

@@ -10,6 +10,7 @@ import sttp.tapir.server.model.ValuedEndpointOutput
 import sttp.tapir.{DecodeResult, EndpointIO, EndpointInput, ValidationError, Validator, server, _}
 
 import scala.annotation.tailrec
+import sttp.capabilities.StreamMaxLengthExceededException
 
 trait DecodeFailureHandler[F[_]] {
 
@@ -122,11 +123,12 @@ object DefaultDecodeFailureHandler {
       case (_: EndpointIO.Header[_], _) => respondBadRequest
       case (fh: EndpointIO.FixedHeader[_], _: DecodeResult.Mismatch) if fh.h.name == HeaderNames.ContentType =>
         respondUnsupportedMediaType
-      case (_: EndpointIO.FixedHeader[_], _)                         => respondBadRequest
-      case (_: EndpointIO.Headers[_], _)                             => respondBadRequest
-      case (_: EndpointIO.Body[_, _], _)                             => respondBadRequest
-      case (_: EndpointIO.OneOfBody[_, _], _: DecodeResult.Mismatch) => respondUnsupportedMediaType
-      case (_: EndpointIO.StreamBodyWrapper[_, _], _)                => respondBadRequest
+      case (_: EndpointIO.FixedHeader[_], _)                               => respondBadRequest
+      case (_: EndpointIO.Headers[_], _)                                   => respondBadRequest
+      case (_, DecodeResult.Error(_, _: StreamMaxLengthExceededException)) => respondPayloadTooLarge
+      case (_: EndpointIO.Body[_, _], _)                                   => respondBadRequest
+      case (_: EndpointIO.OneOfBody[_, _], _: DecodeResult.Mismatch)       => respondUnsupportedMediaType
+      case (_: EndpointIO.StreamBodyWrapper[_, _], _)                      => respondBadRequest
       // we assume that the only decode failure that might happen during path segment decoding is an error
       // a non-standard path decoder might return Missing/Multiple/Mismatch, but that would be indistinguishable from
       // a path shape mismatch
@@ -143,6 +145,7 @@ object DefaultDecodeFailureHandler {
   }
   private val respondBadRequest = Some(onlyStatus(StatusCode.BadRequest))
   private val respondUnsupportedMediaType = Some(onlyStatus(StatusCode.UnsupportedMediaType))
+  private val respondPayloadTooLarge = Some(onlyStatus(StatusCode.PayloadTooLarge))
 
   def respondNotFoundIfHasAuth(
       ctx: DecodeFailureContext,
@@ -224,10 +227,12 @@ object DefaultDecodeFailureHandler {
             }
             .mkString(", ")
         )
-      case Missing        => Some("missing")
-      case Multiple(_)    => Some("multiple values")
-      case Mismatch(_, _) => Some("value mismatch")
-      case _              => None
+      case Missing                                              => Some("missing")
+      case Multiple(_)                                          => Some("multiple values")
+      case Mismatch(_, _)                                       => Some("value mismatch")
+      case Error(_, StreamMaxLengthExceededException(maxBytes)) => Some(s"Content length limit: $maxBytes bytes")
+      case _: Error                                             => None
+      case _: InvalidValue                                      => None
     }
 
     def combineSourceAndDetail(source: String, detail: Option[String]): String =

server/core/src/main/scala/sttp/tapir/server/interpreter/RequestBody.scala

@@ -3,17 +3,12 @@ package sttp.tapir.server.interpreter
 import sttp.capabilities.Streams
 import sttp.model.Part
 import sttp.tapir.model.ServerRequest
-import sttp.tapir.AttributeKey
-import sttp.tapir.EndpointInfo
 import sttp.tapir.{FileRange, RawBodyType, RawPart}
 
-case class MaxContentLength(value: Long)
-
 trait RequestBody[F[_], S] {
   val streams: Streams[S]
-  def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R]): F[RawValue[R]]
+  def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R], maxBytes: Option[Long]): F[RawValue[R]]
   def toStream(serverRequest: ServerRequest, maxBytes: Option[Long]): streams.BinaryStream
-
 }
 
 case class RawValue[R](value: R, createdFiles: Seq[FileRange] = Nil)

server/core/src/main/scala/sttp/tapir/server/interpreter/ServerInterpreter.scala

@@ -1,13 +1,14 @@
 package sttp.tapir.server.interpreter
 
+import sttp.capabilities.StreamMaxLengthExceededException
 import sttp.model.{Headers, StatusCode}
 import sttp.monad.MonadError
 import sttp.monad.syntax._
 import sttp.tapir.internal.{Params, ParamsAsAny, RichOneOfBody}
 import sttp.tapir.model.ServerRequest
-import sttp.tapir.server.{model, _}
 import sttp.tapir.server.interceptor._
-import sttp.tapir.server.model.{ServerResponse, ValuedEndpointOutput}
+import sttp.tapir.server.model.{MaxContentLength, ServerResponse, ValuedEndpointOutput}
+import sttp.tapir.server.{model, _}
 import sttp.tapir.{DecodeResult, EndpointIO, EndpointInput, TapirFile}
 import sttp.tapir.EndpointInfo
 import sttp.tapir.AttributeKey
@@ -157,7 +158,7 @@ class ServerInterpreter[R, F[_], B, S](
         values.bodyInputWithIndex match {
           case Some((Left(oneOfBodyInput), _)) =>
             oneOfBodyInput.chooseBodyToDecode(request.contentTypeParsed) match {
-              case Some(Left(body))                                          => decodeBody(request, values, body)
+              case Some(Left(body))                                          => decodeBody(request, values, body, maxBodyLength)
               case Some(Right(body: EndpointIO.StreamBodyWrapper[Any, Any])) => decodeStreamingBody(request, values, body, maxBodyLength)
               case None                                                      => unsupportedInputMediaTypeResponse(request, oneOfBodyInput)
             }
@@ -182,17 +183,23 @@ class ServerInterpreter[R, F[_], B, S](
   private def decodeBody[RAW, T](
       request: ServerRequest,
       values: DecodeBasicInputsResult.Values,
-      bodyInput: EndpointIO.Body[RAW, T]
+      bodyInput: EndpointIO.Body[RAW, T],
+      maxBodyLength: Option[Long]
   ): F[DecodeBasicInputsResult] = {
-    requestBody.toRaw(request, bodyInput.bodyType).flatMap { v =>
-      bodyInput.codec.decode(v.value) match {
-        case DecodeResult.Value(bodyV) => (values.setBodyInputValue(bodyV): DecodeBasicInputsResult).unit
-        case failure: DecodeResult.Failure =>
-          v.createdFiles
-            .foldLeft(monad.unit(()))((u, f) => u.flatMap(_ => deleteFile(f.file)))
-            .map(_ => DecodeBasicInputsResult.Failure(bodyInput, failure): DecodeBasicInputsResult)
+    requestBody
+      .toRaw(request, bodyInput.bodyType, maxBodyLength)
+      .flatMap { v =>
+        bodyInput.codec.decode(v.value) match {
+          case DecodeResult.Value(bodyV) => (values.setBodyInputValue(bodyV): DecodeBasicInputsResult).unit
+          case failure: DecodeResult.Failure =>
+            v.createdFiles
+              .foldLeft(monad.unit(()))((u, f) => u.flatMap(_ => deleteFile(f.file)))
+              .map(_ => DecodeBasicInputsResult.Failure(bodyInput, failure): DecodeBasicInputsResult)
+        }
+      }
+      .handleError { case e: StreamMaxLengthExceededException =>
+        (DecodeBasicInputsResult.Failure(bodyInput, DecodeResult.Error("", e)): DecodeBasicInputsResult).unit
       }
-    }
   }
 
   private def unsupportedInputMediaTypeResponse(

server/core/src/main/scala/sttp/tapir/server/model/EndpointExtensions.scala

@@ -0,0 +1,33 @@
+package sttp.tapir.server.model
+
+import sttp.tapir.EndpointInfoOps
+import sttp.tapir.AttributeKey
+
+/** Can be used as an endpoint attribute.
+  * @example
+  *   {{{
+  * endpoint.attribute(AttributeKey[MaxContentLength], MaxContentLength(16384L))
+  *   }}}
+  */
+case class MaxContentLength(value: Long) extends AnyVal
+
+object EndpointExtensions {
+  private val MaxContentLengthAttributeKey: AttributeKey[MaxContentLength] = AttributeKey[MaxContentLength]
+
+  implicit class RichServerEndpoint[E <: EndpointInfoOps[_]](e: E) {
+
+    /** Enables checks that prevent loading full request body into memory if it exceeds given limit. Otherwise causes endpoint to reply with
+      * HTTP 413 Payload Too Loarge.
+      *
+      * Please refer to Tapir docs to ensure which backends are supported: https://tapir.softwaremill.com/en/latest/endpoint/security.html
+      * @example
+      *   {{{
+      * endpoint.maxRequestBodyLength(16384L)
+      *   }}}
+      * @param maxBytes
+      *   maximum allowed size of request body in bytes.
+      */
+    def maxRequestBodyLength(maxBytes: Long): E =
+      e.attribute(MaxContentLengthAttributeKey, MaxContentLength(maxBytes)).asInstanceOf[E]
+  }
+}

server/core/src/test/scala/sttp/tapir/server/TestUtil.scala

@@ -14,7 +14,7 @@ import scala.util.{Success, Try}
 object TestUtil {
   object TestRequestBody extends RequestBody[Id, NoStreams] {
     override val streams: Streams[NoStreams] = NoStreams
-    override def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R]): Id[RawValue[R]] = ???
+    override def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R], maxBytes: Option[Long]): Id[RawValue[R]] = ???
     override def toStream(serverRequest: ServerRequest, maxBytes: Option[Long]): streams.BinaryStream = ???
   }
 

server/finatra-server/src/main/scala/sttp/tapir/server/finatra/FinatraRequestBody.scala

@@ -20,7 +20,7 @@ import scala.collection.immutable.Seq
 class FinatraRequestBody(serverOptions: FinatraServerOptions) extends RequestBody[Future, NoStreams] {
   override val streams: NoStreams = NoStreams
 
-  override def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R]): Future[RawValue[R]] = {
+  override def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R], maxBytes: Option[Long]): Future[RawValue[R]] = {
     val request = finatraRequest(serverRequest)
     toRaw(request, bodyType, request.content, request.charset.map(Charset.forName))
   }

server/http4s-server/src/main/scala/sttp/tapir/server/http4s/Http4sRequestBody.scala

@@ -18,7 +18,7 @@ private[http4s] class Http4sRequestBody[F[_]: Async](
     serverOptions: Http4sServerOptions[F]
 ) extends RequestBody[F, Fs2Streams[F]] {
   override val streams: Fs2Streams[F] = Fs2Streams[F]
-  override def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R]): F[RawValue[R]] = {
+  override def toRaw[R](serverRequest: ServerRequest, bodyType: RawBodyType[R], maxBytes: Option[Long]): F[RawValue[R]] = {
     val r = http4sRequest(serverRequest)
     toRawFromStream(serverRequest, r.body, bodyType, r.charset)
   }

server/jdkhttp-server/src/main/scala/sttp/tapir/server/jdkhttp/internal/JdkHttpRequestBody.scala

@@ -18,7 +18,7 @@ private[jdkhttp] class JdkHttpRequestBody(createFile: ServerRequest => TapirFile
     extends RequestBody[Id, NoStreams] {
   override val streams: capabilities.Streams[NoStreams] = NoStreams
 
-  override def toRaw[RAW](serverRequest: ServerRequest, bodyType: RawBodyType[RAW]): RawValue[RAW] = {
+  override def toRaw[RAW](serverRequest: ServerRequest, bodyType: RawBodyType[RAW], maxBytes: Option[Long]): RawValue[RAW] = {
     val request = jdkHttpRequest(serverRequest)
     toRaw(serverRequest, bodyType, request.getRequestBody)
   }

server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/NettyCatsServer.scala

@@ -73,7 +73,7 @@ case class NettyCatsServer[F[_]: Async](routes: Vector[Route[F]], options: Netty
     val channelFuture =
       NettyBootstrap(
         config,
-        new NettyServerHandler(route, unsafeRunAsync, config.maxContentLength, channelGroup, isShuttingDown),
+        new NettyServerHandler(route, unsafeRunAsync, channelGroup, isShuttingDown),
         eventLoopGroup,
         socketOverride
       )
@@ -123,9 +123,9 @@ case class NettyCatsServer[F[_]: Async](routes: Vector[Route[F]], options: Netty
 
 object NettyCatsServer {
   def apply[F[_]: Async](dispatcher: Dispatcher[F]): NettyCatsServer[F] =
-    NettyCatsServer(Vector.empty, NettyCatsServerOptions.default(dispatcher), NettyConfig.defaultWithStreaming)
+    NettyCatsServer(Vector.empty, NettyCatsServerOptions.default(dispatcher), NettyConfig.default)
   def apply[F[_]: Async](options: NettyCatsServerOptions[F]): NettyCatsServer[F] =
-    NettyCatsServer(Vector.empty, options, NettyConfig.defaultWithStreaming)
+    NettyCatsServer(Vector.empty, options, NettyConfig.default)
   def apply[F[_]: Async](dispatcher: Dispatcher[F], config: NettyConfig): NettyCatsServer[F] =
     NettyCatsServer(Vector.empty, NettyCatsServerOptions.default(dispatcher), config)
   def apply[F[_]: Async](options: NettyCatsServerOptions[F], config: NettyConfig): NettyCatsServer[F] =

server/netty-server/cats/src/main/scala/sttp/tapir/server/netty/cats/NettyCatsServerInterpreter.scala

@@ -2,6 +2,7 @@ package sttp.tapir.server.netty.cats
 
 import cats.effect.Async
 import cats.effect.std.Dispatcher
+import internal.Fs2StreamCompatible
 import sttp.capabilities.fs2.Fs2Streams
 import sttp.monad.MonadError
 import sttp.monad.syntax._
@@ -11,6 +12,7 @@ import sttp.tapir.server.interceptor.RequestResult
 import sttp.tapir.server.interceptor.reject.RejectInterceptor
 import sttp.tapir.server.interpreter.{BodyListener, FilterServerEndpoints, ServerInterpreter}
 import sttp.tapir.server.netty.internal.{NettyBodyListener, RunAsync, _}
+import sttp.tapir.server.netty.cats.internal.NettyCatsRequestBody
 import sttp.tapir.server.netty.{NettyResponse, NettyServerRequest, Route}
 
 trait NettyCatsServerInterpreter[F[_]] {
@@ -31,8 +33,8 @@ trait NettyCatsServerInterpreter[F[_]] {
 
     val serverInterpreter = new ServerInterpreter[Fs2Streams[F], F, NettyResponse, Fs2Streams[F]](
       FilterServerEndpoints(ses),
-      new NettyCatsRequestBody(createFile),
-      new NettyCatsToResponseBody(nettyServerOptions.dispatcher, delegate = new NettyToResponseBody),
+      new NettyCatsRequestBody(createFile, Fs2StreamCompatible[F](nettyServerOptions.dispatcher)),
+      new NettyToStreamsResponseBody(Fs2StreamCompatible[F](nettyServerOptions.dispatcher)),
       RejectInterceptor.disableWhenSingleEndpoint(interceptors, ses),
       deleteFile
     )