[Self-service error codes] Adapt error responses in ledger-api-auth […

…DPP-617] (#11223)

* [Self-service error codes] Implement V2 in Authorizer

CHANGELOG_BEGIN
CHANGELOG_END

* Added unit test for authorize (non-streamed)

* Fix after rebase

* Do not expose the error codes switching mechanism to the Java bindings

* Adjust InternalAuthorizationError to be SystemInternalAssumptionViolated

* Parameter names in test

* Testing AuthorizationInterceptor with regard to returned error codes

* Do not use default error code version switchers at instance creation

* Addressed Pawel's review comments

* Using ErrorFactories for error dispatching

* Pass loggingContext to Authorizer where available

* Generic internal authorization error

language-support/java/bindings-rxjava/BUILD.bazel

@@ -76,6 +76,7 @@ da_scala_library(
         "//language-support/java/bindings:bindings-java",
         "//ledger-api/grpc-definitions:ledger_api_proto_scala",
         "//ledger-api/rs-grpc-bridge",
+        "//ledger/error",
         "//ledger/ledger-api-auth",
         "//ledger/ledger-api-common",
         "@maven//:com_google_protobuf_protobuf_java",

language-support/java/bindings-rxjava/src/test/scala/com/daml/ledger/rxjava/grpc/helpers/LedgerServices.scala

@@ -3,6 +3,8 @@
 
 package com.daml.ledger.rxjava.grpc.helpers
 
+import com.daml.error.ErrorCodesVersionSwitcher
+
 import java.net.{InetSocketAddress, SocketAddress}
 import java.time.{Clock, Duration}
 import java.util.concurrent.TimeUnit
@@ -30,10 +32,10 @@ import com.daml.ledger.api.v1.package_service.{
 }
 import com.daml.ledger.api.v1.testing.time_service.GetTimeResponse
 import com.google.protobuf.empty.Empty
-
 import io.grpc._
 import io.grpc.netty.NettyServerBuilder
 import io.reactivex.Observable
+
 import scala.concurrent.ExecutionContext.global
 import scala.concurrent.{ExecutionContext, Future}
 
@@ -45,7 +47,12 @@ final class LedgerServices(val ledgerId: String) {
   private val esf: ExecutionSequencerFactory = new SingleThreadExecutionSequencerPool(ledgerId)
   private val participantId = "LedgerServicesParticipant"
   private val authorizer =
-    new Authorizer(() => Clock.systemUTC().instant(), ledgerId, participantId)
+    Authorizer(
+      () => Clock.systemUTC().instant(),
+      ledgerId,
+      participantId,
+      new ErrorCodesVersionSwitcher(enableSelfServiceErrorCodes = true),
+    )
 
   def newServerBuilder(): NettyServerBuilder = NettyServerBuilder.forAddress(nextAddress())
 
@@ -83,12 +90,18 @@ final class LedgerServices(val ledgerId: String) {
   private def createServer(
       authService: AuthService,
       services: Seq[ServerServiceDefinition],
-  ): Server =
+  ): Server = {
+    val authorizationInterceptor = AuthorizationInterceptor(
+      authService,
+      executionContext,
+      new ErrorCodesVersionSwitcher(enableSelfServiceErrorCodes = true),
+    )
     services
       .foldLeft(newServerBuilder())(_ addService _)
-      .intercept(AuthorizationInterceptor(authService, executionContext))
+      .intercept(authorizationInterceptor)
       .build()
       .start()
+  }
 
   private def createChannel(port: Int): ManagedChannel =
     ManagedChannelBuilder

language-support/java/bindings-rxjava/src/test/scala/com/daml/ledger/rxjava/package.scala

@@ -3,9 +3,10 @@
 
 package com.daml.ledger
 
+import com.daml.error.ErrorCodesVersionSwitcher
+
 import java.time.Clock
 import java.util.UUID
-
 import com.daml.lf.data.Ref
 import com.daml.ledger.api.auth.{
   AuthServiceStatic,
@@ -24,7 +25,12 @@ package object rxjava {
     throw new UnsupportedOperationException("Untested endpoint, implement if needed")
 
   private[rxjava] val authorizer =
-    new Authorizer(() => Clock.systemUTC().instant(), "testLedgerId", "testParticipantId")
+    Authorizer(
+      () => Clock.systemUTC().instant(),
+      "testLedgerId",
+      "testParticipantId",
+      new ErrorCodesVersionSwitcher(enableSelfServiceErrorCodes = true),
+    )
 
   private[rxjava] val emptyToken = "empty"
   private[rxjava] val publicToken = "public"

ledger/error/src/main/scala/com/daml/error/ErrorCodesVersionSwitcher.scala

@@ -2,7 +2,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
 package com.daml.error
-
 import io.grpc.StatusRuntimeException
 
 import scala.concurrent.Future

ledger/error/src/main/scala/com/daml/error/definitions/LedgerApiErrors.scala

@@ -118,13 +118,28 @@ object LedgerApiErrors extends LedgerApiErrorGroup {
           id = "UNAUTHENTICATED",
           ErrorCategory.AuthInterceptorInvalidAuthenticationCredentials,
         ) {
-      case class Reject()(implicit
+      case class MissingJwtToken()(implicit
           loggingContext: ContextualizedErrorLogger
       ) extends LoggingTransactionErrorImpl(
             cause = "The command is missing a JWT token"
           )
     }
 
+    @Explanation("An internal system authorization error occurred.")
+    @Resolution("Contact the participant operator.")
+    object InternalAuthorizationError
+        extends ErrorCode(
+          id = "INTERNAL_AUTHORIZATION_ERROR",
+          ErrorCategory.SystemInternalAssumptionViolated,
+        ) {
+      case class Reject(message: String, throwable: Throwable)(implicit
+          loggingContext: ContextualizedErrorLogger
+      ) extends LoggingTransactionErrorImpl(
+            cause = message,
+            throwableO = Some(throwable),
+          )
+    }
+
     @Explanation(
       """This rejection is given if the supplied JWT token is not sufficient for the intended command.
         |The exact reason is logged on the participant, but not given to the user for security reasons."""
@@ -134,10 +149,11 @@ object LedgerApiErrors extends LedgerApiErrorGroup {
     )
     object PermissionDenied
         extends ErrorCode(id = "PERMISSION_DENIED", ErrorCategory.InsufficientPermission) {
-      case class Reject()(implicit
+      case class Reject(override val cause: String)(implicit
           loggingContext: ContextualizedErrorLogger
       ) extends LoggingTransactionErrorImpl(
-            cause = "The provided JWT token is not sufficient to authorize the intended command"
+            cause =
+              s"The provided JWT token is not sufficient to authorize the intended command: $cause"
           )
     }
   }

ledger/ledger-api-auth/BUILD.bazel

@@ -29,6 +29,7 @@ da_scala_library(
         "//ledger-service/jwt",
         "//ledger/error",
         "//ledger/ledger-api-common",
+        "//libs-scala/contextualized-logging",
         "@maven//:com_auth0_java_jwt",
         "@maven//:io_grpc_grpc_api",
         "@maven//:io_grpc_grpc_context",
@@ -55,6 +56,7 @@ da_scala_test_suite(
     srcs = glob(["src/test/suite/**/*.scala"]),
     scala_deps = [
         "@maven//:io_spray_spray_json",
+        "@maven//:org_mockito_mockito_scala",
         "@maven//:org_scalacheck_scalacheck",
         "@maven//:org_scalatest_scalatest_core",
         "@maven//:org_scalatest_scalatest_matchers_core",
@@ -64,6 +66,14 @@ da_scala_test_suite(
     ],
     deps = [
         ":ledger-api-auth",
+        "//ledger/error",
+        "//ledger/test-common",
+        "@maven//:com_google_api_grpc_proto_google_common_protos",
+        "@maven//:com_google_protobuf_protobuf_java",
+        "@maven//:io_grpc_grpc_api",
+        "@maven//:io_grpc_grpc_context",
+        "@maven//:io_grpc_grpc_protobuf",
+        "@maven//:org_mockito_mockito_core",
         "@maven//:org_scalatest_scalatest_compatible",
     ],
 )

ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/Authorizer.scala

@@ -3,12 +3,16 @@
 
 package com.daml.ledger.api.auth
 
-import com.daml.error.{ContextualizedErrorLogger, NoLogging}
+import com.daml.error.{
+  ContextualizedErrorLogger,
+  DamlContextualizedErrorLogger,
+  ErrorCodesVersionSwitcher,
+}
 import com.daml.ledger.api.auth.interceptor.AuthorizationInterceptor
 import com.daml.ledger.api.v1.transaction_filter.TransactionFilter
-import com.daml.platform.server.api.validation.ErrorFactories.{permissionDenied, unauthenticated}
+import com.daml.logging.{ContextualizedLogger, LoggingContext}
+import com.daml.platform.server.api.validation.ErrorFactories
 import io.grpc.stub.{ServerCallStreamObserver, StreamObserver}
-import org.slf4j.LoggerFactory
 
 import java.time.Instant
 import scala.collection.compat._
@@ -18,11 +22,16 @@ import scala.util.{Failure, Success, Try}
 /** A simple helper that allows services to use authorization claims
   * that have been stored by [[AuthorizationInterceptor]].
   */
-final class Authorizer(now: () => Instant, ledgerId: String, participantId: String) {
-
-  private val logger = LoggerFactory.getLogger(this.getClass)
-  // TODO error codes: Enable logging
-  private implicit val contextualizedErrorLogger: ContextualizedErrorLogger = NoLogging
+final class Authorizer(
+    now: () => Instant,
+    ledgerId: String,
+    participantId: String,
+    errorCodesVersionSwitcher: ErrorCodesVersionSwitcher,
+)(implicit loggingContext: LoggingContext) {
+  private val logger = ContextualizedLogger.get(this.getClass)
+  private val errorFactories = ErrorFactories(errorCodesVersionSwitcher)
+  private implicit val errorLogger: ContextualizedErrorLogger =
+    new DamlContextualizedErrorLogger(logger, loggingContext, None)
 
   /** Validates all properties of claims that do not depend on the request,
     * such as expiration time or ledger ID.
@@ -174,74 +183,86 @@ final class Authorizer(now: () => Instant, ledgerId: String, participantId: Stri
   private def ongoingAuthorization[Res](
       scso: ServerCallStreamObserver[Res],
       claims: ClaimSet.Claims,
-  ) =
-    new OngoingAuthorizationObserver[Res](
-      scso,
-      claims,
-      _.notExpired(now()),
-      authorizationError => {
-        logger.warn(s"Permission denied. Reason: ${authorizationError.reason}.")
-        permissionDenied()
-      },
-    )
+  ) = new OngoingAuthorizationObserver[Res](
+    scso,
+    claims,
+    _.notExpired(now()),
+    authorizationError => {
+      errorFactories.permissionDenied(authorizationError.reason)
+    },
+  )
 
   private def authenticatedClaimsFromContext(): Try[ClaimSet.Claims] =
     AuthorizationInterceptor
       .extractClaimSetFromContext()
-      .fold[Try[ClaimSet.Claims]](Failure(unauthenticated())) {
-        case ClaimSet.Unauthenticated => Failure(unauthenticated())
+      .fold[Try[ClaimSet.Claims]](Failure(errorFactories.unauthenticatedMissingJwtToken())) {
+        case ClaimSet.Unauthenticated =>
+          Failure(errorFactories.unauthenticatedMissingJwtToken())
         case claims: ClaimSet.Claims => Success(claims)
       }
 
   private def authorize[Req, Res](call: (Req, ServerCallStreamObserver[Res]) => Unit)(
       authorized: ClaimSet.Claims => Either[AuthorizationError, Unit]
-  ): (Req, StreamObserver[Res]) => Unit =
-    (request, observer) => {
-      val scso = assertServerCall(observer)
-      authenticatedClaimsFromContext()
-        .fold(
-          ex => {
-            logger.debug(
-              s"No authenticated claims found in the request context. Returning UNAUTHENTICATED"
-            )
-            observer.onError(ex)
+  ): (Req, StreamObserver[Res]) => Unit = (request, observer) => {
+    val scso = assertServerCall(observer)
+    authenticatedClaimsFromContext()
+      .fold(
+        ex => {
+          // TODO error codes: Remove once fully relying on self-service error codes with logging on creation
+          logger.debug(
+            s"No authenticated claims found in the request context. Returning UNAUTHENTICATED"
+          )
+          observer.onError(ex)
+        },
+        claims =>
+          authorized(claims) match {
+            case Right(_) =>
+              call(
+                request,
+                if (claims.expiration.isDefined)
+                  ongoingAuthorization(scso, claims)
+                else
+                  scso,
+              )
+            case Left(authorizationError) =>
+              observer.onError(
+                errorFactories.permissionDenied(authorizationError.reason)
+              )
           },
-          claims =>
-            authorized(claims) match {
-              case Right(_) =>
-                call(
-                  request,
-                  if (claims.expiration.isDefined)
-                    ongoingAuthorization(scso, claims)
-                  else
-                    scso,
-                )
-              case Left(authorizationError) =>
-                logger.warn(s"Permission denied. Reason: ${authorizationError.reason}.")
-                observer.onError(permissionDenied())
-            },
-        )
-    }
+      )
+  }
 
-  private def authorize[Req, Res](call: Req => Future[Res])(
+  private[auth] def authorize[Req, Res](call: Req => Future[Res])(
       authorized: ClaimSet.Claims => Either[AuthorizationError, Unit]
-  ): Req => Future[Res] =
-    request =>
-      authenticatedClaimsFromContext()
-        .fold(
-          ex => {
-            logger.debug(
-              s"No authenticated claims found in the request context. Returning UNAUTHENTICATED"
-            )
-            Future.failed(ex)
+  ): Req => Future[Res] = request =>
+    authenticatedClaimsFromContext()
+      .fold(
+        ex => {
+          // TODO error codes: Remove once fully relying on self-service error codes with logging on creation
+          logger.debug(
+            s"No authenticated claims found in the request context. Returning UNAUTHENTICATED"
+          )
+          Future.failed(ex)
+        },
+        claims =>
+          authorized(claims) match {
+            case Right(_) => call(request)
+            case Left(authorizationError) =>
+              Future.failed(
+                errorFactories.permissionDenied(authorizationError.reason)
+              )
           },
-          claims =>
-            authorized(claims) match {
-              case Right(_) => call(request)
-              case Left(authorizationError) =>
-                logger.warn(s"Permission denied. Reason: ${authorizationError.reason}.")
-                Future.failed(permissionDenied())
-            },
-        )
+      )
+}
 
+object Authorizer {
+  def apply(
+      now: () => Instant,
+      ledgerId: String,
+      participantId: String,
+      errorCodesVersionSwitcher: ErrorCodesVersionSwitcher,
+  ): Authorizer =
+    LoggingContext.newLoggingContext { loggingContext =>
+      new Authorizer(now, ledgerId, participantId, errorCodesVersionSwitcher)(loggingContext)
+    }
 }