[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
2024-09-20 01:07:18 +03:00 · 2021-10-20 15:28:21 +02:00 · 2021-10-20 15:28:21 +02:00 · f9e67adafc
commit f9e67adafc
parent 728296575b
19 changed files with 462 additions and 160 deletions
--- a/language-support/java/bindings-rxjava/BUILD.bazel
+++ b/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",
--- a/language-support/java/bindings-rxjava/src/test/scala/com/daml/ledger/rxjava/grpc/helpers/LedgerServices.scala
+++ b/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
--- a/language-support/java/bindings-rxjava/src/test/scala/com/daml/ledger/rxjava/package.scala
+++ b/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"
--- a/ledger/error/src/main/scala/com/daml/error/ErrorCodesVersionSwitcher.scala
+++ b/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
--- a/ledger/error/src/main/scala/com/daml/error/definitions/LedgerApiErrors.scala
+++ b/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"
          )
    }
  }
--- a/ledger/ledger-api-auth/BUILD.bazel
+++ b/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",
    ],
 )
--- a/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/Authorizer.scala
+++ b/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)
+    }
 }
--- a/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/interceptor/AuthorizationInterceptor.scala
+++ b/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/interceptor/AuthorizationInterceptor.scala
@ -3,17 +3,11 @@

 package com.daml.ledger.api.auth.interceptor

+import com.daml.error.{DamlContextualizedErrorLogger, ErrorCodesVersionSwitcher}
 import com.daml.ledger.api.auth.{AuthService, ClaimSet}
-import io.grpc.{
-  Context,
-  Contexts,
-  Metadata,
-  ServerCall,
-  ServerCallHandler,
-  ServerInterceptor,
-  Status,
-}
-import org.slf4j.{Logger, LoggerFactory}
+import com.daml.logging.{ContextualizedLogger, LoggingContext}
+import com.daml.platform.server.api.validation.ErrorFactories
+import io.grpc._

 import scala.compat.java8.FutureConverters
 import scala.concurrent.ExecutionContext
@ -22,12 +16,15 @@ import scala.util.{Failure, Success}
 /** This interceptor uses the given [[AuthService]] to get [[Claims]] for the current request,
  * and then stores them in the current [[Context]].
  */
