Require authorization on DAR upload endpoint (#8193)

* Test authentication on upload_dar endpoint

changelog_begin
changelog_end

* require authentication on upload_dar endpoint

* push Directive into auth

* Fully upload request before auth redirection

* Make HTTP entity upload parameters configurable

changelog_begin
changelog_end

* Shorten help message

https://github.com/digital-asset/daml/pull/8193#discussion_r538428368

* maxHttpEntityUploadSize as Long

https://github.com/digital-asset/daml/pull/8193#discussion_r538431773

* use DefaultMaxInboundMessageSize for DefaultMaxHttpEntityUploadSize

Co-authored-by: Andreas Herrmann <andreas.herrmann@tweag.io>

triggers/service/src/main/scala/com/digitalasset/daml/lf/engine/trigger/Server.scala

@@ -59,6 +59,8 @@ import scala.language.postfixOps
 import scala.util.{Failure, Success, Try}
 
 class Server(
+    maxHttpEntityUploadSize: Long,
+    httpEntityUploadTimeout: FiniteDuration,
     authConfig: AuthConfig,
     triggerDao: RunningTriggerDao,
     val logTriggerStatus: (UUID, String) => Unit)(
@@ -268,12 +270,14 @@ class Server(
             }
           }
         }
-        Directive { inner =>
-          auth {
+        auth.flatMap {
           // Authorization successful - pass token to continuation
-            case Some(authorization) => inner(Tuple1(Some(authorization)))
+          case Some(authorization) => provide(Some(authorization))
           // Authorization failed - login and retry on callback request.
-            case None => { ctx =>
+          case None =>
+            // Ensure that the request is fully uploaded.
+            toStrictEntity(httpEntityUploadTimeout, maxHttpEntityUploadSize).tflatMap { _ =>
+              Directive { inner => ctx =>
                 val requestId = UUID.randomUUID()
                 authCallbacks.update(
                   requestId, {
@@ -298,7 +302,8 @@ class Server(
                   .withPath(Path./("cb"))
                 val uri = authUri
                   .withPath(Path./("login"))
-                .withQuery(AuthRequest.Login(callbackUri, claims, Some(requestId.toString)).toQuery)
+                  .withQuery(
+                    AuthRequest.Login(callbackUri, claims, Some(requestId.toString)).toQuery)
                 ctx.redirect(uri, StatusCodes.Found)
               }
             }
@@ -432,6 +437,9 @@ class Server(
       // upload a DAR as a multi-part form request with a single field called
       // "dar".
       post {
+        val claims = Claims(admin = true)
+        authorize(claims)(ec, system) {
+          _ =>
             fileUpload("dar") {
               case (_, byteSource) =>
                 val byteStringF: Future[ByteString] = byteSource.runFold(ByteString(""))(_ ++ _)
@@ -445,7 +453,8 @@ class Server(
                       case Success(dar) =>
                         onComplete(addDar(dar)) {
                           case Failure(err: ParseError) =>
-                        complete(errorResponse(StatusCodes.UnprocessableEntity, err.description))
+                            complete(
+                              errorResponse(StatusCodes.UnprocessableEntity, err.description))
                           case Failure(exception) =>
                             complete(
                               errorResponse(StatusCodes.InternalServerError, exception.description))
@@ -458,6 +467,7 @@ class Server(
                 }
             }
         }
+      }
     },
     path("livez") {
       complete((StatusCodes.OK, JsObject(("status", "pass".toJson))))
@@ -528,6 +538,8 @@ object Server {
   def apply(
       host: String,
       port: Int,
+      maxHttpEntityUploadSize: Long,
+      httpEntityUploadTimeout: FiniteDuration,
       authConfig: AuthConfig,
       ledgerConfig: LedgerConfig,
       restartConfig: TriggerRestartConfig,
@@ -548,11 +560,21 @@ object Server {
     val (dao, server, initializeF): (RunningTriggerDao, Server, Future[Unit]) = jdbcConfig match {
       case None =>
         val dao = InMemoryTriggerDao()
-        val server = new Server(authConfig, dao, logTriggerStatus)
+        val server = new Server(
+          maxHttpEntityUploadSize,
+          httpEntityUploadTimeout,
+          authConfig,
+          dao,
+          logTriggerStatus)
         (dao, server, Future.successful(()))
       case Some(c) =>
         val dao = DbTriggerDao(c)
-        val server = new Server(authConfig, dao, logTriggerStatus)
+        val server = new Server(
+          maxHttpEntityUploadSize,
+          httpEntityUploadTimeout,
+          authConfig,
+          dao,
+          logTriggerStatus)
         val initialize = for {
           _ <- dao.initialize
           packages <- dao.readPackages

triggers/service/src/main/scala/com/digitalasset/daml/lf/engine/trigger/ServiceConfig.scala

@@ -26,6 +26,8 @@ private[trigger] final case class ServiceConfig(
     maxInboundMessageSize: Int,
     minRestartInterval: FiniteDuration,
     maxRestartInterval: FiniteDuration,
+    maxHttpEntityUploadSize: Long,
+    httpEntityUploadTimeout: FiniteDuration,
     timeProviderType: TimeProviderType,
     commandTtl: Duration,
     init: Boolean,
@@ -81,6 +83,8 @@ private[trigger] object ServiceConfig {
   val DefaultMaxInboundMessageSize: Int = RunnerConfig.DefaultMaxInboundMessageSize
   private val DefaultMinRestartInterval: FiniteDuration = FiniteDuration(5, duration.SECONDS)
   val DefaultMaxRestartInterval: FiniteDuration = FiniteDuration(60, duration.SECONDS)
+  val DefaultMaxHttpEntityUploadSize: Long = RunnerConfig.DefaultMaxInboundMessageSize.toLong
+  val DefaultHttpEntityUploadTimeout: FiniteDuration = FiniteDuration(1, duration.MINUTES)
 
   @SuppressWarnings(Array("org.wartremover.warts.NonUnitStatements")) // scopt builders
   private val parser = new scopt.OptionParser[ServiceConfig]("trigger-service") {
@@ -113,7 +117,7 @@ private[trigger] object ServiceConfig {
       .optional()
       .action((t, c) => c.copy(authUri = Some(Uri(t))))
       .text("Auth middleware URI.")
-      // TODO[AH] Expose once the feature is fully implemented.
+      // TODO[AH] Expose once the auth feature is fully implemented.
       .hidden()
 
     opt[Int]("max-inbound-message-size")
@@ -134,6 +138,21 @@ private[trigger] object ServiceConfig {
       .text(
         s"Maximum time interval between restarting a failed trigger. Defaults to ${DefaultMaxRestartInterval.toSeconds} seconds.")
 
+    opt[Long]("max-http-entity-upload-size")
+      .action((x, c) => c.copy(maxHttpEntityUploadSize = x))
+      .optional()
+      .text(s"Optional max HTTP entity upload size. Defaults to ${DefaultMaxHttpEntityUploadSize}.")
+      // TODO[AH] Expose once the auth feature is fully implemented.
+      .hidden()
+
+    opt[Long]("http-entity-upload-timeout")
+      .action((x, c) => c.copy(httpEntityUploadTimeout = FiniteDuration(x, duration.MINUTES)))
+      .optional()
+      .text(
+        s"Optional HTTP entity upload timeout. Defaults to ${DefaultHttpEntityUploadTimeout.toSeconds} seconds.")
+      // TODO[AH] Expose once the auth feature is fully implemented.
+      .hidden()
+
     opt[Unit]('w', "wall-clock-time")
       .action { (_, c) =>
         c.copy(timeProviderType = TimeProviderType.WallClock)
@@ -172,6 +191,8 @@ private[trigger] object ServiceConfig {
         maxInboundMessageSize = DefaultMaxInboundMessageSize,
         minRestartInterval = DefaultMinRestartInterval,
         maxRestartInterval = DefaultMaxRestartInterval,
+        maxHttpEntityUploadSize = DefaultMaxHttpEntityUploadSize,
+        httpEntityUploadTimeout = DefaultHttpEntityUploadTimeout,
         timeProviderType = TimeProviderType.Static,
         commandTtl = Duration.ofSeconds(30L),
         init = false,

triggers/service/src/main/scala/com/digitalasset/daml/lf/engine/trigger/ServiceMain.scala

@@ -35,6 +35,8 @@ object ServiceMain {
   def startServer(
       host: String,
       port: Int,
+      maxHttpEntityUploadSize: Long,
+      httpEntityUploadTimeout: FiniteDuration,
       authConfig: AuthConfig,
       ledgerConfig: LedgerConfig,
       restartConfig: TriggerRestartConfig,
@@ -48,6 +50,8 @@ object ServiceMain {
         Server(
           host,
           port,
+          maxHttpEntityUploadSize,
+          httpEntityUploadTimeout,
           authConfig,
           ledgerConfig,
           restartConfig,
@@ -119,6 +123,8 @@ object ServiceMain {
             Server(
               config.address,
               config.httpPort,
+              config.maxHttpEntityUploadSize,
+              config.httpEntityUploadTimeout,
               authConfig,
               ledgerConfig,
               restartConfig,

triggers/service/src/test/scala/com/digitalasset/daml/lf/engine/trigger/TriggerServiceFixture.scala

@@ -458,6 +458,8 @@ trait TriggerServiceFixture
               r <- ServiceMain.startServer(
                 host.getHostName,
                 lock.port.value,
+                ServiceConfig.DefaultMaxHttpEntityUploadSize,
+                ServiceConfig.DefaultHttpEntityUploadTimeout,
                 authConfig,
                 ledgerConfig,
                 restartConfig,

triggers/service/src/test/scala/com/digitalasset/daml/lf/engine/trigger/TriggerServiceTest.scala

@@ -142,7 +142,7 @@ trait AbstractTriggerServiceTest
       uri = uri.withPath(Uri.Path(s"/v1/packages")),
       entity = multipartForm.toEntity
     )
-    Http().singleRequest(req)
+    httpRequestFollow(req)
   }
 
   def responseBodyToString(resp: HttpResponse): Future[String] = {
@@ -564,6 +564,14 @@ trait AbstractTriggerServiceTestAuthMiddleware
       } yield succeed
   }
 
+  it should "forbid a non-authorized user to upload a DAR" in withTriggerService(Nil) { uri: Uri =>
+    authServer.revokeAdmin()
+    for {
+      resp <- uploadDar(uri, darPath) // same dar as in initialization
+      _ <- resp.status shouldBe StatusCodes.Forbidden
+    } yield succeed
+  }
+
   it should "request a fresh token after expiry on user request" in withTriggerService(Nil) {
     uri: Uri =>
       for {