PROD-972 Limit google group syncs (#1475)

src/main/scala/org/broadinstitute/dsde/workbench/sam/dataAccess/AccessPolicyDAO.scala

@@ -31,8 +31,16 @@ trait AccessPolicyDAO {
       samRequestContext: SamRequestContext
   ): IO[Unit]
 
+  /** Lists policies on resources that are constrained by the given group. If relevantMembers is provided, only policies that contain any of the relevantMembers
+    * (directly or indirectly) will be returned.
+    * @param groupId
+    *   the group to constrain by
+    * @param relevantMembers
+    *   if provided, only policies that contain any of the relevantMembers (directly or indirectly) will be returned, if empty, all policies will be returned
+    */
   def listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(
       groupId: WorkbenchGroupIdentity,
+      relevantMembers: Set[WorkbenchSubject],
       samRequestContext: SamRequestContext
   ): IO[Set[FullyQualifiedPolicyId]]
 

src/main/scala/org/broadinstitute/dsde/workbench/sam/dataAccess/PostgresAccessPolicyDAO.scala

@@ -641,6 +641,7 @@ class PostgresAccessPolicyDAO(
 
   override def listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(
       groupId: WorkbenchGroupIdentity,
+      relevantMembers: Set[WorkbenchSubject],
       samRequestContext: SamRequestContext
   ): IO[Set[FullyQualifiedPolicyId]] =
     readOnlyTransaction("listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup", samRequestContext) { implicit session =>
@@ -667,14 +668,40 @@ class PostgresAccessPolicyDAO(
            and ${policy.resource.resourceTypeName} = ${rt.name}"""
       }
 
+      // if relevantMembers is empty, assume all members are relevant and don't join on GroupMemberFlatTable
+      // otherwise, only include policies where the member is in the relevantMembers set
+      // need to account for both member groups and users
+      val pu = GroupMemberFlatTable.syntax("pu")
+      val (relevantMembersJoin, relevantMembersCondition) = if (relevantMembers.isEmpty) {
+        (samsqls"", samsqls"")
+      } else {
+        val groupPKs = queryForGroupPKs(relevantMembers)
+        val groupCondition = if (groupPKs.isEmpty) {
+          samsqls"false"
+        } else {
+          samsqls"${pu.memberGroupId} in (${groupPKs})"
+        }
+
+        val userIds = collectUserIds(relevantMembers)
+        val userCondition = if (userIds.isEmpty) {
+          samsqls"false"
+        } else {
+          samsqls"${pu.memberUserId} in (${userIds})"
+        }
+
+        (samsqls"""join ${GroupMemberFlatTable as pu} on ${pu.groupId} = ${p.groupId}""", samsqls"""and (${userCondition} or ${groupCondition})""")
+      }
+
       samsql"""
           select ${rt.result.name}, ${r.result.name}, ${p.result.name}
            from ${ResourceTable as r}
            join ${ResourceTypeTable as rt} on ${r.resourceTypeId} = ${rt.id}
            join ${PolicyTable as p} on ${r.id} = ${p.resourceId}
            join ${GroupTable as g} on ${p.groupId} = ${g.id}
+           ${relevantMembersJoin}
            where ${r.id} in (${constrainedResourcesPKs})
-           and ${g.synchronizedDate} is not null"""
+           and ${g.synchronizedDate} is not null
+           ${relevantMembersCondition}"""
         .map(rs =>
           FullyQualifiedPolicyId(
             FullyQualifiedResourceId(rs.get[ResourceTypeName](rt.resultName.name), rs.get[ResourceId](r.resultName.name)),

src/main/scala/org/broadinstitute/dsde/workbench/sam/google/GoogleExtensions.scala

@@ -188,7 +188,7 @@ class GoogleExtensions(
 
   /*
     - managed groups and access policies are both "groups"
-    - You can have a bunch of resources constrained an auth domain (a collection of managed groups).
+    - You can have a bunch of resources constrained by an auth domain (a collection of managed groups).
     - A user must be a member of the auth domain in order to access some actions on the resources in that auth domain.
     - The user must be a member of all groups in an auth domain in order to access a resource
     - An access policy is specific to a single resource
@@ -207,7 +207,11 @@ class GoogleExtensions(
 
      see GoogleGroupSynchronizer for the background process that does the group synchronization
    */
-  override def onGroupUpdate(groupIdentities: Seq[WorkbenchGroupIdentity], samRequestContext: SamRequestContext): IO[Unit] =
+  override def onGroupUpdate(
+      groupIdentities: Seq[WorkbenchGroupIdentity],
+      relevantMembers: Set[WorkbenchSubject],
+      samRequestContext: SamRequestContext
+  ): IO[Unit] =
     for {
       start <- clock.monotonic
       // only sync groups that have been synchronized in the past
@@ -219,14 +223,14 @@ class GoogleExtensions(
       messages <- previouslySyncedIds.traverse {
         // it is a group that isn't an access policy, could be a managed group
         case groupName: WorkbenchGroupName =>
-          makeConstrainedResourceAccessPolicyMessages(groupName, samRequestContext).map(_ :+ groupName.toJson.compactPrint)
+          makeConstrainedResourceAccessPolicyMessages(groupName, relevantMembers, samRequestContext).map(_ :+ groupName.toJson.compactPrint)
 
         // it is the admin or member access policy of a managed group
         case accessPolicyId @ FullyQualifiedPolicyId(
               FullyQualifiedResourceId(ManagedGroupService.managedGroupTypeName, id),
               ManagedGroupService.adminPolicyName | ManagedGroupService.memberPolicyName
             ) =>
-          makeConstrainedResourceAccessPolicyMessages(accessPolicyId, samRequestContext).map(_ :+ accessPolicyId.toJson.compactPrint)
+          makeConstrainedResourceAccessPolicyMessages(accessPolicyId, relevantMembers, samRequestContext).map(_ :+ accessPolicyId.toJson.compactPrint)
 
         // it is an access policy on a resource that's not a managed group
         case accessPolicyId: FullyQualifiedPolicyId => IO.pure(List(accessPolicyId.toJson.compactPrint))
@@ -245,7 +249,11 @@ class GoogleExtensions(
       )
     }
 
-  private def makeConstrainedResourceAccessPolicyMessages(groupIdentity: WorkbenchGroupIdentity, samRequestContext: SamRequestContext): IO[List[String]] =
+  private def makeConstrainedResourceAccessPolicyMessages(
+      groupIdentity: WorkbenchGroupIdentity,
+      relevantMembers: Set[WorkbenchSubject],
+      samRequestContext: SamRequestContext
+  ): IO[List[String]] =
     // start with a group
     for {
       // get all the ancestors of that group
@@ -262,7 +270,7 @@ class GoogleExtensions(
 
       // get all access policies on any resource that is constrained by the groups
       constrainedResourceAccessPolicyIds <- managedGroupIds.toList.traverse(
-        accessPolicyDAO.listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(_, samRequestContext)
+        accessPolicyDAO.listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(_, relevantMembers, samRequestContext)
       )
 
       // return messages for all the affected access policies and the original group we started with

src/main/scala/org/broadinstitute/dsde/workbench/sam/model/SamModel.scala

@@ -192,7 +192,9 @@ object RolesAndActions {
     policyName: AccessPolicyName,
     resourceTypeName: ResourceTypeName,
     resourceId: ResourceId
-)
+) {
+  def toFullyQualifiedPolicyId: FullyQualifiedPolicyId = FullyQualifiedPolicyId(FullyQualifiedResourceId(resourceTypeName, resourceId), policyName)
+}
 
 @Lenses case class AccessPolicyName(value: String) extends ValueObject
 @Lenses final case class CreateResourceRequest(

src/main/scala/org/broadinstitute/dsde/workbench/sam/service/CloudExtensions.scala

@@ -31,7 +31,13 @@ trait CloudExtensions {
 
   def publishGroup(id: WorkbenchGroupName): Future[Unit]
 
-  def onGroupUpdate(groupIdentities: Seq[WorkbenchGroupIdentity], samRequestContext: SamRequestContext): IO[Unit]
+  /** This method is called when a group is updated.
+    * @param groupIdentities
+    *   the identities of the groups that were updated
+    * @param relevantMembers
+    *   the members of the groups that were added or removed or empty if the members are not known
+    */
+  def onGroupUpdate(groupIdentities: Seq[WorkbenchGroupIdentity], relevantMembers: Set[WorkbenchSubject], samRequestContext: SamRequestContext): IO[Unit]
 
   def onGroupDelete(groupEmail: WorkbenchEmail): IO[Unit]
 
@@ -74,7 +80,11 @@ trait NoExtensions extends CloudExtensions {
 
   override def publishGroup(id: WorkbenchGroupName): Future[Unit] = Future.successful(())
 
-  override def onGroupUpdate(groupIdentities: Seq[WorkbenchGroupIdentity], samRequestContext: SamRequestContext): IO[Unit] = IO.unit
+  override def onGroupUpdate(
+      groupIdentities: Seq[WorkbenchGroupIdentity],
+      relevantMembers: Set[WorkbenchSubject],
+      samRequestContext: SamRequestContext
+  ): IO[Unit] = IO.unit
 
   override def onGroupDelete(groupEmail: WorkbenchEmail): IO[Unit] = IO.unit
 

src/main/scala/org/broadinstitute/dsde/workbench/sam/service/ResourceService.scala

@@ -288,7 +288,7 @@ class ResourceService(
       policies <- listResourcePolicies(resource, samRequestContext)
       _ <- accessPolicyDAO.addResourceAuthDomain(resource, authDomains, samRequestContext)
       _ <- policies.traverse(p => directoryDAO.updateGroupUpdatedDateAndVersionWithSession(FullyQualifiedPolicyId(resource, p.policyName), samRequestContext))
-      _ <- cloudExtensions.onGroupUpdate(policies.map(p => FullyQualifiedPolicyId(resource, p.policyName)), samRequestContext)
+      _ <- cloudExtensions.onGroupUpdate(policies.map(p => FullyQualifiedPolicyId(resource, p.policyName)), Set.empty, samRequestContext)
       authDomains <- loadResourceAuthDomain(resource, samRequestContext)
     } yield authDomains
 
@@ -678,11 +678,13 @@ class ResourceService(
   private def onPolicyUpdate(policyId: FullyQualifiedPolicyId, originalPolicies: Iterable[AccessPolicy], samRequestContext: SamRequestContext): IO[Unit] =
     for {
       updatedPolicies <- accessPolicyDAO.listAccessPolicies(policyId.resource, samRequestContext)
-      changeEvents = createAccessChangeEvents(policyId.resource, originalPolicies, updatedPolicies)
+      removedMembers = originalPolicies.flatMap(_.members).toSet -- updatedPolicies.flatMap(_.members).toSet
+      addedMembers = updatedPolicies.flatMap(_.members).toSet -- originalPolicies.flatMap(_.members).toSet
 
+      changeEvents = createAccessChangeEvents(policyId.resource, originalPolicies, updatedPolicies)
       _ <- AuditLogger.logAuditEventIO(samRequestContext, changeEvents.toSeq: _*)
 
-      _ <- cloudExtensions.onGroupUpdate(Seq(policyId), samRequestContext).attempt.flatMap {
+      _ <- cloudExtensions.onGroupUpdate(Seq(policyId), removedMembers ++ addedMembers, samRequestContext).attempt.flatMap {
         case Left(regrets) => IO(logger.error(s"error calling cloudExtensions.onGroupUpdate for $policyId", regrets))
         case Right(_) => IO.unit
       }

src/main/scala/org/broadinstitute/dsde/workbench/sam/service/UserService.scala

@@ -227,7 +227,7 @@ class UserService(
     for {
       _ <- updateInvitedUser(userToRegister, samRequestContext)
       groups <- directoryDAO.listUserDirectMemberships(userToRegister.id, samRequestContext)
-      _ <- cloudExtensions.onGroupUpdate(groups, samRequestContext)
+      _ <- cloudExtensions.onGroupUpdate(groups, Set(invitedUserId), samRequestContext)
     } yield userToRegister
   }
 

src/test/scala/org/broadinstitute/dsde/workbench/sam/dataAccess/MockAccessPolicyDAO.scala

@@ -95,6 +95,7 @@ class MockAccessPolicyDAO(private val resourceTypes: mutable.Map[ResourceTypeNam
 
   override def listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(
       groupId: WorkbenchGroupIdentity,
+      relevantMembers: Set[WorkbenchSubject],
       samRequestContext: SamRequestContext
   ): IO[Set[FullyQualifiedPolicyId]] = IO {
     val groupName: WorkbenchGroupName = groupId match {

src/test/scala/org/broadinstitute/dsde/workbench/sam/dataAccess/PostgresAccessPolicyDAOSpec.scala

@@ -668,6 +668,14 @@ class PostgresAccessPolicyDAOSpec extends AnyFreeSpec with Matchers with BeforeA
       "can find all synced policies for resources with the group in its auth domain" in {
         assume(databaseEnabled, databaseEnabledClue)
 
+        val user = Generator.genWorkbenchUserBoth.sample.get
+        dirDao.createUser(user, samRequestContext).unsafeRunSync()
+        val groupUser = Generator.genWorkbenchUserBoth.sample.get
+        dirDao.createUser(groupUser, samRequestContext).unsafeRunSync()
+
+        val group = BasicWorkbenchGroup(Generator.genWorkbenchGroupName.sample.get, Set(groupUser.id), Generator.genNonPetEmail.sample.get)
+        dirDao.createGroup(group, samRequestContext = samRequestContext).unsafeRunSync()
+
         dao.createResourceType(resourceType, samRequestContext).unsafeRunSync()
         val secondResourceType = resourceType.copy(name = ResourceTypeName("superAwesomeResourceType"))
         dao.createResourceType(secondResourceType, samRequestContext).unsafeRunSync()
@@ -681,7 +689,7 @@ class PostgresAccessPolicyDAOSpec extends AnyFreeSpec with Matchers with BeforeA
         val resource2FullyQualifiedId = FullyQualifiedResourceId(secondResourceType.name, ResourceId("resource2"))
         val policy1 = AccessPolicy(
           FullyQualifiedPolicyId(resource1FullyQualifiedId, AccessPolicyName("policyName1")),
-          Set.empty,
+          Set(user.id),
           WorkbenchEmail("[email protected]"),
           resourceType.roles.map(_.roleName),
           Set(readAction, writeAction),
@@ -690,7 +698,7 @@ class PostgresAccessPolicyDAOSpec extends AnyFreeSpec with Matchers with BeforeA
         )
         val policy2 = AccessPolicy(
           FullyQualifiedPolicyId(resource1FullyQualifiedId, AccessPolicyName("policyName2")),
-          Set.empty,
+          Set(group.id, user.id),
           WorkbenchEmail("[email protected]"),
           resourceType.roles.map(_.roleName),
           Set(readAction, writeAction),
@@ -699,25 +707,69 @@ class PostgresAccessPolicyDAOSpec extends AnyFreeSpec with Matchers with BeforeA
         )
         val policy3 = AccessPolicy(
           FullyQualifiedPolicyId(resource2FullyQualifiedId, AccessPolicyName("policyName3")),
-          Set.empty,
+          Set(group.id),
           WorkbenchEmail("[email protected]"),
           secondResourceType.roles.map(_.roleName),
           Set(readAction, writeAction),
           Set.empty,
           false
         )
+        val policy4 = AccessPolicy(
+          FullyQualifiedPolicyId(resource2FullyQualifiedId, AccessPolicyName("policyName4")),
+          Set.empty,
+          WorkbenchEmail("[email protected]"),
+          secondResourceType.roles.map(_.roleName),
+          Set(readAction, writeAction),
+          Set.empty,
+          false
+        )
         val resource1 =
           Resource(resource1FullyQualifiedId.resourceTypeName, resource1FullyQualifiedId.resourceId, Set(sharedAuthDomain.id), Set(policy1, policy2))
         val resource2 =
-          Resource(resource2FullyQualifiedId.resourceTypeName, resource2FullyQualifiedId.resourceId, Set(sharedAuthDomain.id, otherGroup.id), Set(policy3))
+          Resource(
+            resource2FullyQualifiedId.resourceTypeName,
+            resource2FullyQualifiedId.resourceId,
+            Set(sharedAuthDomain.id, otherGroup.id),
+            Set(policy3, policy4)
+          )
         dao.createResource(resource1, samRequestContext).unsafeRunSync()
         dao.createResource(resource2, samRequestContext).unsafeRunSync()
 
         dirDao.updateSynchronizedDateAndVersion(policy1, samRequestContext).unsafeRunSync()
         dirDao.updateSynchronizedDateAndVersion(policy3, samRequestContext).unsafeRunSync()
+        dirDao.updateSynchronizedDateAndVersion(policy4, samRequestContext).unsafeRunSync()
 
-        dao.listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(sharedAuthDomain.id, samRequestContext).unsafeRunSync() should contain theSameElementsAs Set(
+        // finds all synced policies when no members specified
+        dao
+          .listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(sharedAuthDomain.id, Set.empty, samRequestContext)
+          .unsafeRunSync() should contain theSameElementsAs Set(
           policy1.id,
+          policy3.id,
+          policy4.id
+        )
+        // finds only relevant synced policies when user and group specified
+        dao
+          .listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(sharedAuthDomain.id, Set(user.id, group.id), samRequestContext)
+          .unsafeRunSync() should contain theSameElementsAs Set(
+          policy1.id,
+          policy3.id
+        )
+        // finds only relevant synced policies when user specified
+        dao
+          .listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(sharedAuthDomain.id, Set(user.id), samRequestContext)
+          .unsafeRunSync() should contain theSameElementsAs Set(
+          policy1.id
+        )
+        // finds only relevant synced policies when group specified
+        dao
+          .listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(sharedAuthDomain.id, Set(group.id), samRequestContext)
+          .unsafeRunSync() should contain theSameElementsAs Set(
+          policy3.id
+        )
+        // finds only relevant synced policies when user in group specified
+        dao
+          .listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(sharedAuthDomain.id, Set(groupUser.id), samRequestContext)
+          .unsafeRunSync() should contain theSameElementsAs Set(
           policy3.id
         )
       }
@@ -728,13 +780,15 @@ class PostgresAccessPolicyDAOSpec extends AnyFreeSpec with Matchers with BeforeA
         val group = BasicWorkbenchGroup(WorkbenchGroupName("boringGroup"), Set.empty, WorkbenchEmail("[email protected]"))
         dirDao.createGroup(group, samRequestContext = samRequestContext).unsafeRunSync()
 
-        dao.listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(group.id, samRequestContext).unsafeRunSync() shouldEqual Set.empty
+        dao.listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(group.id, Set.empty, samRequestContext).unsafeRunSync() shouldEqual Set.empty
       }
 
       "returns an empty list if group doesn't exist" in {
         assume(databaseEnabled, databaseEnabledClue)
 
-        dao.listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(WorkbenchGroupName("notEvenReal"), samRequestContext).unsafeRunSync() shouldEqual Set.empty
+        dao
+          .listSyncedAccessPolicyIdsOnResourcesConstrainedByGroup(WorkbenchGroupName("notEvenReal"), Set.empty, samRequestContext)
+          .unsafeRunSync() shouldEqual Set.empty
       }
     }