-final class AuthorizationInterceptor(protected val authService: AuthService, ec: ExecutionContext)
+final class AuthorizationInterceptor(
+    protected val authService: AuthService,
+    ec: ExecutionContext,
+    errorCodesVersionSwitcher: ErrorCodesVersionSwitcher,
+)(implicit loggingContext: LoggingContext)
    extends ServerInterceptor {
-
-  private val logger: Logger = LoggerFactory.getLogger(AuthorizationInterceptor.getClass)
-  private val internalAuthenticationError =
-    Status.INTERNAL.withDescription("Failed to get claims from request metadata")
+  private val logger = ContextualizedLogger.get(getClass)
+  private val errorLogger = new DamlContextualizedErrorLogger(logger, loggingContext, None)
+  private val errorFactories = ErrorFactories(errorCodesVersionSwitcher)

  override def interceptCall[ReqT, RespT](
      call: ServerCall[ReqT, RespT],
@ -47,8 +44,11 @@ final class AuthorizationInterceptor(protected val authService: AuthService, ec:
        .toScala(authService.decodeMetadata(headers))
        .onComplete {
          case Failure(exception) =>
-            logger.warn(s"Failed to get claims from request metadata: ${exception.getMessage}")
-            call.close(internalAuthenticationError, new Metadata())
+            val error = errorFactories.internalAuthenticationError(
+              securitySafeMessage = "Failed to get claims from request metadata",
+              exception = exception,
+            )(errorLogger)
+            call.close(error.getStatus, error.getTrailers)
            new ServerCall.Listener[Nothing]() {}
          case Success(claimSet) =>
            val nextCtx = prevCtx.withValue(AuthorizationInterceptor.contextKeyClaimSet, claimSet)
@ -65,12 +65,17 @@ final class AuthorizationInterceptor(protected val authService: AuthService, ec:

 object AuthorizationInterceptor {

-  private val contextKeyClaimSet = Context.key[ClaimSet]("AuthServiceDecodedClaim")
+  private[auth] val contextKeyClaimSet = Context.key[ClaimSet]("AuthServiceDecodedClaim")

  def extractClaimSetFromContext(): Option[ClaimSet] =
    Option(contextKeyClaimSet.get())

-  def apply(authService: AuthService, ec: ExecutionContext): AuthorizationInterceptor =
-    new AuthorizationInterceptor(authService, ec)
-
+  def apply(
+      authService: AuthService,
+      ec: ExecutionContext,
+      errorCodesStatusSwitcher: ErrorCodesVersionSwitcher,
+  ): AuthorizationInterceptor =
+    LoggingContext.newLoggingContext { implicit loggingContext: LoggingContext =>
+      new AuthorizationInterceptor(authService, ec, errorCodesStatusSwitcher)
+    }
 }
--- a/ledger/ledger-api-auth/src/test/suite/scala/com/digitalasset/ledger/api/auth/AuthorizationInterceptorSpec.scala
+++ b/ledger/ledger-api-auth/src/test/suite/scala/com/digitalasset/ledger/api/auth/AuthorizationInterceptorSpec.scala
@ -0,0 +1,72 @@
+// Copyright (c) 2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.ledger.api.auth
+
+import com.daml.error.ErrorCodesVersionSwitcher
+import com.daml.ledger.api.auth.interceptor.AuthorizationInterceptor
+import com.google.rpc.ErrorInfo
+import io.grpc.{Metadata, ServerCall, Status}
+import org.mockito.captor.ArgCaptor
+import org.mockito.{ArgumentMatchersSugar, MockitoSugar}
+import org.scalatest.Assertion
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.util.concurrent.CompletableFuture
+import scala.concurrent.ExecutionContext.global
+import io.grpc.protobuf.StatusProto
+
+class AuthorizationInterceptorSpec
+    extends AnyFlatSpec
+    with MockitoSugar
+    with Matchers
+    with ArgumentMatchersSugar {
+  private val className = classOf[AuthorizationInterceptor].getSimpleName
+
+  behavior of s"$className.interceptCall"
+
+  it should "close the ServerCall with a V1 status code on decoding failure" in {
+    testServerCloseError(usesSelfServiceErrorCodes = false) { case (actualStatus, actualMetadata) =>
+      actualStatus.getCode shouldBe Status.Code.INTERNAL
+      actualStatus.getDescription shouldBe "Failed to get claims from request metadata"
+      actualMetadata.keys() shouldBe empty
+    }
+  }
+
+  it should "close the ServerCall with a V2 status code on decoding failure" in {
+    testServerCloseError(usesSelfServiceErrorCodes = true) { case (actualStatus, actualMetadata) =>
+      actualStatus.getCode shouldBe Status.Code.INTERNAL
+      actualStatus.getDescription shouldBe "An error occurred. Please contact the operator and inquire about the request <no-correlation-id>"
+
+      val actualRpcStatus = StatusProto.fromStatusAndTrailers(actualStatus, actualMetadata)
+      actualRpcStatus.getDetailsList.size() shouldBe 1
+      val errorInfo = actualRpcStatus.getDetailsList.get(0).unpack(classOf[ErrorInfo])
+      errorInfo.getReason shouldBe "INTERNAL_AUTHORIZATION_ERROR"
+    }
+  }
+
+  private def testServerCloseError(
+      usesSelfServiceErrorCodes: Boolean
+  )(assertRpcStatus: (Status, Metadata) => Assertion) = {
+    val authService = mock[AuthService]
+    val serverCall = mock[ServerCall[Nothing, Nothing]]
+    val failedMetadataDecode = CompletableFuture.supplyAsync[ClaimSet](() =>
+      throw new RuntimeException("some internal failure")
+    )
+
+    val errorCodesStatusSwitcher = new ErrorCodesVersionSwitcher(usesSelfServiceErrorCodes)
+    val authorizationInterceptor =
+      AuthorizationInterceptor(authService, global, errorCodesStatusSwitcher)
+
+    val statusCaptor = ArgCaptor[Status]
+    val metadataCaptor = ArgCaptor[Metadata]
+
+    when(authService.decodeMetadata(any[Metadata])).thenReturn(failedMetadataDecode)
+    authorizationInterceptor.interceptCall[Nothing, Nothing](serverCall, new Metadata(), null)
+
+    verify(serverCall, timeout(1000)).close(statusCaptor.capture, metadataCaptor.capture)
+
+    assertRpcStatus(statusCaptor.value, metadataCaptor.value)
+  }
+}
--- a/ledger/ledger-api-auth/src/test/suite/scala/com/digitalasset/ledger/api/auth/AuthorizerSpec.scala
+++ b/ledger/ledger-api-auth/src/test/suite/scala/com/digitalasset/ledger/api/auth/AuthorizerSpec.scala
@ -0,0 +1,101 @@
+// Copyright (c) 2021 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.ledger.api.auth
+
+import com.daml.error.ErrorCodesVersionSwitcher
+import com.daml.ledger.api.auth.interceptor.AuthorizationInterceptor
+import io.grpc.{Status, StatusRuntimeException}
+import org.scalatest.Assertion
+import org.scalatest.flatspec.AsyncFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.time.Instant
+import scala.concurrent.Future
+import scala.util.{Failure, Success, Try}
+
+class AuthorizerSpec extends AsyncFlatSpec with Matchers {
+  private val className = classOf[Authorizer].getSimpleName
+  private val dummyRequest = 1337L
+  private val expectedSuccessfulResponse = "expectedSuccessfulResponse"
+  private val dummyReqRes: Long => Future[String] =
+    Map(dummyRequest -> Future.successful(expectedSuccessfulResponse))
+  private val allAuthorized: ClaimSet.Claims => Either[AuthorizationError, Unit] = _ => Right(())
+  private val unauthorized: ClaimSet.Claims => Either[AuthorizationError, Unit] = _ =>
+    Left(AuthorizationError.MissingAdminClaim)
+
+  behavior of s"$className.authorize"
+
+  it should "authorize if claims are valid" in {
+    contextWithClaims {
+      authorizer(selfServiceErrorCodes = false)
+        .authorize(dummyReqRes)(allAuthorized)(dummyRequest)
+    }.map(_ shouldBe expectedSuccessfulResponse)
+  }
+
+  behavior of s"$className.authorize (V1 error codes)"
+
+  it should "return unauthenticated if missing claims" in {
+    testUnauthenticated(selfServiceErrorCodes = false)
+  }
+
+  it should "return permission denied on authorization error" in {
+    testPermissionDenied(selfServiceErrorCodes = false)
+  }
+
+  behavior of s"$className.authorize (V2 error codes)"
+
+  it should "return unauthenticated if missing claims" in {
+    testUnauthenticated(selfServiceErrorCodes = true)
+  }
+
+  it should "return permission denied on authorization error" in {
+    testPermissionDenied(selfServiceErrorCodes = true)
+  }
+
+  private def testPermissionDenied(selfServiceErrorCodes: Boolean) =
+    contextWithClaims {
+      authorizer(selfServiceErrorCodes).authorize(dummyReqRes)(unauthorized)(dummyRequest)
+    }
+      .transform(
+        assertExpectedFailure(selfServiceErrorCodes = selfServiceErrorCodes)(
+          Status.PERMISSION_DENIED.getCode
+        )
+      )
+
+  private def testUnauthenticated(selfServiceErrorCodes: Boolean) =
+    contextWithoutClaims {
+      authorizer(selfServiceErrorCodes).authorize(dummyReqRes)(allAuthorized)(dummyRequest)
+    }
+      .transform(
+        assertExpectedFailure(selfServiceErrorCodes = selfServiceErrorCodes)(
+          Status.UNAUTHENTICATED.getCode
+        )
+      )
+
+  private def assertExpectedFailure[T](
+      selfServiceErrorCodes: Boolean
+  )(expectedStatusCode: Status.Code): Try[T] => Try[Assertion] = {
+    case Failure(ex: StatusRuntimeException) =>
+      ex.getStatus.getCode shouldBe expectedStatusCode
+      if (selfServiceErrorCodes) {
+        ex.getStatus.getDescription shouldBe "An error occurred. Please contact the operator and inquire about the request <no-correlation-id>"
+      }
+      Success(succeed)
+    case ex => fail(s"Expected a failure with StatusRuntimeException but got $ex")
+  }
+
+  private def contextWithoutClaims[R](f: => R): R = io.grpc.Context.ROOT.call(() => f)
+
+  private def contextWithClaims[R](f: => R): R =
+    io.grpc.Context.ROOT
+      .withValue(AuthorizationInterceptor.contextKeyClaimSet, ClaimSet.Claims.Wildcard)
+      .call(() => f)
+
+  private def authorizer(selfServiceErrorCodes: Boolean) = Authorizer(
+    () => Instant.ofEpochSecond(1337L),
+    "some-ledger-id",
+    "participant-id",
+    new ErrorCodesVersionSwitcher(selfServiceErrorCodes),
+  )
+}
--- a/ledger/ledger-api-common/src/main/scala/com/digitalasset/platform/server/api/validation/ErrorFactories.scala
+++ b/ledger/ledger-api-common/src/main/scala/com/digitalasset/platform/server/api/validation/ErrorFactories.scala
@ -3,6 +3,7 @@

 package com.daml.platform.server.api.validation

+import com.daml.error.ErrorCode.ApiException
 import com.daml.error.definitions.LedgerApiErrors
 import com.daml.error.{ContextualizedErrorLogger, ErrorCodesVersionSwitcher}
 import com.daml.ledger.api.domain.LedgerId
@ -12,12 +13,12 @@ import com.daml.platform.server.api.validation.ErrorFactories.{
  addDefiniteAnswerDetails,
  definiteAnswers,
 }
-import com.daml.platform.server.api.{ApiException, ValidationLogger}
+import com.daml.platform.server.api.{ValidationLogger, ApiException => NoStackTraceApiException}
 import com.google.protobuf.{Any => AnyProto}
 import com.google.rpc.{ErrorInfo, Status}
 import io.grpc.Status.Code
-import io.grpc.StatusRuntimeException
 import io.grpc.protobuf.StatusProto
+import io.grpc.{Metadata, StatusRuntimeException}
 import scalaz.syntax.tag._

 class ErrorFactories private (errorCodesVersionSwitcher: ErrorCodesVersionSwitcher) {
@ -26,7 +27,7 @@ class ErrorFactories private (errorCodesVersionSwitcher: ErrorCodesVersionSwitch
      contextualizedErrorLogger: ContextualizedErrorLogger,
      logger: ContextualizedLogger,
      loggingContext: LoggingContext,
-  ): StatusRuntimeException = {
+  ): StatusRuntimeException =
    errorCodesVersionSwitcher.choose(
      v1 = ValidationLogger.logFailureWithContext(
        request,
@ -40,16 +41,14 @@ class ErrorFactories private (errorCodesVersionSwitcher: ErrorCodesVersionSwitch
        )
        .asGrpcError,
    )
-  }

  def packageNotFound(packageId: String)(implicit
      contextualizedErrorLogger: ContextualizedErrorLogger
-  ): StatusRuntimeException = {
+  ): StatusRuntimeException =
    errorCodesVersionSwitcher.choose(
      v1 = io.grpc.Status.NOT_FOUND.asRuntimeException(),
      v2 = LedgerApiErrors.ReadErrors.PackageNotFound.Reject(packageId = packageId).asGrpcError,
    )
-  }

  def duplicateCommandException(implicit
      contextualizedErrorLogger: ContextualizedErrorLogger
@ -197,30 +196,49 @@ class ErrorFactories private (errorCodesVersionSwitcher: ErrorCodesVersionSwitch
  }

  // permission denied is intentionally without description to ensure we don't leak security relevant information by accident
-  def permissionDenied()(implicit
+  def permissionDenied(cause: String)(implicit
      contextualizedErrorLogger: ContextualizedErrorLogger
  ): StatusRuntimeException = errorCodesVersionSwitcher.choose(
-    v1 = grpcError(
-      Status
-        .newBuilder()
-        .setCode(Code.PERMISSION_DENIED.value())
-        .build()
-    ),
-    v2 = LedgerApiErrors.AuthorizationChecks.PermissionDenied.Reject().asGrpcError,
+    v1 = {
+      contextualizedErrorLogger.warn(s"Permission denied. Reason: $cause.")
+      new ApiException(
+        io.grpc.Status.PERMISSION_DENIED,
+        new Metadata(),
+      )
+    },
+    v2 = LedgerApiErrors.AuthorizationChecks.PermissionDenied.Reject(cause).asGrpcError,
  )

-  def unauthenticated()(implicit
+  def unauthenticatedMissingJwtToken()(implicit
      contextualizedErrorLogger: ContextualizedErrorLogger
  ): StatusRuntimeException = errorCodesVersionSwitcher.choose(
-    v1 = grpcError(
-      Status
-        .newBuilder()
-        .setCode(Code.UNAUTHENTICATED.value())
-        .build()
+    v1 = new ApiException(
+      io.grpc.Status.UNAUTHENTICATED,
+      new Metadata(),
    ),
-    v2 = LedgerApiErrors.AuthorizationChecks.Unauthenticated.Reject().asGrpcError,
+    v2 = LedgerApiErrors.AuthorizationChecks.Unauthenticated
+      .MissingJwtToken()
+      .asGrpcError,
  )

+  def internalAuthenticationError(securitySafeMessage: String, exception: Throwable)(implicit
+      contextualizedErrorLogger: ContextualizedErrorLogger
+  ): StatusRuntimeException =
+    errorCodesVersionSwitcher.choose(
+      v1 = {
+        contextualizedErrorLogger.warn(
+          s"$securitySafeMessage: ${exception.getMessage}"
+        )
+        new ApiException(
+          io.grpc.Status.INTERNAL.withDescription(securitySafeMessage),
+          new Metadata(),
+        )
+      },
+      v2 = LedgerApiErrors.AuthorizationChecks.InternalAuthorizationError
+        .Reject(securitySafeMessage, exception)
+        .asGrpcError,
+    )
+
  /** @param definiteAnswer A flag that says whether it is a definite answer. Provided only in the context of command deduplication.
    * @return An exception with the [[Code.UNAVAILABLE]] status code.
    */
@ -298,7 +316,7 @@ class ErrorFactories private (errorCodesVersionSwitcher: ErrorCodesVersionSwitch
    * @param status A Protobuf [[Status]] object.
    * @return An exception without a stack trace.
    */
-  def grpcError(status: Status): StatusRuntimeException = new ApiException(
+  def grpcError(status: Status): StatusRuntimeException = new NoStackTraceApiException(
    StatusProto.toStatusRuntimeException(status)
  )
 }
--- a/ledger/ledger-api-common/src/test/suite/scala/com/digitalasset/platform/server/api/validation/ErrorFactoriesSpec.scala
+++ b/ledger/ledger-api-common/src/test/suite/scala/com/digitalasset/platform/server/api/validation/ErrorFactoriesSpec.scala
@ -3,23 +3,20 @@

 package com.daml

-import com.daml.error.{
-  ContextualizedErrorLogger,
-  DamlContextualizedErrorLogger,
-  ErrorCodesVersionSwitcher,
-}
-import com.daml.ledger.api.domain.LedgerId
-import com.daml.logging.{ContextualizedLogger, LoggingContext}
-import com.daml.platform.server.api.validation.ErrorFactories
-import com.daml.platform.server.api.validation.ErrorFactories._
-import com.google.rpc.{ErrorInfo, RequestInfo, ResourceInfo, RetryInfo, Status}
+import error.{ContextualizedErrorLogger, DamlContextualizedErrorLogger, ErrorCodesVersionSwitcher}
+import ledger.api.domain.LedgerId
+import logging.{ContextualizedLogger, LoggingContext}
+import platform.server.api.validation.ErrorFactories
+import platform.server.api.validation.ErrorFactories._
+
+import com.google.protobuf
+import com.google.rpc._
 import io.grpc.Status.Code
 import io.grpc.StatusRuntimeException
 import io.grpc.protobuf.StatusProto
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.prop.TableDrivenPropertyChecks
 import org.scalatest.wordspec.AnyWordSpec
-import com.google.protobuf

 import scala.jdk.CollectionConverters._

@ -87,7 +84,7 @@ class ErrorFactoriesSpec extends AnyWordSpec with Matchers with TableDrivenPrope
    }

    "return a permissionDenied error" in {
-      assertVersionedError(_.permissionDenied())(
+      assertVersionedError(_.permissionDenied("some cause"))(
        v1_code = Code.PERMISSION_DENIED,
        v1_message = "",
        v1_details = Seq.empty,
@ -101,6 +98,38 @@ class ErrorFactoriesSpec extends AnyWordSpec with Matchers with TableDrivenPrope
      )
    }

+    "return an unauthenticatedMissingJwtToken error" in {
+      assertVersionedError(_.unauthenticatedMissingJwtToken())(
+        v1_code = Code.UNAUTHENTICATED,
+        v1_message = "",
+        v1_details = Seq.empty,
+        v2_code = Code.UNAUTHENTICATED,
+        v2_message =
+          s"An error occurred. Please contact the operator and inquire about the request $correlationId",
+        v2_details = Seq[ErrorDetails.ErrorDetail](
+          ErrorDetails.ErrorInfoDetail("UNAUTHENTICATED"),
+          DefaultTraceIdRequestInfo,
+        ),
+      )
+    }
+
+    "return an internalAuthenticationError" in {
+      val someSecuritySafeMessage = "nothing security sensitive in here"
+      val someThrowable = new RuntimeException("some internal authentication error")
+      assertVersionedError(_.internalAuthenticationError(someSecuritySafeMessage, someThrowable))(
+        v1_code = Code.INTERNAL,
+        v1_message = someSecuritySafeMessage,
+        v1_details = Seq.empty,
+        v2_code = Code.INTERNAL,
+        v2_message =
+          s"An error occurred. Please contact the operator and inquire about the request $correlationId",
+        v2_details = Seq[ErrorDetails.ErrorDetail](
+          ErrorDetails.ErrorInfoDetail("INTERNAL_AUTHORIZATION_ERROR"),
+          DefaultTraceIdRequestInfo,
+        ),
+      )
+    }
+
    "return a missingLedgerConfig error" in {
      val testCases = Table(
        ("definite answer", "expected details"),
@ -165,21 +194,6 @@ class ErrorFactoriesSpec extends AnyWordSpec with Matchers with TableDrivenPrope
      }
    }

-    "return an unauthenticated error" in {
-      assertVersionedError(_.unauthenticated())(
-        v1_code = Code.UNAUTHENTICATED,
-        v1_message = "",
-        v1_details = Seq.empty,
-        v2_code = Code.UNAUTHENTICATED,
-        v2_message =
-          s"An error occurred. Please contact the operator and inquire about the request $correlationId",
-        v2_details = Seq[ErrorDetails.ErrorDetail](
-          ErrorDetails.ErrorInfoDetail("UNAUTHENTICATED"),
-          DefaultTraceIdRequestInfo,
-        ),
-      )
-    }
-
    "return a ledgerIdMismatch error" in {
      val testCases = Table(
        ("definite answer", "expected details"),
--- a/ledger/participant-integration-api/src/main/scala/platform/apiserver/StandaloneApiServer.scala
+++ b/ledger/participant-integration-api/src/main/scala/platform/apiserver/StandaloneApiServer.scala
@ -3,13 +3,11 @@

 package com.daml.platform.apiserver

-import java.io.File
-import java.time.{Clock, Instant}
-
 import akka.actor.ActorSystem
 import akka.stream.Materializer
 import com.daml.api.util.TimeProvider
 import com.daml.buildinfo.BuildInfo
+import com.daml.error.ErrorCodesVersionSwitcher
 import com.daml.ledger.api.auth.interceptor.AuthorizationInterceptor
 import com.daml.ledger.api.auth.{AuthService, Authorizer}
 import com.daml.ledger.api.domain
@ -35,6 +33,8 @@ import com.daml.ports.{Port, PortFiles}
 import io.grpc.{BindableService, ServerInterceptor}
 import scalaz.{-\/, \/-}

+import java.io.File
+import java.time.{Clock, Instant}
 import scala.collection.immutable
 import scala.concurrent.ExecutionContextExecutor
 import scala.util.{Failure, Success, Try}
@ -92,7 +92,15 @@ final class StandaloneApiServer(
          enableInMemoryFanOutForLedgerApi = config.enableInMemoryFanOutForLedgerApi,
        )
        .map(index => new SpannedIndexService(new TimedIndexService(index, metrics)))
-      authorizer = new Authorizer(Clock.systemUTC.instant _, ledgerId, participantId)
+      errorCodesVersionSwitcher = new ErrorCodesVersionSwitcher(
+        config.enableSelfServiceErrorCodes
+      )
+      authorizer = new Authorizer(
+        Clock.systemUTC.instant _,
+        ledgerId,
+        participantId,
+        errorCodesVersionSwitcher,
+      )
      healthChecksWithIndexService = healthChecks + ("index" -> indexService)
      executionSequencerFactory <- new ExecutionSequencerFactoryOwner()
      apiServicesOwner = new ApiServices.Owner(
@ -126,7 +134,11 @@ final class StandaloneApiServer(
        config.maxInboundMessageSize,
        config.address,
        config.tlsConfig,
-        AuthorizationInterceptor(authService, executionContext) :: otherInterceptors,
+        AuthorizationInterceptor(
+          authService,
+          executionContext,
+          errorCodesVersionSwitcher,
+        ) :: otherInterceptors,
        servicesExecutionContext,
        metrics,
      )
--- a/ledger/participant-integration-api/src/main/scala/platform/apiserver/services/ApiPackageService.scala
+++ b/ledger/participant-integration-api/src/main/scala/platform/apiserver/services/ApiPackageService.scala
@ -114,7 +114,6 @@ private[apiserver] final class ApiPackageService private (
      loggingContext: LoggingContext
  ): DamlContextualizedErrorLogger =
    new DamlContextualizedErrorLogger(logger, loggingContext, None)
-
 }

 private[platform] object ApiPackageService {
--- a/ledger/participant-integration-api/src/main/scala/platform/apiserver/services/ApiSubmissionService.scala
+++ b/ledger/participant-integration-api/src/main/scala/platform/apiserver/services/ApiSubmissionService.scala
@ -4,13 +4,13 @@
 package com.daml.platform.apiserver.services

 import com.daml.api.util.TimeProvider
-import com.daml.error.definitions.{ErrorCauseExport, RejectionGenerators}
 import com.daml.error.{
  ContextualizedErrorLogger,
  DamlContextualizedErrorLogger,
  ErrorCause,
  ErrorCodesVersionSwitcher,
 }
+import com.daml.error.definitions.{ErrorCauseExport, RejectionGenerators}
 import com.daml.ledger.api.domain.{LedgerId, Commands => ApiCommands}
 import com.daml.ledger.api.messages.command.submission.SubmitRequest
 import com.daml.ledger.api.{DeduplicationPeriod, SubmissionIdGenerator}
--- a/ledger/sandbox-classic/BUILD.bazel
+++ b/ledger/sandbox-classic/BUILD.bazel
@ -62,6 +62,7 @@ alias(
            "//language-support/scala/bindings",
            "//ledger-api/rs-grpc-bridge",
            "//ledger/caching",
+            "//ledger/error",
            "//ledger/ledger-api-auth",
            "//ledger/ledger-api-common",
            "//ledger/ledger-api-domain",
--- a/ledger/sandbox-classic/src/main/scala/platform/sandbox/SandboxServer.scala
+++ b/ledger/sandbox-classic/src/main/scala/platform/sandbox/SandboxServer.scala
@ -3,17 +3,13 @@

 package com.daml.platform.sandbox

-import java.io.File
-import java.nio.file.Files
-import java.time.Instant
-import java.util.concurrent.Executors
-
 import akka.actor.ActorSystem
 import akka.stream.Materializer
 import com.codahale.metrics.MetricRegistry
 import com.daml.api.util.TimeProvider
 import com.daml.buildinfo.BuildInfo
 import com.daml.dec.DirectExecutionContext
+import com.daml.error.ErrorCodesVersionSwitcher
 import com.daml.ledger.api.auth.interceptor.AuthorizationInterceptor
 import com.daml.ledger.api.auth.{AuthService, AuthServiceWildcard, Authorizer}
 import com.daml.ledger.api.domain.LedgerId
@ -49,6 +45,10 @@ import com.daml.platform.store.{FlywayMigrations, LfValueTranslationCache}
 import com.daml.ports.Port
 import scalaz.syntax.tag._

+import java.io.File
+import java.nio.file.Files
+import java.time.Instant
+import java.util.concurrent.Executors
 import scala.concurrent.duration.DurationInt
 import scala.concurrent.{Await, ExecutionContext, Future}
 import scala.jdk.CollectionConverters._
@ -349,10 +349,14 @@ final class SandboxServer(
          )
      }).acquire()
      ledgerId <- Resource.fromFuture(indexAndWriteService.indexService.getLedgerId())
+      errorCodesVersionSwitcher = new ErrorCodesVersionSwitcher(
+        config.enableSelfServiceErrorCodes
+      )
      authorizer = new Authorizer(
        () => java.time.Clock.systemUTC.instant(),
        LedgerId.unwrap(ledgerId),
        config.participantId,
+        errorCodesVersionSwitcher,
      )
      healthChecks = new HealthChecks(
        "index" -> indexAndWriteService.indexService,
@ -397,7 +401,11 @@ final class SandboxServer(
        config.address,
        config.tlsConfig,
        List(
-          AuthorizationInterceptor(authService, executionContext),
+          AuthorizationInterceptor(
+            authService,
+            executionContext,
+            errorCodesVersionSwitcher,
+          ),
          resetService,
        ),
        servicesExecutionContext,
--- a/ledger/sandbox/BUILD.bazel
+++ b/ledger/sandbox/BUILD.bazel
@ -46,6 +46,7 @@ alias(
            "//daml-lf/transaction",
            "//language-support/scala/bindings",
            "//ledger/caching",
+            "//ledger/error",
            "//ledger/ledger-api-auth",
            "//ledger/ledger-api-common",
            "//ledger/ledger-api-domain",
--- a/ledger/sandbox/src/main/scala/platform/sandboxnext/Runner.scala
+++ b/ledger/sandbox/src/main/scala/platform/sandboxnext/Runner.scala
@ -7,12 +7,12 @@ import java.io.File
 import java.time.{Clock, Instant}
 import java.util.UUID
 import java.util.concurrent.Executors
-
 import akka.actor.ActorSystem
 import akka.stream.Materializer
 import com.daml.api.util.TimeProvider
 import com.daml.buildinfo.BuildInfo
 import com.daml.caching
+import com.daml.error.ErrorCodesVersionSwitcher
 import com.daml.ledger.api.auth.{AuthServiceWildcard, Authorizer}
 import com.daml.ledger.api.domain
 import com.daml.ledger.api.health.HealthChecks
@ -223,7 +223,12 @@ class Runner(config: SandboxConfig) extends ResourceOwner[Port] {
              resetService = {
                val clock = Clock.systemUTC()
                val authorizer =
-                  new Authorizer(() => clock.instant(), ledgerId, config.participantId)
+                  new Authorizer(
+                    () => clock.instant(),
+                    ledgerId,
+                    config.participantId,
+                    new ErrorCodesVersionSwitcher(config.enableSelfServiceErrorCodes),
+                  )
                new SandboxResetService(
                  domain.LedgerId(ledgerId),
                  () => {