Support authentication in DAML triggers (#3730)

* Support authentication in DAML triggers

fixes #3259

CHANGELOG_BEGIN

- [DAML Triggers - Experimental] DAML triggers can now be run against
an authenticated ledger.

CHANGELOG_END

* Remove debug printf

* Windows is bad

docs/source/triggers/index.rst

@@ -254,6 +254,9 @@ have created as ``Alice``. Once you archive the ``Subscriber``
 contract, you can see that the ``Copy`` contract will also be
 archived.
 
+When using DAML triggers against a Ledger with authentication, you can
+pass ``--access-token-file token.jwt`` to ``daml trigger`` which will
+read the token from the file ``token.jwt``.
 
 When not to use DAML triggers
 =============================

triggers/runner/BUILD.bazel

@@ -24,6 +24,7 @@ da_scala_library(
         "//language-support/scala/bindings",
         "//language-support/scala/bindings-akka",
         "//ledger-api/rs-grpc-bridge",
+        "//ledger-service/utils",
         "//ledger/ledger-api-common",
         "@maven//:com_github_scopt_scopt_2_12",
         "@maven//:com_typesafe_akka_akka_stream_2_12",

triggers/runner/src/main/scala/com/digitalasset/daml/lf/engine/trigger/RunnerConfig.scala

@@ -3,13 +3,13 @@
 
 package com.digitalasset.daml.lf.engine.trigger
 
-import java.io.File
+import java.nio.file.{Path, Paths}
 import java.time.Duration
 
 import com.digitalasset.platform.services.time.TimeProviderType
 
 case class RunnerConfig(
-    darPath: File,
+    darPath: Path,
     // If true, we will only list the triggers in the DAR and exit.
     listTriggers: Boolean,
     triggerIdentifier: String,
@@ -18,15 +18,16 @@ case class RunnerConfig(
     ledgerParty: String,
     timeProviderType: TimeProviderType,
     commandTtl: Duration,
+    accessTokenFile: Option[Path],
 )
 
 object RunnerConfig {
   private val parser = new scopt.OptionParser[RunnerConfig]("trigger-runner") {
     head("trigger-runner")
 
-    opt[File]("dar")
+    opt[String]("dar")
       .required()
-      .action((f, c) => c.copy(darPath = f))
+      .action((f, c) => c.copy(darPath = Paths.get(f)))
       .text("Path to the dar file containing the trigger")
 
     opt[String]("trigger-name")
@@ -57,6 +58,12 @@ object RunnerConfig {
       }
       .text("TTL in seconds used for commands emitted by the trigger. Defaults to 30s.")
 
+    opt[String]("access-token-file")
+      .action { (f, c) =>
+        c.copy(accessTokenFile = Some(Paths.get(f)))
+      }
+      .text("File from which the access token will be read, required to interact with an authenticated ledger")
+
     cmd("list")
       .action((_, c) => c.copy(listTriggers = true))
       .text("List the triggers in the DAR.")
@@ -93,6 +100,7 @@ object RunnerConfig {
         ledgerParty = null,
         timeProviderType = TimeProviderType.Static,
         commandTtl = Duration.ofSeconds(30L),
+        accessTokenFile = None,
       )
     )
 }

triggers/runner/src/main/scala/com/digitalasset/daml/lf/engine/trigger/RunnerMain.scala

@@ -23,6 +23,7 @@ import com.digitalasset.ledger.client.configuration.{
   LedgerClientConfiguration,
   LedgerIdRequirement
 }
+import com.digitalasset.ledger.service.TokenHolder
 
 object RunnerMain {
 
@@ -50,13 +51,13 @@ object RunnerMain {
       case None => sys.exit(1)
       case Some(config) => {
         val encodedDar: Dar[(PackageId, DamlLf.ArchivePayload)] =
-          DarReader().readArchiveFromFile(config.darPath).get
+          DarReader().readArchiveFromFile(config.darPath.toFile).get
         val dar: Dar[(PackageId, Package)] = encodedDar.map {
           case (pkgId, pkgArchive) => Decode.readArchivePayload(pkgId, pkgArchive)
         }
 
         if (config.listTriggers) {
-          listTriggers(config.darPath, dar)
+          listTriggers(config.darPath.toFile, dar)
           sys.exit(0)
         }
 
@@ -68,12 +69,19 @@ object RunnerMain {
         val sequencer = new AkkaExecutionSequencerPool("TriggerRunnerPool")(system)
         implicit val ec: ExecutionContext = system.dispatcher
 
+        val tokenHolder = config.accessTokenFile.map(new TokenHolder(_))
+        // We probably want to refresh the token at some point but given that triggers
+        // are expected to be written such that they can be killed and restarted at
+        // any time it would in principle also be fine to just have the auth failure due
+        // to an expired token tear the trigger down and have some external monitoring process (e.g. systemd)
+        // restart it.
         val applicationId = ApplicationId("Trigger Runner")
         val clientConfig = LedgerClientConfiguration(
           applicationId = ApplicationId.unwrap(applicationId),
           ledgerIdRequirement = LedgerIdRequirement("", enabled = false),
           commandClient = CommandClientConfiguration.default.copy(ttl = config.commandTtl),
-          sslContext = None
+          sslContext = None,
+          token = tokenHolder.flatMap(_.token)
         )
 
         val flow: Future[Unit] = for {

triggers/tests/BUILD.bazel

@@ -63,6 +63,7 @@ da_scala_binary(
         "//language-support/scala/bindings",
         "//language-support/scala/bindings-akka",
         "//ledger-api/rs-grpc-bridge",
+        "//ledger-service/utils",
         "//ledger/ledger-api-common",
         "//triggers/runner:trigger-runner-lib",
         "@maven//:com_github_scopt_scopt_2_12",
@@ -97,6 +98,43 @@ client_server_test(
     server_files = ["$(rootpath :acs.dar)"],
 )
 
+AUTH_TOKEN = "I_CAN_HAZ_AUTH"
+
+# This is a genrule so we can replace it by something nicer that actually generates the token
+# from some readable input so we can change it more easily.
+# For now, this corresponds to a token that has admin set to false
+# and actAs to Alice1, …, Alice100
+genrule(
+    name = "test-auth-token",
+    outs = ["test-auth-token.jwt"],
+    cmd = """
+      cat <<EOF > $(location test-auth-token.jwt)
+Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY3RBcyI6WyJBbGljZTEiLCJBbGljZTIiLCJBbGljZTMiLCJBbGljZTQiLCJBbGljZTUiLCJBbGljZTYiLCJBbGljZTciLCJBbGljZTgiLCJBbGljZTkiLCJBbGljZTEwIiwiQWxpY2UxMSIsIkFsaWNlMTIiLCJBbGljZTEzIiwiQWxpY2UxNCIsIkFsaWNlMTUiLCJBbGljZTE2IiwiQWxpY2UxNyIsIkFsaWNlMTgiLCJBbGljZTE5IiwiQWxpY2UyMCIsIkFsaWNlMjEiLCJBbGljZTIyIiwiQWxpY2UyMyIsIkFsaWNlMjQiLCJBbGljZTI1IiwiQWxpY2UyNiIsIkFsaWNlMjciLCJBbGljZTI4IiwiQWxpY2UyOSIsIkFsaWNlMzAiLCJBbGljZTMxIiwiQWxpY2UzMiIsIkFsaWNlMzMiLCJBbGljZTM0IiwiQWxpY2UzNSIsIkFsaWNlMzYiLCJBbGljZTM3IiwiQWxpY2UzOCIsIkFsaWNlMzkiLCJBbGljZTQwIiwiQWxpY2U0MSIsIkFsaWNlNDIiLCJBbGljZTQzIiwiQWxpY2U0NCIsIkFsaWNlNDUiLCJBbGljZTQ2IiwiQWxpY2U0NyIsIkFsaWNlNDgiLCJBbGljZTQ5IiwiQWxpY2U1MCIsIkFsaWNlNTEiLCJBbGljZTUyIiwiQWxpY2U1MyIsIkFsaWNlNTQiLCJBbGljZTU1IiwiQWxpY2U1NiIsIkFsaWNlNTciLCJBbGljZTU4IiwiQWxpY2U1OSIsIkFsaWNlNjAiLCJBbGljZTYxIiwiQWxpY2U2MiIsIkFsaWNlNjMiLCJBbGljZTY0IiwiQWxpY2U2NSIsIkFsaWNlNjYiLCJBbGljZTY3IiwiQWxpY2U2OCIsIkFsaWNlNjkiLCJBbGljZTcwIiwiQWxpY2U3MSIsIkFsaWNlNzIiLCJBbGljZTczIiwiQWxpY2U3NCIsIkFsaWNlNzUiLCJBbGljZTc2IiwiQWxpY2U3NyIsIkFsaWNlNzgiLCJBbGljZTc5IiwiQWxpY2U4MCIsIkFsaWNlODEiLCJBbGljZTgyIiwiQWxpY2U4MyIsIkFsaWNlODQiLCJBbGljZTg1IiwiQWxpY2U4NiIsIkFsaWNlODciLCJBbGljZTg4IiwiQWxpY2U4OSIsIkFsaWNlOTAiLCJBbGljZTkxIiwiQWxpY2U5MiIsIkFsaWNlOTMiLCJBbGljZTk0IiwiQWxpY2U5NSIsIkFsaWNlOTYiLCJBbGljZTk3IiwiQWxpY2U5OCIsIkFsaWNlOTkiLCJBbGljZTEwMCJdfQ.p78Bgrx0kX2tPwXoc2p5Uz22HifzfELjnmf7XwmCI4k
+EOF
+    """,
+)
+
+client_server_test(
+    name = "test_static_time_authenticated",
+    timeout = "long",
+    client = ":test_client",
+    client_args = ["--access-token-file"],
+    client_files = [
+        "$(rootpath :test-auth-token.jwt)",
+        "$(rootpath :acs.dar)",
+    ],
+    data = [
+        ":acs.dar",
+        ":test-auth-token.jwt",
+    ],
+    server = "//ledger/sandbox:sandbox-binary",
+    server_args = [
+        "--port=0",
+        "--auth-jwt-hs256-unsafe={}".format(AUTH_TOKEN),
+    ],
+    server_files = ["$(rootpath :acs.dar)"],
+)
+
 sh_test(
     name = "list-triggers",
     srcs = ["list-triggers.sh"],

triggers/tests/src/test/scala/com/digitalasset/daml/lf/engine/trigger/test/TestMain.scala

@@ -3,7 +3,7 @@
 
 package com.digitalasset.daml.lf.engine.trigger.test
 
-import java.io.File
+import java.nio.file.{Path, Paths}
 import java.time.Instant
 
 import akka.actor.ActorSystem
@@ -28,6 +28,7 @@ import com.digitalasset.ledger.api.v1.command_submission_service._
 import com.digitalasset.ledger.api.v1.commands._
 import com.digitalasset.ledger.api.v1.value
 import com.digitalasset.ledger.api.v1.transaction_filter.{Filters, TransactionFilter}
+import com.digitalasset.ledger.service.TokenHolder
 import com.digitalasset.daml.lf.archive.DarReader
 import com.digitalasset.daml.lf.archive.Dar
 import com.digitalasset.daml.lf.language.Ast._
@@ -49,7 +50,11 @@ import com.digitalasset.platform.services.time.TimeProviderType
 
 import com.digitalasset.daml.lf.engine.trigger.{Runner, TriggerMsg}
 
-case class Config(ledgerPort: Int, darPath: File, timeProviderType: TimeProviderType)
+case class Config(
+    ledgerPort: Int,
+    darPath: Path,
+    timeProviderType: TimeProviderType,
+    accessTokenFile: Option[Path])
 
 // We do not use scalatest here since that doesn’t work nicely with
 // the client_server_test macro.
@@ -78,12 +83,14 @@ class TestRunner(val config: Config) extends StrictLogging {
   var partyCount = 0
 
   val applicationId = ApplicationId("Trigger Test Runner")
+  val tokenHolder = config.accessTokenFile.map(new TokenHolder(_))
 
   val clientConfig = LedgerClientConfiguration(
     applicationId = applicationId.unwrap,
     ledgerIdRequirement = LedgerIdRequirement("", enabled = false),
     commandClient = CommandClientConfiguration.default,
-    sslContext = None
+    sslContext = None,
+    token = tokenHolder.flatMap(_.token)
   )
 
   def getNewParty(): String = {
@@ -917,15 +924,19 @@ object TestMain {
       .required()
       .action((p, c) => c.copy(ledgerPort = p))
 
-    arg[File]("<dar>")
+    arg[String]("<dar>")
       .required()
-      .action((d, c) => c.copy(darPath = d))
+      .action((d, c) => c.copy(darPath = Paths.get(d)))
 
     opt[Unit]('w', "wall-clock-time")
       .action { (t, c) =>
         c.copy(timeProviderType = TimeProviderType.WallClock)
       }
       .text("Use wall clock time (UTC). When not provided, static time is used.")
+    opt[String]("access-token-file")
+      .action { (f, c) =>
+        c.copy(accessTokenFile = Some(Paths.get(f)))
+      }
   }
 
   private val applicationId = ApplicationId("AscMain test")
@@ -936,12 +947,12 @@ object TestMain {
   case class FailedCompletions(num: Long)
 
   def main(args: Array[String]): Unit = {
-    configParser.parse(args, Config(0, null, TimeProviderType.Static)) match {
+    configParser.parse(args, Config(0, null, TimeProviderType.Static, None)) match {
       case None =>
         sys.exit(1)
       case Some(config) =>
         val encodedDar: Dar[(PackageId, DamlLf.ArchivePayload)] =
-          DarReader().readArchiveFromFile(config.darPath).get
+          DarReader().readArchiveFromFile(config.darPath.toFile).get
         val dar: Dar[(PackageId, Package)] = encodedDar.map {
           case (pkgId, pkgArchive) => Decode.readArchivePayload(pkgId, pkgArchive)
         }