Authorization rule changes for IDPs isolation [DPP-1336] (digital-ass…

…et#15848)
kgeonw · Jan 12, 2023 · f184892 · f184892
1 parent 092c8b1
commit f184892
Show file tree

Hide file tree

Showing 25 changed files with 642 additions and 79 deletions.
diff --git a/UNRELEASED.md b/UNRELEASED.md
@@ -45,3 +45,26 @@ The following index db metrics were added:
 The following index db metrics were removed:
 - daml_index_db_get_flat_transactions*
 - daml_index_db_get_transaction_trees*
+
+### Changes to the metrics 
+
+The following metrics were added:
+- daml.identity_provider_config_store.*
+- daml_index_db_flat_transactions_stream_translation*
+- daml_index_db_flat_transactions_stream_fetch_event_consuming_ids_stakeholder*
+- daml_index_db_flat_transactions_stream_fetch_event_consuming_payloads*
+- daml_index_db_flat_transactions_stream_fetch_event_create_ids_stakeholder*
+- daml_index_db_flat_transactions_stream_fetch_event_create_payloads*
+- daml_index_db_tree_transactions_stream_translation*
+- daml_index_db_tree_transactions_stream_fetch_event_consuming_ids_non_stakeholder*
+- daml_index_db_tree_transactions_stream_fetch_event_consuming_ids_stakeholder*
+- daml_index_db_tree_transactions_stream_fetch_event_consuming_payloads*
+- daml_index_db_tree_transactions_stream_fetch_event_create_ids_non_stakeholder*
+- daml_index_db_tree_transactions_stream_fetch_event_create_ids_stakeholder*
+- daml_index_db_tree_transactions_stream_fetch_event_create_payloads*
+- daml_index_db_tree_transactions_stream_fetch_event_non_consuming_ids_informee*
+- daml_index_db_tree_transactions_stream_fetch_event_non_consuming_payloads*
+
+The following metrics were removed:
+- daml_index_db_get_flat_transactions*
+- daml_index_db_get_transaction_trees*
diff --git a/language-support/java/bindings-rxjava/BUILD.bazel b/language-support/java/bindings-rxjava/BUILD.bazel
@@ -78,6 +78,7 @@ da_scala_library(
         "//ledger-api/rs-grpc-bridge",
         "//ledger/ledger-api-auth",
         "//ledger/ledger-api-common",
+        "//ledger/ledger-api-domain",
         "//ledger/participant-local-store",
         "//libs-scala/contextualized-logging",
         "@maven//:com_google_protobuf_protobuf_java",

diff --git a/ledger-api/grpc-definitions/com/daml/ledger/api/v1/admin/party_management_service.proto b/ledger-api/grpc-definitions/com/daml/ledger/api/v1/admin/party_management_service.proto
@@ -188,7 +188,7 @@ message PartyDetails {
   // Optional
   string display_name = 2;
 
-  // true if party is hosted by the participant.
+  // true if party is hosted by the participant and the party shares the same identity provider as the user issuing the request.
   // Optional
   bool is_local = 3;
 
@@ -198,6 +198,9 @@ message PartyDetails {
   ObjectMeta local_metadata = 4;
 
   // The id of the ``Identity Provider``
-  // Optional, if not set, assume the party is managed by the default identity provider or party is not hosted by the participant.
+  // Optional, if not set, there could be 3 options:
+  // 1) the party is managed by the default identity provider.
+  // 2) party is not hosted by the participant.
+  // 3) party is hosted by the participant, but is outside of the user's identity provider.
   string identity_provider_id = 5;
 }
diff --git a/ledger/ledger-api-auth/BUILD.bazel b/ledger/ledger-api-auth/BUILD.bazel
@@ -16,6 +16,7 @@ da_scala_library(
         "@maven//:io_spray_spray_json",
         "@maven//:org_scalaz_scalaz_core",
         "@maven//:com_typesafe_akka_akka_actor",
+        "@maven//:com_typesafe_scala_logging_scala_logging",
     ],
     tags = ["maven_coordinates=com.daml:ledger-api-auth:__VERSION__"],
     visibility = [
@@ -25,14 +26,19 @@ da_scala_library(
     deps = [
         "//daml-lf/data",
         "//ledger-api/grpc-definitions:ledger_api_proto_scala",
+        "//ledger/caching",
         "//ledger/error",
         "//ledger/ledger-api-common",
         "//ledger/ledger-api-domain",
         "//ledger/ledger-api-errors",
+        "//ledger/metrics",
         "//ledger/participant-local-store",
         "//libs-scala/contextualized-logging",
         "//libs-scala/jwt",
+        "//observability/metrics",
         "@maven//:com_auth0_java_jwt",
+        "@maven//:com_auth0_jwks_rsa",
+        "@maven//:com_github_ben_manes_caffeine_caffeine",
         "@maven//:io_grpc_grpc_api",
         "@maven//:io_grpc_grpc_context",
         "@maven//:org_slf4j_slf4j_api",

diff --git a/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/AuthServiceJWT.scala b/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/AuthServiceJWT.scala
@@ -7,6 +7,7 @@ import com.daml.jwt.{Error, JwtFromBearerHeader, JwtVerifier, JwtVerifierBase}
 
 import java.util.concurrent.{CompletableFuture, CompletionStage}
 import com.daml.lf.data.Ref
+import com.daml.ledger.api.domain.IdentityProviderId
 import io.grpc.Metadata
 import org.slf4j.{Logger, LoggerFactory}
 import spray.json._
@@ -82,10 +83,12 @@ class AuthServiceJWT(verifier: JwtVerifierBase) extends AuthService {
         applicationId = payload.applicationId,
         expiration = payload.exp,
         resolvedFromUser = false,
+        identityProviderId = IdentityProviderId.Default,
       )
 
     case payload: StandardJWTPayload =>
       ClaimSet.AuthenticatedUser(
+        identityProviderId = IdentityProviderId.Default,
         participantId = payload.participantId,
         userId = payload.userId,
         expiration = payload.exp,

diff --git a/.../ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/AuthorizationError.scala b/.../ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/AuthorizationError.scala
@@ -3,6 +3,8 @@
 
 package com.daml.ledger.api.auth
 
+import com.daml.ledger.api.domain.IdentityProviderId
+
 import java.time.Instant
 
 sealed abstract class AuthorizationError {
@@ -53,4 +55,13 @@ object AuthorizationError {
   final case class MissingActClaim(party: String) extends AuthorizationError {
     override val reason = s"Claims do not authorize to act as party '$party'"
   }
+
+  final case class InvalidIdentityProviderId(identityProviderId: IdentityProviderId)
+      extends AuthorizationError {
+    private val id = identityProviderId.toRequestString
+    override val reason =
+      s"identity_provider_id from the request `$id` does not match the one provided in the authorization claims"
+  }
+
+  final case class InvalidField(fieldName: String, reason: String) extends AuthorizationError
 }
diff --git 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,13 +3,12 @@
 
 package com.daml.ledger.api.auth
 
-import java.time.Instant
-
 import akka.actor.Scheduler
 import com.daml.error.definitions.LedgerApiErrors
 import com.daml.error.{ContextualizedErrorLogger, DamlContextualizedErrorLogger}
 import com.daml.jwt.JwtTimestampLeeway
 import com.daml.ledger.api.auth.interceptor.AuthorizationInterceptor
+import com.daml.ledger.api.domain.IdentityProviderId
 import com.daml.ledger.api.v1.transaction_filter.TransactionFilter
 import com.daml.ledger.api.validation.ValidationErrors
 import com.daml.logging.{ContextualizedLogger, LoggingContext}
@@ -18,6 +17,7 @@ import io.grpc.StatusRuntimeException
 import io.grpc.stub.{ServerCallStreamObserver, StreamObserver}
 import scalapb.lenses.Lens
 
+import java.time.Instant
 import scala.concurrent.{ExecutionContext, Future}
 import scala.util.{Failure, Success, Try}
 
@@ -82,6 +82,62 @@ final class Authorizer(
       }
     }
 
+  def requireIdpAdminClaimsAndMatchingRequestIdpId[Req, Res](
+      identityProviderId: String,
+      call: Req => Future[Res],
+  ): Req => Future[Res] =
+    authorize(call) { claims =>
+      for {
+        _ <- valid(claims)
+        _ <- claims.isAdminOrIDPAdmin
+        requestIdentityProviderId <- requireIdentityProviderId(identityProviderId)
+        _ <- validateRequestIdentityProviderId(requestIdentityProviderId, claims)
+      } yield ()
+    }
+
+  def requireMatchingRequestIdpId[Req, Res](
+      identityProviderId: String,
+      call: Req => Future[Res],
+  ): Req => Future[Res] =
+    authorize(call) { claims =>
+      for {
+        _ <- valid(claims)
+        requestIdentityProviderId <- requireIdentityProviderId(identityProviderId)
+        _ <- validateRequestIdentityProviderId(requestIdentityProviderId, claims)
+      } yield ()
+    }
+
+  def requireIdpAdminClaims[Req, Res](
+      call: Req => Future[Res]
+  ): Req => Future[Res] =
+    authorize(call) { claims =>
+      for {
+        _ <- valid(claims)
+        _ <- claims.isAdminOrIDPAdmin
+      } yield {
+        ()
+      }
+    }
+
+  private def requireIdentityProviderId(
+      identityProviderId: String
+  ): Either[AuthorizationError, IdentityProviderId] =
+    IdentityProviderId.fromString(identityProviderId).left.map { reason =>
+      AuthorizationError.InvalidField("identity_provider_id", reason)
+    }
+
+  private def validateRequestIdentityProviderId(
+      requestIdentityProviderId: IdentityProviderId,
+      claims: ClaimSet.Claims,
+  ): Either[AuthorizationError, Unit] = claims.identityProviderId match {
+    case id: IdentityProviderId.Id if requestIdentityProviderId != id =>
+      // Claim is valid only for the specific Identity Provider,
+      // and identity_provider_id in the request matches the one provided in the claim.
+      Left(AuthorizationError.InvalidIdentityProviderId(requestIdentityProviderId))
+    case _ =>
+      Right(())
+  }
+
   private[this] def requireForAll[T](
       xs: IterableOnce[T],
       f: T => Either[AuthorizationError, Unit],
@@ -216,8 +272,18 @@ final class Authorizer(
       errOrV: Either[AuthorizationError, T]
   ): Either[StatusRuntimeException, T] =
     errOrV.fold(
-      err =>
-        Left(LedgerApiErrors.AuthorizationChecks.PermissionDenied.Reject(err.reason).asGrpcError),
+      {
+        case AuthorizationError.InvalidField(fieldName, reason) =>
+          Left(
+            LedgerApiErrors.RequestValidation.InvalidField
+              .Reject(fieldName = fieldName, message = reason)
+              .asGrpcError
+          )
+        case err =>
+          Left(
+            LedgerApiErrors.AuthorizationChecks.PermissionDenied.Reject(err.reason).asGrpcError
+          )
+      },
       Right(_),
     )
 

diff --git a/...er-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/CachedJwtVerifierLoader.scala b/...er-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/CachedJwtVerifierLoader.scala
@@ -0,0 +1,99 @@
+// Copyright (c) 2023 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.auth0.jwk.UrlJwkProvider
+import com.daml.caching.CaffeineCache
+import com.daml.caching.CaffeineCache.FutureAsyncCacheLoader
+import com.daml.jwt.{JwtTimestampLeeway, JwtVerifier, RSA256Verifier}
+import com.daml.ledger.api.domain.JwksUrl
+import com.daml.metrics.Metrics
+import com.github.benmanes.caffeine.{cache => caffeine}
+import scalaz.\/
+import com.daml.jwt.{Error => JwtError}
+import com.daml.ledger.api.auth.CachedJwtVerifierLoader.CacheKey
+
+import java.security.interfaces.RSAPublicKey
+import java.util.concurrent.TimeUnit
+import scala.concurrent.{ExecutionContext, Future}
+
+/** A JWK verifier loader, where the public keys are automatically fetched from the given JWKS URL.
+  * The keys are then transformed into JWK Verifier
+  *
+  * The verifiers are kept in cache, in order to prevent having to do a remote network access for each token validation.
+  *
+  * The cache is limited both in size and time.
+  * A size limit protects against infinitely growing memory consumption.
+  * A time limit is a safety catch for the case where a public key is used to sign a token without an expiration time
+  * and then is revoked.
+  *
+  * @param cacheMaxSize Maximum number of public keys to keep in the cache.
+  * @param cacheExpirationTime Maximum time to keep public keys in the cache.
+  * @param connectionTimeout Timeout for connecting to the JWKS URL.
+  * @param readTimeout Timeout for reading from the JWKS URL.
+  */
+class CachedJwtVerifierLoader(
+    // Large enough such that malicious users can't cycle through all keys from reasonably sized JWKS,
+    // forcing cache eviction and thus introducing additional latency.
+    cacheMaxSize: Long = 1000,
+    cacheExpirationTime: Long = 10,
+    cacheExpirationUnit: TimeUnit = TimeUnit.HOURS,
+    connectionTimeout: Long = 10,
+    connectionTimeoutUnit: TimeUnit = TimeUnit.SECONDS,
+    readTimeout: Long = 10,
+    readTimeoutUnit: TimeUnit = TimeUnit.SECONDS,
+    jwtTimestampLeeway: Option[JwtTimestampLeeway] = None,
+    metrics: Metrics,
+)(implicit
+    executionContext: ExecutionContext
+) extends JwtVerifierLoader {
+
+  private val cache: CaffeineCache.AsyncLoadingCaffeineCache[
+    CacheKey,
+    JwtVerifier,
+  ] =
+    new CaffeineCache.AsyncLoadingCaffeineCache(
+      caffeine.Caffeine
+        .newBuilder()
+        .expireAfterWrite(cacheExpirationTime, cacheExpirationUnit)
+        .maximumSize(cacheMaxSize)
+        .buildAsync(
+          new FutureAsyncCacheLoader[CacheKey, JwtVerifier](key => getVerifier(key))
+        ),
+      metrics.daml.identityProviderConfigStore.verifierCache,
+    )
+
+  override def loadJwtVerifier(jwksUrl: JwksUrl, keyId: Option[String]): Future[JwtVerifier] =
+    cache.get(CacheKey(jwksUrl, keyId))
+
+  private def jwkProvider(jwksUrl: JwksUrl) =
+    new UrlJwkProvider(
+      jwksUrl.toURL,
+      Integer.valueOf(
+        connectionTimeoutUnit.toMillis(connectionTimeout).toInt
+      ),
+      Integer.valueOf(readTimeoutUnit.toMillis(readTimeout).toInt),
+    )
+
+  private def getVerifier(
+      key: CacheKey
+  ): Future[JwtVerifier] =
+    for {
+      jwk <- Future(jwkProvider(key.jwksUrl).get(key.keyId.orNull))
+      publicKey = jwk.getPublicKey.asInstanceOf[RSAPublicKey]
+      verifier <- fromDisjunction(RSA256Verifier(publicKey, jwtTimestampLeeway))
+    } yield verifier
+
+  private def fromDisjunction[T](e: \/[JwtError, T]): Future[T] =
+    e.fold(err => Future.failed(new Exception(err.message)), Future.successful)
+
+}
+
+object CachedJwtVerifierLoader {
+
+  case class CacheKey(
+      jwksUrl: JwksUrl,
+      keyId: Option[String],
+  )
+}
diff --git a/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/Claims.scala b/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/Claims.scala
@@ -4,6 +4,7 @@
 package com.daml.ledger.api.auth
 
 import com.daml.jwt.JwtTimestampLeeway
+import com.daml.ledger.api.domain.IdentityProviderId
 import com.daml.lf.data.Ref
 
 import java.time.{Duration, Instant}
@@ -71,13 +72,15 @@ object ClaimSet {
     * @param applicationId  If set, the claims will only be valid on the given application identifier.
     * @param expiration     If set, the claims will cease to be valid at the given time.
     * @param resolvedFromUser  If set, then the claims were resolved from a user in the user management service.
+    * @param identityProviderId  If set, the claims will only be valid on the given Identity Provider configuration.
     */
   final case class Claims(
       claims: Seq[Claim],
       ledgerId: Option[String],
       participantId: Option[String],
       applicationId: Option[String],
       expiration: Option[Instant],
+      identityProviderId: IdentityProviderId,
       resolvedFromUser: Boolean,
   ) extends ClaimSet {
     def validForLedger(id: String): Either[AuthorizationError, Unit] =
@@ -118,6 +121,16 @@ object ClaimSet {
     def isAdmin: Either[AuthorizationError, Unit] =
       Either.cond(claims.contains(ClaimAdmin), (), AuthorizationError.MissingAdminClaim)
 
+    /** Returns true if the set of claims authorizes the user as an administrator or
+      * an identity provider administrator, unless the claims expired
+      */
+    def isAdminOrIDPAdmin: Either[AuthorizationError, Unit] =
+      Either.cond(
+        claims.contains(ClaimIdentityProviderAdmin) || claims.contains(ClaimAdmin),
+        (),
+        AuthorizationError.MissingAdminClaim,
+      )
+
     /** Returns true if the set of claims authorizes the user to use public services, unless the claims expired */
     def isPublic: Either[AuthorizationError, Unit] =
       Either.cond(claims.contains(ClaimPublic), (), AuthorizationError.MissingPublicClaim)
@@ -152,6 +165,7 @@ object ClaimSet {
 
   /** The representation of a user that was authenticated, but whose [[Claims]] have not yet been resolved. */
   final case class AuthenticatedUser(
+      identityProviderId: IdentityProviderId,
       userId: String,
       participantId: Option[String],
       expiration: Option[Instant],
@@ -167,6 +181,7 @@ object ClaimSet {
       applicationId = None,
       expiration = None,
       resolvedFromUser = false,
+      identityProviderId = IdentityProviderId.Default,
     )
 
     /** A set of [[Claims]] that has all possible authorizations */