Restrict the use of foreign IDP parties within user management API [DPP-1383] (#16025)

2024-11-10 10:46:11 +03:00 · 2023-01-19 14:10:30 +01:00 · 2023-01-19 14:10:30 +01:00 · b546f6f8b5
commit b546f6f8b5
parent 10fdd77fe8
22 changed files with 430 additions and 96 deletions
--- 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
@ -85,11 +85,18 @@ final class Authorizer(
  def requireIdpAdminClaimsAndMatchingRequestIdpId[Req, Res](
      identityProviderId: String,
      call: Req => Future[Res],
+  ): Req => Future[Res] =
+    requireIdpAdminClaimsAndMatchingRequestIdpId(identityProviderId, false, call)
+
+  def requireIdpAdminClaimsAndMatchingRequestIdpId[Req, Res](
+      identityProviderId: String,
+      mustBeParticipantAdmin: Boolean,
+      call: Req => Future[Res],
  ): Req => Future[Res] =
    authorize(call) { claims =>
      for {
        _ <- valid(claims)
-        _ <- claims.isAdminOrIDPAdmin
+        _ <- if (mustBeParticipantAdmin) claims.isAdmin else claims.isAdminOrIDPAdmin
        requestIdentityProviderId <- requireIdentityProviderId(identityProviderId)
        _ <- validateRequestIdentityProviderId(requestIdentityProviderId, claims)
      } yield ()
--- a/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/services/UserManagementServiceAuthorization.scala
+++ b/ledger/ledger-api-auth/src/main/scala/com/digitalasset/ledger/api/auth/services/UserManagementServiceAuthorization.scala
@ -28,11 +28,16 @@ private[daml] final class UserManagementServiceAuthorization(
  private implicit val errorLogger: ContextualizedErrorLogger =
    new DamlContextualizedErrorLogger(logger, loggingContext, None)

+  // Only ParticipantAdmin is allowed to grant ParticipantAdmin right
+  private def containsParticipantAdmin(rights: Seq[Right]): Boolean =
+    rights.contains(Right(Right.Kind.ParticipantAdmin(Right.ParticipantAdmin())))
+
  override def createUser(request: CreateUserRequest): Future[CreateUserResponse] =
    request.user match {
      case Some(user) =>
        authorizer.requireIdpAdminClaimsAndMatchingRequestIdpId(
          user.identityProviderId,
+          containsParticipantAdmin(request.rights),
          service.createUser,
        )(
          request
@ -72,6 +77,7 @@ private[daml] final class UserManagementServiceAuthorization(
  override def grantUserRights(request: GrantUserRightsRequest): Future[GrantUserRightsResponse] =
    authorizer.requireIdpAdminClaimsAndMatchingRequestIdpId(
      request.identityProviderId,
+      containsParticipantAdmin(request.rights),
      service.grantUserRights,
    )(
      request
@ -82,6 +88,7 @@ private[daml] final class UserManagementServiceAuthorization(
  ): Future[RevokeUserRightsResponse] =
    authorizer.requireIdpAdminClaimsAndMatchingRequestIdpId(
      request.identityProviderId,
+      containsParticipantAdmin(request.rights),
      service.revokeUserRights,
    )(
      request
--- a/ledger/ledger-api-tests/infrastructure/src/main/scala/com/daml/ledger/api/testtool/infrastructure/participant/ParticipantTestContexts.scala
+++ b/ledger/ledger-api-tests/infrastructure/src/main/scala/com/daml/ledger/api/testtool/infrastructure/participant/ParticipantTestContexts.scala
@ -140,6 +140,7 @@ trait ParticipantTestContext extends UserManagementTestContext {
      partyIdHint: Option[String] = None,
      displayName: Option[String] = None,
      localMetadata: Option[ObjectMeta] = None,
+      identityProviderId: Option[String] = None,
  ): Future[Primitive.Party]

  def allocateParty(req: AllocatePartyRequest): Future[AllocatePartyResponse]
--- a/ledger/ledger-api-tests/infrastructure/src/main/scala/com/daml/ledger/api/testtool/infrastructure/participant/SingleParticipantTestContext.scala
+++ b/ledger/ledger-api-tests/infrastructure/src/main/scala/com/daml/ledger/api/testtool/infrastructure/participant/SingleParticipantTestContext.scala
@ -232,6 +232,7 @@ final class SingleParticipantTestContext private[participant] (
      partyIdHint: Option[String] = None,
      displayName: Option[String] = None,
      localMetadata: Option[ObjectMeta] = None,
+      identityProviderId: Option[String] = None,
  ): Future[Party] =
    services.partyManagement
      .allocateParty(
@ -239,6 +240,7 @@ final class SingleParticipantTestContext private[participant] (
          partyIdHint = partyIdHint.getOrElse(""),
          displayName = displayName.getOrElse(""),
          localMetadata = localMetadata,
+          identityProviderId = identityProviderId.getOrElse(""),
        )
      )
      .map(r => Party(r.partyDetails.get.party))
--- a/ledger/ledger-api-tests/infrastructure/src/main/scala/com/daml/ledger/api/testtool/infrastructure/participant/TimeoutParticipantTestContext.scala
+++ b/ledger/ledger-api-tests/infrastructure/src/main/scala/com/daml/ledger/api/testtool/infrastructure/participant/TimeoutParticipantTestContext.scala
@ -143,9 +143,10 @@ class TimeoutParticipantTestContext(timeoutScaleFactor: Double, delegate: Partic
      partyIdHint: Option[String] = None,
      displayName: Option[String] = None,
      localMetadata: Option[ObjectMeta] = None,
+      identityProviderId: Option[String] = None,
  ): Future[Primitive.Party] = withTimeout(
    s"Allocate party with hint $partyIdHint and display name $displayName",
-    delegate.allocateParty(partyIdHint, displayName, localMetadata),
+    delegate.allocateParty(partyIdHint, displayName, localMetadata, identityProviderId),
  )

  override def getParties(req: GetPartiesRequest): Future[GetPartiesResponse] = withTimeout(
--- a/ledger/ledger-api-tests/infrastructure/src/main/scala/com/daml/ledger/api/testtool/infrastructure/participant/UserManagementTestContext.scala
+++ b/ledger/ledger-api-tests/infrastructure/src/main/scala/com/daml/ledger/api/testtool/infrastructure/participant/UserManagementTestContext.scala
@ -4,10 +4,14 @@
 package com.daml.ledger.api.testtool.infrastructure.participant

 import java.util.concurrent.ConcurrentHashMap
-
 import com.daml.error.definitions.LedgerApiErrors
 import com.daml.error.utils.ErrorDetails
 import com.daml.ledger.api.testtool.infrastructure.LedgerServices
+import com.daml.ledger.api.v1.admin.identity_provider_config_service.{
+  CreateIdentityProviderConfigRequest,
+  CreateIdentityProviderConfigResponse,
+  IdentityProviderConfig,
+}
 import com.daml.ledger.api.v1.admin.user_management_service.UserManagementServiceGrpc.UserManagementService
 import com.daml.ledger.api.v1.admin.user_management_service.{
  CreateUserRequest,
@ -17,6 +21,7 @@ import com.daml.ledger.api.v1.admin.user_management_service.{
  User,
 }

+import java.util.UUID
 import scala.concurrent.{ExecutionContext, Future}

 trait UserManagementTestContext {
@ -77,4 +82,21 @@ trait UserManagementTestContext {
    Future.sequence(deletions).map(_ => ())
  }

+  def createIdentityProviderConfig(
+      identityProviderId: String
+  ): Future[CreateIdentityProviderConfigResponse] = {
+    services.identityProviderConfig.createIdentityProviderConfig(
+      CreateIdentityProviderConfigRequest(
+        Some(
+          IdentityProviderConfig(
+            identityProviderId = identityProviderId,
+            isDeactivated = false,
+            issuer = UUID.randomUUID().toString,
+            jwksUrl = "http://daml.com/jwks.json",
+          )
+        )
+      )
+    )
+  }
+
 }
--- a/ledger/metrics/src/main/scala/com/daml/metrics/PartyRecordStoreMetrics.scala
+++ b/ledger/metrics/src/main/scala/com/daml/metrics/PartyRecordStoreMetrics.scala
@ -16,6 +16,7 @@ class PartyRecordStoreMetrics(
 ) extends DatabaseMetricsFactory(prefix, factory) {

  val getPartyRecord: DatabaseMetrics = createDbMetrics("get_party_record")
+  val partiesExist: DatabaseMetrics = createDbMetrics("parties_exist")
  val createPartyRecord: DatabaseMetrics = createDbMetrics("create_party_record")
  val updatePartyRecord: DatabaseMetrics = createDbMetrics("update_party_record")

--- a/ledger/participant-integration-api/src/main/scala/platform/apiserver/ApiServices.scala
+++ b/ledger/participant-integration-api/src/main/scala/platform/apiserver/ApiServices.scala
@ -209,6 +209,7 @@ private[daml] object ApiServices {
              maxUsersPageSize = userManagementConfig.maxUsersPageSize,
              submissionIdGenerator = SubmissionIdGenerator.Random,
              identityProviderExists = new IdentityProviderExists(identityProviderConfigStore),
+              partyRecordExist = new PartyRecordsExist(partyRecordStore),
            )
          val identityProvider =
            new ApiIdentityProviderConfigService(identityProviderConfigStore)
--- a/ledger/participant-integration-api/src/main/scala/platform/apiserver/services/admin/ApiUserManagementService.scala
+++ b/ledger/participant-integration-api/src/main/scala/platform/apiserver/services/admin/ApiUserManagementService.scala
@ -5,6 +5,8 @@ package com.daml.platform.apiserver.services.admin

 import com.daml.error.definitions.LedgerApiErrors
 import com.daml.error.{ContextualizedErrorLogger, DamlContextualizedErrorLogger}
+import com.daml.ledger.api.SubmissionIdGenerator
+import com.daml.ledger.api.auth.ClaimAdmin
 import com.daml.ledger.api.auth.ClaimSet.Claims
 import com.daml.ledger.api.auth.interceptor.AuthorizationInterceptor
 import com.daml.ledger.api.domain._
@ -15,7 +17,6 @@ import com.daml.ledger.api.v1.admin.user_management_service.{
  UpdateUserResponse,
 }
 import com.daml.ledger.api.v1.admin.{user_management_service => proto}
-import com.daml.ledger.api.SubmissionIdGenerator
 import com.daml.lf.data.Ref
 import com.daml.logging.LoggingContext.withEnrichedLoggingContext
 import com.daml.logging.{ContextualizedLogger, LoggingContext}
@ -39,6 +40,7 @@ import scala.util.Try
 private[apiserver] final class ApiUserManagementService(
    userManagementStore: UserManagementStore,
    identityProviderExists: IdentityProviderExists,
+    partyRecordExist: PartyRecordsExist,
    maxUsersPageSize: Int,
    submissionIdGenerator: SubmissionIdGenerator,
 )(implicit
@ -62,6 +64,9 @@ private[apiserver] final class ApiUserManagementService(

  override def createUser(request: proto.CreateUserRequest): Future[CreateUserResponse] =
    withSubmissionId { implicit loggingContext =>
+      // Retrieving the authenticated user context from the thread-local context
+      val authorizedUserContextF: Future[AuthenticatedUserContext] =
+        resolveAuthenticatedUserContext()
      withValidation {
        for {
          pUser <- requirePresence(request.user, "user")
@ -100,7 +105,13 @@ private[apiserver] final class ApiUserManagementService(
      } { case (user, pRights) =>
        for {
          _ <- identityProviderExistsOrError(user.identityProviderId)
-          _ <- verifyUserIsNonExistentOrInIdp(user.identityProviderId, user.id)
+          authorizedUserContext <- authorizedUserContextF
+          _ <- verifyPartyExistInIdp(
+            pRights,
+            user.identityProviderId,
+            authorizedUserContext.isParticipantAdmin,
+          )
+          _ <- verifyUserDoesNotExist(user.identityProviderId, user.id, "creating user")
          result <- userManagementStore
            .createUser(
              user = user,
@ -113,8 +124,8 @@ private[apiserver] final class ApiUserManagementService(

  override def updateUser(request: UpdateUserRequest): Future[UpdateUserResponse] = {
    withSubmissionId { implicit loggingContext =>
-      // Retrieving the authenticated user from the context
-      val authorizedUserIdFO: Future[Option[String]] = resolveAuthenticatedUser()
+      val authorizedUserContextF: Future[AuthenticatedUserContext] =
+        resolveAuthenticatedUserContext()
      withValidation {
        for {
          pUser <- requirePresence(request.user, "user")
@ -153,11 +164,16 @@ private[apiserver] final class ApiUserManagementService(
        for {
          userUpdate <- handleUpdatePathResult(user.id, UserUpdateMapper.toUpdate(user, fieldMask))
          _ <- identityProviderExistsOrError(user.identityProviderId)
-          _ <- verifyUserIsNonExistentOrInIdp(user.identityProviderId, user.id)
-          authorizedUserIdO <- authorizedUserIdFO
+          authorizedUserContext <- authorizedUserContextF
+          _ <- verifyPartyExistInIdp(
+            Set(),
+            user.identityProviderId,
+            authorizedUserContext.isParticipantAdmin,
+          )
+          _ <- verifyUserExistsAndInIdp(user.identityProviderId, user.id, "updating user")
          _ <-
            if (
-              authorizedUserIdO
+              authorizedUserContext.userId
                .contains(userUpdate.id) && userUpdate.isDeactivatedUpdateO.contains(true)
            ) {
              Future.failed(
@ -181,7 +197,7 @@ private[apiserver] final class ApiUserManagementService(
    }
  }

-  private def resolveAuthenticatedUser(): Future[Option[String]] = {
+  private def resolveAuthenticatedUserContext(): Future[AuthenticatedUserContext] = {
    AuthorizationInterceptor
      .extractClaimSetFromContext()
      .fold(
@ -192,9 +208,8 @@ private[apiserver] final class ApiUserManagementService(
              .asGrpcError
          ),
        fb = {
-          case claims: Claims if claims.resolvedFromUser =>
-            Future.successful(claims.applicationId)
-          case claims: Claims if !claims.resolvedFromUser => Future.successful(None)
+          case claims: Claims =>
+            Future.successful(AuthenticatedUserContext(claims))
          case claimsSet =>
            Future.failed(
              LedgerApiErrors.InternalError
@ -217,7 +232,7 @@ private[apiserver] final class ApiUserManagementService(
        )
      } yield (userId, identityProviderId)
    } { case (userId, identityProviderId) =>
-      verifyUserIsNonExistentOrInIdp(identityProviderId, userId)
+      verifyUserExistsAndInIdp(identityProviderId, userId, "getting user")
        .flatMap { _ => userManagementStore.getUser(userId) }
        .flatMap(handleResult("getting user"))
        .map(u => GetUserResponse(Some(toProtoUser(u))))
@ -234,7 +249,7 @@ private[apiserver] final class ApiUserManagementService(
          )
        } yield (userId, identityProviderId)
      } { case (userId, identityProviderId) =>
-        verifyUserIsNonExistentOrInIdp(identityProviderId, userId)
+        verifyUserExistsAndInIdp(identityProviderId, userId, "deleting user")
          .flatMap { _ => userManagementStore.deleteUser(userId) }
          .flatMap(handleResult("deleting user"))
          .map(_ => proto.DeleteUserResponse())
@ -279,6 +294,9 @@ private[apiserver] final class ApiUserManagementService(
  override def grantUserRights(
      request: proto.GrantUserRightsRequest
  ): Future[proto.GrantUserRightsResponse] = withSubmissionId { implicit loggingContext =>
+    // Retrieving the authenticated user context from the thread-local context
+    val authorizedUserContextF: Future[AuthenticatedUserContext] =
+      resolveAuthenticatedUserContext()
    withValidation(
      for {
        userId <- requireUserId(request.userId, "user_id")
@ -289,17 +307,30 @@ private[apiserver] final class ApiUserManagementService(
        )
      } yield (userId, rights, identityProviderId)
    ) { case (userId, rights, identityProviderId) =>
-      verifyUserIsNonExistentOrInIdp(identityProviderId, userId)
-        .flatMap { _ => userManagementStore.grantRights(id = userId, rights = rights) }
-        .flatMap(handleResult("grant user rights"))
-        .map(_.view.map(toProtoRight).toList)
-        .map(proto.GrantUserRightsResponse(_))
+      for {
+        _ <- verifyUserExistsAndInIdp(identityProviderId, userId, "grant user rights")
+        authorizedUserContext <- authorizedUserContextF
+        _ <- verifyPartyExistInIdp(
+          rights,
+          identityProviderId,
+          authorizedUserContext.isParticipantAdmin,
+        )
+        result <- userManagementStore
+          .grantRights(
+            id = userId,
+            rights = rights,
+          )
+        handledResult <- handleResult("grant user rights")(result)
+      } yield proto.GrantUserRightsResponse(handledResult.view.map(toProtoRight).toList)
    }
  }

  override def revokeUserRights(
      request: proto.RevokeUserRightsRequest
  ): Future[proto.RevokeUserRightsResponse] = withSubmissionId { implicit loggingContext =>
+    // Retrieving the authenticated user context from the thread-local context
+    val authorizedUserContextF: Future[AuthenticatedUserContext] =
+      resolveAuthenticatedUserContext()
    withValidation(
      for {
        userId <- FieldValidations.requireUserId(request.userId, "user_id")
@ -310,11 +341,21 @@ private[apiserver] final class ApiUserManagementService(
        )
      } yield (userId, rights, identityProviderId)
    ) { case (userId, rights, identityProviderId) =>
-      verifyUserIsNonExistentOrInIdp(identityProviderId, userId)
-        .flatMap { _ => userManagementStore.revokeRights(id = userId, rights = rights) }
-        .flatMap(handleResult("revoke user rights"))
-        .map(_.view.map(toProtoRight).toList)
-        .map(proto.RevokeUserRightsResponse(_))
+      for {
+        authorizedUserContext <- authorizedUserContextF
+        _ <- verifyPartyExistInIdp(
+          rights,
+          identityProviderId,
+          authorizedUserContext.isParticipantAdmin,
+        )
+        _ <- verifyUserExistsAndInIdp(identityProviderId, userId, "revoke user rights")
+        result <- userManagementStore
+          .revokeRights(
+            id = userId,
+            rights = rights,
+          )
+        handledResult <- handleResult("revoke user rights")(result)
+      } yield proto.RevokeUserRightsResponse(handledResult.view.map(toProtoRight).toList)
    }
  }

@ -330,7 +371,7 @@ private[apiserver] final class ApiUserManagementService(
        )
      } yield (userId, identityProviderId)
    } { case (userId, identityProviderId) =>
-      verifyUserIsNonExistentOrInIdp(identityProviderId, userId)
+      verifyUserExistsAndInIdp(identityProviderId, userId, "list user rights")
        .flatMap { _ => userManagementStore.listUserRights(userId) }
        .flatMap(handleResult("list user rights"))
        .map(_.view.map(toProtoRight).toList)
@ -349,6 +390,34 @@ private[apiserver] final class ApiUserManagementService(
        Future.successful(t)
    }

+  private def verifyPartyExistInIdp(
+      rights: Set[UserRight],
+      identityProviderId: IdentityProviderId,
+      isParticipantAdmin: Boolean,
+  ): Future[Unit] =
+    if (isParticipantAdmin) Future.unit
+    else partyExistsOrError(userParties(rights), identityProviderId)
+
+  private def partyExistsOrError(
+      parties: Set[Ref.Party],
+      identityProviderId: IdentityProviderId,
+  ): Future[Unit] =
+    partyRecordExist
+      .filterPartiesExistingInPartyRecordStore(identityProviderId, parties)
+      .flatMap { partiesExist =>
+        val partiesNotExist = parties -- partiesExist
+        if (partiesNotExist.isEmpty)
+          Future.successful(())
+        else
+          Future.failed(
+            LedgerApiErrors.RequestValidation.InvalidArgument
+              .Reject(
+                s"Provided parties [${partiesNotExist.mkString(",")}] have not been found in identity_provider_id=${identityProviderId.toRequestString}."
+              )
+              .asGrpcError
+          )
+      }
+
  private def identityProviderExistsOrError(id: IdentityProviderId): Future[Unit] =
    identityProviderExists(id)
      .flatMap { idpExists =>
@ -362,12 +431,13 @@ private[apiserver] final class ApiUserManagementService(
          )
      }

-  // Check if user either doesn't exist or exists and belongs to the requested Identity Provider
+  // Check if user either doesn't exist or if it does - it belongs to the requested Identity Provider
  // Alternatively, identity_provider_id could be part of the compound unique key within user database.
  // It was considered as complication for the implementation and for now is simplified.
-  private def verifyUserIsNonExistentOrInIdp(
+  private def verifyUserDoesNotExist(
      identityProviderId: IdentityProviderId,
      userId: Ref.UserId,
+      operation: String,
  ): Future[Unit] = {
    userManagementStore.getUser(userId).flatMap {
      case Right(user) if user.identityProviderId != identityProviderId =>
@ -376,6 +446,38 @@ private[apiserver] final class ApiUserManagementService(
            .Reject(s"User ${user.id} belongs to another Identity Provider")
            .asGrpcError
        )
+      case Right(_) =>
+        Future.failed(
+          LedgerApiErrors.Admin.UserManagement.UserAlreadyExists
+            .Reject(operation, userId.toString)
+            .asGrpcError
+        )
+      case _ =>
+        Future.unit
+    }
+  }
+
+  // Check if user exists and belongs to the requested Identity Provider
+  // There is a possible race condition here, as user might be deleted and reintroduced within another IDP
+  // In order to fix this, proper atomicity is to be considered, by implementing this check in the persistence layer
+  private def verifyUserExistsAndInIdp(
+      identityProviderId: IdentityProviderId,
+      userId: Ref.UserId,
+      operation: String,
+  ): Future[Unit] = {
+    userManagementStore.getUser(userId).flatMap {
+      case Right(user) if user.identityProviderId != identityProviderId =>
+        Future.failed(
+          LedgerApiErrors.AuthorizationChecks.PermissionDenied
+            .Reject(s"User ${user.id} belongs to another Identity Provider")
+            .asGrpcError
+        )
+      case Left(UserManagementStore.UserNotFound(id)) =>
+        Future.failed(
+          LedgerApiErrors.Admin.UserManagement.UserNotFound
+            .Reject(operation, id.toString)
+            .asGrpcError
+        )
      case _ => Future.unit
    }
  }
@ -420,6 +522,11 @@ private[apiserver] final class ApiUserManagementService(
        Future.successful(t)
    }

+  private def userParties(rights: Set[UserRight]): Set[Ref.Party] = rights.collect {
+    case UserRight.CanActAs(party) => party
+    case UserRight.CanReadAs(party) => party
+  }
+
  private def withValidation[A, B](validatedResult: Either[StatusRuntimeException, A])(
      f: A => Future[B]
  ): Future[B] =
@ -460,6 +567,16 @@ private[apiserver] final class ApiUserManagementService(
 }

 object ApiUserManagementService {
+  case class AuthenticatedUserContext(userId: Option[String], isParticipantAdmin: Boolean)
+  object AuthenticatedUserContext {
+    def apply(claims: Claims): AuthenticatedUserContext = claims match {
+      case claims: Claims if claims.resolvedFromUser =>
+        AuthenticatedUserContext(claims.applicationId, claims.claims.contains(ClaimAdmin))
+      case claims: Claims =>
+        AuthenticatedUserContext(None, claims.claims.contains(ClaimAdmin))
+    }
+  }
+
  private def toProtoUser(user: User): proto.User =
    proto.User(
      id = user.id,
--- a/ledger/participant-integration-api/src/main/scala/platform/apiserver/services/admin/PartyRecordsExist.scala
+++ b/ledger/participant-integration-api/src/main/scala/platform/apiserver/services/admin/PartyRecordsExist.scala
@ -0,0 +1,20 @@
+// Copyright (c) 2023 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.platform.apiserver.services.admin
+
+import com.daml.ledger.api.domain.IdentityProviderId
+import com.daml.lf.data.Ref
+import com.daml.logging.LoggingContext
+import com.daml.platform.localstore.api.PartyRecordStore
+
+import scala.concurrent.Future
+
+class PartyRecordsExist(partyRecordStore: PartyRecordStore) {
+
+  def filterPartiesExistingInPartyRecordStore(id: IdentityProviderId, parties: Set[Ref.Party])(
+      implicit loggingContext: LoggingContext
+  ): Future[Set[Ref.Party]] =
+    partyRecordStore.filterExistingParties(parties, id)
+
+}
--- a/ledger/participant-integration-api/src/main/scala/platform/localstore/PersistentPartyRecordStore.scala
+++ b/ledger/participant-integration-api/src/main/scala/platform/localstore/PersistentPartyRecordStore.scala
@ -280,4 +280,13 @@ class PersistentPartyRecordStore(
    val now = timeProvider.getCurrentTime
    (now.getEpochSecond * 1000 * 1000) + (now.getNano / 1000)
  }
+
+  override def filterExistingParties(parties: Set[Party], identityProviderId: IdentityProviderId)(
+      implicit loggingContext: LoggingContext
+  ): Future[Set[Party]] = inTransaction(_.partiesExist) { implicit connection =>
+    Right(backend.filterExistingParties(parties, identityProviderId.toDb)(connection))
+  }.map {
+    case Right(value) => value
+    case Left(_) => Set.empty
+  }
 }
--- a/ledger/participant-integration-api/src/main/scala/platform/store/backend/localstore/IdentityProviderAwareBackend.scala
+++ b/ledger/participant-integration-api/src/main/scala/platform/store/backend/localstore/IdentityProviderAwareBackend.scala
@ -1,28 +0,0 @@
-// Copyright (c) 2023 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
-// SPDX-License-Identifier: Apache-2.0
-
-package com.daml.platform.store.backend.localstore
-
-import anorm.SqlStringInterpolation
-import com.daml.ledger.api.domain.IdentityProviderId
-
-import java.sql.Connection
-
-object IdentityProviderAwareBackend {
-
-  def updateIdentityProviderId(
-      tableName: String
-  )(internalId: Int, identityProviderId: Option[IdentityProviderId.Id])(
-      connection: Connection
-  ): Boolean = {
-    val rowsUpdated =
-      SQL"""
-         UPDATE #$tableName
-         SET identity_provider_id = ${identityProviderId.map(_.value): Option[String]}
-         WHERE
-             internal_id = ${internalId}
-       """.executeUpdate()(connection)
-    rowsUpdated == 1
-  }
-
-}
--- a/ledger/participant-integration-api/src/main/scala/platform/store/backend/localstore/PartyRecordStorageBackend.scala
+++ b/ledger/participant-integration-api/src/main/scala/platform/store/backend/localstore/PartyRecordStorageBackend.scala
@ -26,10 +26,10 @@ trait PartyRecordStorageBackend extends ResourceVersionOps {

  def deletePartyAnnotations(internalId: Int)(connection: Connection): Unit

-  def updateIdentityProviderId(
-      internalId: Int,
+  def filterExistingParties(
+      parties: Set[Ref.Party],
      identityProviderId: Option[IdentityProviderId.Id],
-  )(connection: Connection): Boolean
+  )(connection: Connection): Set[Ref.Party]

 }

--- a/ledger/participant-integration-api/src/main/scala/platform/store/backend/localstore/PartyRecordStorageBackendImpl.scala
+++ b/ledger/participant-integration-api/src/main/scala/platform/store/backend/localstore/PartyRecordStorageBackendImpl.scala
@ -7,15 +7,15 @@ import anorm.SqlParser.{int, long, str}
 import anorm.{RowParser, SqlParser, SqlStringInterpolation, ~}
 import com.daml.ledger.api.domain.IdentityProviderId
 import com.daml.lf.data.Ref
-
+import com.daml.platform.store.backend.Conversions.party
 import java.sql.Connection
 import scala.util.Try

 object PartyRecordStorageBackendImpl extends PartyRecordStorageBackend {

-  private val PartyRecordParser: RowParser[(Int, String, Option[String], Long, Long)] =
+  private val PartyRecordParser: RowParser[(Int, Ref.Party, Option[String], Long, Long)] =
    int("internal_id") ~
-      str("party") ~
+      party("party") ~
      str("identity_provider_id").? ~
      long("resource_version") ~
      long("created_at") map {
@ -42,7 +42,7 @@ object PartyRecordStorageBackendImpl extends PartyRecordStorageBackend {
        PartyRecordStorageBackend.DbPartyRecord(
          internalId = internalId,
          payload = PartyRecordStorageBackend.DbPartyRecordPayload(
-            party = com.daml.platform.Party.assertFromString(party),
+            party = party,
            identityProviderId = identityProviderId.map(IdentityProviderId.Id.assertFromString),
            resourceVersion = resourceVersion,
            createdAt = createdAt,
@ -105,14 +105,26 @@ object PartyRecordStorageBackendImpl extends PartyRecordStorageBackend {
    )
  }

-  override def updateIdentityProviderId(
-      internalId: Int,
+  override def filterExistingParties(
+      parties: Set[Ref.Party],
      identityProviderId: Option[IdentityProviderId.Id],
-  )(connection: Connection): Boolean =
-    IdentityProviderAwareBackend.updateIdentityProviderId("participant_party_records")(
-      internalId,
-      identityProviderId,
-    )(
-      connection
-    )
+  )(connection: Connection): Set[Ref.Party] = if (parties.nonEmpty) {
+    import com.daml.platform.store.backend.common.SimpleSqlAsVectorOf._
+    import com.daml.platform.store.backend.common.ComposableQuery.SqlStringInterpolation
+    val filteredParties = cSQL"party in (${parties.map(_.toString)})"
+
+    val filteredIdentityProviderId = identityProviderId match {
+      case Some(id) => cSQL"identity_provider_id = ${id.value: String}"
+      case None => cSQL"identity_provider_id is NULL"
+    }
+    SQL"""
+         SELECT
+             party
+         FROM participant_party_records
+         WHERE
+             $filteredIdentityProviderId AND $filteredParties
+       """
+      .asVectorOf(party("party"))(connection)
+      .toSet
+  } else Set.empty
 }
--- a/ledger/participant-integration-api/src/main/scala/platform/store/backend/localstore/UserManagementStorageBackend.scala
+++ b/ledger/participant-integration-api/src/main/scala/platform/store/backend/localstore/UserManagementStorageBackend.scala
@ -58,10 +58,6 @@ trait UserManagementStorageBackend extends ResourceVersionOps {
      isDeactivated: Boolean,
  )(connection: Connection): Boolean

-  def updateUserIdentityProviderId(
-      internalId: Int,
-      identityProviderId: Option[IdentityProviderId.Id],
-  )(connection: Connection): Boolean
 }

 object UserManagementStorageBackend {
--- a/ledger/participant-integration-api/src/main/scala/platform/store/backend/localstore/UserManagementStorageBackendImpl.scala
+++ b/ledger/participant-integration-api/src/main/scala/platform/store/backend/localstore/UserManagementStorageBackendImpl.scala
@ -349,16 +349,4 @@ object UserManagementStorageBackendImpl extends UserManagementStorageBackend {
    rowsUpdated == 1
  }

-  override def updateUserIdentityProviderId(
-      internalId: Int,
-      identityProviderId: Option[IdentityProviderId.Id],
-  )(connection: Connection): Boolean = {
-    IdentityProviderAwareBackend.updateIdentityProviderId("participant_users")(
-      internalId,
-      identityProviderId,
-    )(
-      connection
-    )
-  }
-
 }
--- a/ledger/participant-integration-api/src/test/lib/scala/platform/store/backend/StorageBackendTestsPartyRecord.scala
+++ b/ledger/participant-integration-api/src/test/lib/scala/platform/store/backend/StorageBackendTestsPartyRecord.scala
@ -3,10 +3,12 @@

 package com.daml.platform.store.backend

+import com.daml.ledger.api.domain.{IdentityProviderConfig, IdentityProviderId, JwksUrl}
+
 import java.sql.SQLException
 import java.util.UUID
-
 import com.daml.lf.data.Ref
+import com.daml.lf.data.Ref.LedgerString
 import com.daml.platform.store.backend.localstore.PartyRecordStorageBackend
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
@ -71,16 +73,66 @@ private[backend] trait StorageBackendTestsPartyRecord
    getNonexistent shouldBe None
  }

+  it should "filter parties within the same idp" in {
+    val idpId = IdentityProviderId.Id(LedgerString.assertFromString("abc"))
+    val _ = executeSql(
+      backend.identityProviderStorageBackend.createIdentityProviderConfig(
+        IdentityProviderConfig(
+          identityProviderId = idpId,
+          issuer = "issuer",
+          jwksUrl = JwksUrl("http://daml.com/jwks.json"),
+        )
+      )
+    )
+    val party1 = Ref.Party.assertFromString("party1")
+    val party2 = Ref.Party.assertFromString("party2")
+    val partyRecord1 = newDbPartyRecord(partyId = "party1")
+    val partyRecord2 = newDbPartyRecord(
+      partyId = "party2",
+      identityProviderId = Some(idpId),
+    )
+    val _ = executeSql(tested.createPartyRecord(partyRecord1))
+    val _ = executeSql(tested.createPartyRecord(partyRecord2))
+    executeSql(
+      tested.filterExistingParties(
+        Set(),
+        Some(IdentityProviderId.Id(LedgerString.assertFromString("cde"))),
+      )
+    ) shouldBe Set.empty
+
+    executeSql(
+      tested.filterExistingParties(
+        Set(),
+        None,
+      )
+    ) shouldBe Set.empty
+
+    executeSql(
+      tested.filterExistingParties(
+        Set(party1, party2),
+        None,
+      )
+    ) shouldBe Set(party1)
+
+    executeSql(
+      tested.filterExistingParties(
+        Set(party1, party2),
+        Some(idpId),
+      )
+    ) shouldBe Set(party2)
+  }
+
  private def newDbPartyRecord(
      partyId: String = "",
      resourceVersion: Long = 0,
      createdAt: Long = zeroMicros,
+      identityProviderId: Option[IdentityProviderId.Id] = None,
  ): PartyRecordStorageBackend.DbPartyRecordPayload = {
    val uuid = UUID.randomUUID.toString
    val party = if (partyId != "") partyId else s"party_id_$uuid"
    PartyRecordStorageBackend.DbPartyRecordPayload(
      party = Ref.Party.assertFromString(party),
-      identityProviderId = None,
+      identityProviderId = identityProviderId,
      resourceVersion = resourceVersion,
      createdAt = createdAt,
    )
--- a/ledger/participant-local-store/src/main/scala/com/daml/platform/localstore/InMemoryPartyRecordStore.scala
+++ b/ledger/participant-local-store/src/main/scala/com/daml/platform/localstore/InMemoryPartyRecordStore.scala
@ -194,4 +194,14 @@ class InMemoryPartyRecordStore(executionContext: ExecutionContext) extends Party
      Right(())
    }
  }
+
+  override def filterExistingParties(parties: Set[Party], identityProviderId: IdentityProviderId)(
+      implicit loggingContext: LoggingContext
+  ): Future[Set[Party]] = {
+    withState {
+      parties.map(party => (party, state.get(party))).collect {
+        case (party, Some(record)) if record.identityProviderId == identityProviderId => party
+      }
+    }
+  }
 }
--- a/ledger/participant-local-store/src/main/scala/com/daml/platform/localstore/api/PartyRecordStore.scala
+++ b/ledger/participant-local-store/src/main/scala/com/daml/platform/localstore/api/PartyRecordStore.scala
@ -39,6 +39,10 @@ trait PartyRecordStore {
      loggingContext: LoggingContext
  ): Future[Result[Option[PartyRecord]]]

+  def filterExistingParties(parties: Set[Ref.Party], identityProviderId: IdentityProviderId)(
+      implicit loggingContext: LoggingContext
+  ): Future[Set[Ref.Party]]
+
 }

 object PartyRecordStore {
--- a/ledger/sandbox-on-x/src/test/it/scala/com/daml/ledger/platform/sandbox/auth/PartyRestrictionUserManagementAuthIT.scala
+++ b/ledger/sandbox-on-x/src/test/it/scala/com/daml/ledger/platform/sandbox/auth/PartyRestrictionUserManagementAuthIT.scala
@ -0,0 +1,114 @@
+// Copyright (c) 2023 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.platform.sandbox.auth
+
+import com.daml.error.ErrorsAssertions
+import com.daml.ledger.api.v1.admin.object_meta.ObjectMeta
+import com.daml.ledger.runner.common.Config
+import com.daml.ledger.sandbox.SandboxOnXForTest.{ApiServerConfig, singleParticipant}
+import com.daml.ledger.api.v1.admin.{user_management_service => uproto}
+import com.daml.ledger.api.v1.admin.{party_management_service => pproto}
+
+import java.util.UUID
+import scala.concurrent.Future
+
+final class PartyRestrictionUserManagementAuthIT
+    extends ServiceCallAuthTests
+    with IdentityProviderConfigAuth
+    with ErrorsAssertions {
+
+  private val UserManagementCacheExpiryInSeconds = 1
+
+  override def config: Config = super.config.copy(
+    participants = singleParticipant(
+      ApiServerConfig.copy(
+        userManagement = ApiServerConfig.userManagement
+          .copy(
+            cacheExpiryAfterWriteInSeconds = UserManagementCacheExpiryInSeconds
+          )
+      )
+    )
+  )
+
+  override def serviceCallName: String = ""
+
+  override protected def serviceCall(context: ServiceCallContext): Future[Any] = ???
+
+  private def createUser(
+      userId: String,
+      serviceCallContext: ServiceCallContext,
+      rights: Vector[uproto.Right] = Vector.empty,
+  ): Future[uproto.CreateUserResponse] = {
+    val user = uproto.User(
+      id = userId,
+      metadata = Some(ObjectMeta()),
+      identityProviderId = serviceCallContext.identityProviderId,
+    )
+    val req = uproto.CreateUserRequest(Some(user), rights)
+    stub(uproto.UserManagementServiceGrpc.stub(channel), serviceCallContext.token)
+      .createUser(req)
+  }
+
+  private def allocateParty(serviceCallContext: ServiceCallContext, party: String) =
+    stub(pproto.PartyManagementServiceGrpc.stub(channel), serviceCallContext.token)
+      .allocateParty(
+        pproto.AllocatePartyRequest(
+          partyIdHint = party,
+          identityProviderId = serviceCallContext.identityProviderId,
+        )
+      )
+
+  it should "allow to grant permissions to parties which are allocated in the IDP" in {
+
+    for {
+      idpConfig <- createConfig(canReadAsAdmin)
+      identityProviderConfig = idpConfig.identityProviderConfig.getOrElse(
+        sys.error("Unable to create IdentityProviderConfig")
+      )
+      (_, idpAdmin) <- createUserByAdmin(
+        userId = UUID.randomUUID().toString + "-alice-1",
+        tokenIssuer = Some(identityProviderConfig.issuer),
+        identityProviderId = identityProviderConfig.identityProviderId,
+        rights = Vector(
+          uproto.Right(
+            uproto.Right.Kind.IdentityProviderAdmin(uproto.Right.IdentityProviderAdmin())
+          ),
+          uproto.Right(uproto.Right.Kind.CanReadAs(uproto.Right.CanReadAs("some-party-2"))),
+        ),
+        primaryParty = "some-party-1",
+      )
+      _ <- createUser(
+        UUID.randomUUID().toString + "-alice-2",
+        idpAdmin,
+      )
+
+      _ <- expectInvalidArgument(
+        createUser(
+          UUID.randomUUID().toString + "-alice-3",
+          idpAdmin,
+          Vector(
+            uproto.Right(uproto.Right.Kind.CanReadAs(uproto.Right.CanReadAs("some-party-1"))),
+            uproto.Right(uproto.Right.Kind.CanActAs(uproto.Right.CanActAs("some-party-2"))),
+          ),
+        )
+      )
+
+      _ <- allocateParty(idpAdmin, "some-party-1")
+      _ <- allocateParty(idpAdmin, "some-party-2")
+
+      _ <- createUser(
+        UUID.randomUUID().toString + "-alice-4",
+        idpAdmin,
+        Vector(
+          uproto.Right(uproto.Right.Kind.CanReadAs(uproto.Right.CanReadAs("some-party-1"))),
+          uproto.Right(uproto.Right.Kind.CanActAs(uproto.Right.CanActAs("some-party-2"))),
+        ),
+      )
+    } yield {
+      assert(true)
+    }
+
+  }
+
+}
--- a/ledger/sandbox-on-x/src/test/it/scala/com/daml/ledger/platform/sandbox/auth/UpdateUserSelfDeactivationAuthIT.scala
+++ b/ledger/sandbox-on-x/src/test/it/scala/com/daml/ledger/platform/sandbox/auth/UpdateUserSelfDeactivationAuthIT.scala
@ -8,7 +8,6 @@ import com.daml.error.utils.ErrorDetails
 import com.daml.ledger.api.v1.admin.user_management_service.Right
 import com.daml.ledger.runner.common.Config
 import com.daml.ledger.sandbox.SandboxOnXForTest.{ApiServerConfig, singleParticipant}
-import com.daml.platform.sandbox.services.SubmitAndWaitDummyCommandHelpers
 import com.google.protobuf.field_mask.FieldMask
 import io.grpc.{Status, StatusRuntimeException}

@ -16,10 +15,7 @@ import java.util.UUID
 import scala.concurrent.Future
 import scala.util.{Failure, Success}

-final class UpdateUserSelfDeactivationAuthIT
-    extends ServiceCallAuthTests
-    with SubmitAndWaitDummyCommandHelpers
-    with ErrorsAssertions {
+final class UpdateUserSelfDeactivationAuthIT extends ServiceCallAuthTests with ErrorsAssertions {

  private val UserManagementCacheExpiryInSeconds = 1

--- a/ledger/sandbox-on-x/src/test/lib/scala/com/daml/platform/sandbox/auth/ServiceCallAuthTests.scala
+++ b/ledger/sandbox-on-x/src/test/lib/scala/com/daml/platform/sandbox/auth/ServiceCallAuthTests.scala
@ -212,6 +212,7 @@ trait ServiceCallAuthTests
      rights: Vector[proto.Right] = Vector.empty,
      tokenIssuer: Option[String] = None,
      secret: Option[String] = None,
+      primaryParty: String = "",
  ): Future[(proto.User, ServiceCallContext)] = {
    val userToken = Option(
      toHeader(standardToken(userId, issuer = tokenIssuer), secret = secret.getOrElse(jwtSecret))
@ -220,6 +221,7 @@ trait ServiceCallAuthTests
      id = userId,
      metadata = Some(ObjectMeta()),
      identityProviderId = identityProviderId,
+      primaryParty = primaryParty,
    )
    val req = proto.CreateUserRequest(Some(user), rights)
    stub(proto.UserManagementServiceGrpc.stub(channel), canReadAsAdminStandardJWT.token)