[CORE-211] Convert disk operations to new Sam permissions model

http/src/main/scala/org/broadinstitute/dsde/workbench/leonardo/dao/sam/SamUtils.scala

@@ -5,6 +5,8 @@ import akka.http.scaladsl.model.headers.OAuth2BearerToken
 import cats.effect.Async
 import cats.implicits.{catsSyntaxApplicativeError, toFlatMapOps}
 import cats.mtl.Ask
+import org.broadinstitute.dsde.workbench.google2.DiskName
+import org.broadinstitute.dsde.workbench.leonardo.http.service.{DiskNotFoundByIdException, DiskNotFoundException}
 import org.broadinstitute.dsde.workbench.leonardo.model.{
   ForbiddenError,
   LeoException,
@@ -14,66 +16,110 @@ import org.broadinstitute.dsde.workbench.leonardo.model.{
 import org.broadinstitute.dsde.workbench.leonardo.{
   AppContext,
   CloudContext,
+  DiskId,
+  PersistentDiskAction,
   RuntimeAction,
   RuntimeName,
+  SamResourceAction,
   SamResourceId,
   WorkspaceId
 }
-import org.broadinstitute.dsde.workbench.model.{UserInfo, WorkbenchEmail}
+import org.broadinstitute.dsde.workbench.model.{TraceId, UserInfo, WorkbenchEmail}
 
-trait SamUtils[F[_]] {
-  val samService: SamService[F]
-
-  def checkRuntimeAction(userInfo: UserInfo,
-                         cloudContext: CloudContext,
-                         runtimeName: RuntimeName,
-                         samResourceId: SamResourceId,
-                         action: RuntimeAction,
-                         userEmail: Option[WorkbenchEmail] = None
+object SamUtils {
+  def checkRuntimeAction[F[_]](samService: SamService[F],
+                               userInfo: UserInfo,
+                               cloudContext: CloudContext,
+                               runtimeName: RuntimeName,
+                               samResourceId: SamResourceId,
+                               action: RuntimeAction,
+                               userEmail: Option[WorkbenchEmail] = None
   )(implicit F: Async[F], as: Ask[F, AppContext]): F[Unit] =
-    checkRuntimeActionInternal(
+    checkActionInternal(
+      samService,
       userInfo.accessToken,
       userEmail.getOrElse(userInfo.userEmail),
       samResourceId,
       action,
+      RuntimeAction.GetRuntimeStatus,
       RuntimeNotFoundException(cloudContext, runtimeName, "Not found in database")
     )
 
-  def checkRuntimeAction(userInfo: UserInfo,
-                         workspaceId: WorkspaceId,
-                         runtimeName: RuntimeName,
-                         samResourceId: SamResourceId,
-                         action: RuntimeAction
+  def checkRuntimeAction[F[_]](samService: SamService[F],
+                               userInfo: UserInfo,
+                               workspaceId: WorkspaceId,
+                               runtimeName: RuntimeName,
+                               samResourceId: SamResourceId,
+                               action: RuntimeAction
   )(implicit F: Async[F], as: Ask[F, AppContext]): F[Unit] =
-    checkRuntimeActionInternal(
+    checkActionInternal(
+      samService,
       userInfo.accessToken,
       userInfo.userEmail,
       samResourceId,
       action,
+      RuntimeAction.GetRuntimeStatus,
       RuntimeNotFoundByWorkspaceIdException(workspaceId, runtimeName, "Not found in database")
     )
 
-  private def checkRuntimeActionInternal(userToken: OAuth2BearerToken,
-                                         userEmail: WorkbenchEmail,
-                                         samResourceId: SamResourceId,
-                                         action: RuntimeAction,
-                                         notFoundException: LeoException
+  def checkDiskAction[F[_]](samService: SamService[F],
+                            userInfo: UserInfo,
+                            cloudContext: CloudContext,
+                            diskName: DiskName,
+                            samResourceId: SamResourceId,
+                            action: SamResourceAction,
+                            traceId: TraceId
+  )(implicit F: Async[F], as: Ask[F, AppContext]): F[Unit] =
+    checkActionInternal(
+      samService,
+      userInfo.accessToken,
+      userInfo.userEmail,
+      samResourceId,
+      action,
+      PersistentDiskAction.ReadPersistentDisk,
+      DiskNotFoundException(cloudContext, diskName, traceId)
+    )
+
+  def checkDiskAction[F[_]](samService: SamService[F],
+                            userInfo: UserInfo,
+                            diskId: DiskId,
+                            samResourceId: SamResourceId,
+                            action: SamResourceAction,
+                            traceId: TraceId
+  )(implicit F: Async[F], as: Ask[F, AppContext]): F[Unit] =
+    checkActionInternal(
+      samService,
+      userInfo.accessToken,
+      userInfo.userEmail,
+      samResourceId,
+      action,
+      PersistentDiskAction.ReadPersistentDisk,
+      DiskNotFoundByIdException(diskId, traceId)
+    )
+
+  private def checkActionInternal[F[_]](samService: SamService[F],
+                                        userToken: OAuth2BearerToken,
+                                        userEmail: WorkbenchEmail,
+                                        samResourceId: SamResourceId,
+                                        actionToCheck: SamResourceAction,
+                                        resourceReadAction: SamResourceAction,
+                                        notFoundException: LeoException
   )(implicit F: Async[F], as: Ask[F, AppContext]): F[Unit] =
     samService
-      .checkAuthorized(userToken.token, samResourceId, action)
+      .checkAuthorized(userToken.token, samResourceId, actionToCheck)
       .handleErrorWith {
-        // If we've already checked read access and the user doesn't have it, pretend the runtime doesn't exist to avoid leaking its existence
-        case e: SamException if e.statusCode == StatusCodes.Forbidden && action == RuntimeAction.GetRuntimeStatus =>
+        // If we've already checked read access and the user doesn't have it, pretend the resource doesn't exist to avoid leaking its existence
+        case e: SamException if e.statusCode == StatusCodes.Forbidden && actionToCheck == resourceReadAction =>
           F.raiseError(notFoundException)
-        // Check if the user can read the runtime to determine which error to raise
+        // Check if the user can read the resource to determine which error to raise
         case e: SamException if e.statusCode == StatusCodes.Forbidden =>
           samService
-            .checkAuthorized(userToken.token, samResourceId, RuntimeAction.GetRuntimeStatus)
+            .checkAuthorized(userToken.token, samResourceId, resourceReadAction)
             .attempt
             .flatMap {
-              // The user can read the runtime, but they don't have the required action. Raise the original Forbidden action from Sam
+              // The user can read the resource, but they don't have the required action. Raise the original Forbidden action from Sam
               case Right(_) => F.raiseError(ForbiddenError(userEmail))
-              // The user can't read the runtime, pretend it doesn't exist to avoid leaking its existence
+              // The user can't read the resource, pretend it doesn't exist to avoid leaking its existence
               case Left(_) => F.raiseError(notFoundException)
             }
       }

http/src/main/scala/org/broadinstitute/dsde/workbench/leonardo/db/DiskServiceDbQueries.scala

@@ -3,6 +3,7 @@ package db
 
 import cats.syntax.all._
 import org.broadinstitute.dsde.workbench.google2.DiskName
+import org.broadinstitute.dsde.workbench.leonardo.SamResourceId.PersistentDiskSamResourceId
 import org.broadinstitute.dsde.workbench.leonardo.db.LeoProfile.api._
 import org.broadinstitute.dsde.workbench.leonardo.db.LeoProfile.mappedColumnImplicits._
 import org.broadinstitute.dsde.workbench.leonardo.db.LeoProfile.unmarshalDestroyedDate
@@ -15,19 +16,51 @@ import scala.concurrent.ExecutionContext
 
 object DiskServiceDbQueries {
 
-  def listDisks(labelMap: LabelMap,
-                includeDeleted: Boolean,
-                creatorOnly: Option[WorkbenchEmail],
-                cloudContextOpt: Option[CloudContext] = None,
-                workspaceOpt: Option[WorkspaceId] = None
+  def listDisksBySamIds(samDiskResourceIds: List[PersistentDiskSamResourceId],
+                        labelMap: LabelMap,
+                        includeDeleted: Boolean,
+                        creatorOnly: Option[WorkbenchEmail],
+                        cloudContextOpt: Option[CloudContext] = None,
+                        workspaceOpt: Option[WorkspaceId] = None
   )(implicit
     ec: ExecutionContext
   ): DBIO[List[PersistentDisk]] = {
+    val listDiskQuery = persistentDiskQuery.tableQuery.filter(_.samResourceId inSetBind samDiskResourceIds)
 
+    filterListDisks(listDiskQuery, labelMap, includeDeleted, creatorOnly, cloudContextOpt, workspaceOpt)
+  }
+
+  def listDisks(
+    labelMap: LabelMap,
+    includeDeleted: Boolean,
+    creatorOnly: Option[WorkbenchEmail],
+    cloudContextOpt: Option[CloudContext] = None,
+    workspaceOpt: Option[WorkspaceId] = None
+  )(implicit
+    ec: ExecutionContext
+  ): DBIO[List[PersistentDisk]] =
+    filterListDisks(persistentDiskQuery.tableQuery,
+                    labelMap,
+                    includeDeleted,
+                    creatorOnly,
+                    cloudContextOpt,
+                    workspaceOpt
+    )
+
+  private def filterListDisks(
+    baseQuery: Query[PersistentDiskTable, PersistentDiskRecord, Seq],
+    labelMap: LabelMap,
+    includeDeleted: Boolean,
+    creatorOnly: Option[WorkbenchEmail],
+    cloudContextOpt: Option[CloudContext] = None,
+    workspaceOpt: Option[WorkspaceId] = None
+  )(implicit
+    ec: ExecutionContext
+  ): DBIO[List[PersistentDisk]] = {
     // filtered by creator first as it may have great impact
     val diskQueryFilteredByCreator = creatorOnly match {
-      case Some(email) => persistentDiskQuery.tableQuery.filter(_.creator === email)
-      case None        => persistentDiskQuery.tableQuery
+      case Some(email) => baseQuery.filter(_.creator === email)
+      case None        => baseQuery
     }
 
     val diskQueryFilteredByDeletion =

http/src/main/scala/org/broadinstitute/dsde/workbench/leonardo/http/AppDependenciesBuilder.scala

@@ -91,9 +91,9 @@ class AppDependenciesBuilder(baselineDependenciesBuilder: BaselineDependenciesBu
   ): Resource[IO, ServicesDependencies] = {
     val statusService = new StatusService(baselineDependencies.samDAO, dbReference)
     val diskV2Service = new DiskV2ServiceInterp[IO](
-      baselineDependencies.authProvider,
       baselineDependencies.publisherQueue,
-      baselineDependencies.wsmClientProvider
+      baselineDependencies.wsmClientProvider,
+      baselineDependencies.samService
     )
 
     val azureService = new RuntimeV2ServiceInterp[IO](

http/src/main/scala/org/broadinstitute/dsde/workbench/leonardo/http/AzureDependenciesBuilder.scala

@@ -93,7 +93,6 @@ class AzureDependenciesBuilder extends CloudDependenciesBuilder {
     // Needed for v1 APIs
     val diskService = new DiskServiceInterp[IO](
       ConfigReader.appConfig.persistentDisk,
-      baselineDependencies.authProvider,
       baselineDependencies.publisherQueue,
       None,
       None,
@@ -103,7 +102,6 @@ class AzureDependenciesBuilder extends CloudDependenciesBuilder {
     val runtimeService = RuntimeService(
       baselineDependencies.runtimeServicesConfig,
       ConfigReader.appConfig.persistentDisk,
-      baselineDependencies.authProvider,
       baselineDependencies.dockerDAO,
       None,
       None,

http/src/main/scala/org/broadinstitute/dsde/workbench/leonardo/http/GcpDependenciesBuilder.scala

@@ -274,7 +274,6 @@ class GcpDependencyBuilder extends CloudDependenciesBuilder {
 
     val diskService = new DiskServiceInterp[IO](
       ConfigReader.appConfig.persistentDisk,
-      baselineDependencies.authProvider,
       baselineDependencies.publisherQueue,
       Some(gcpDependencies.googleDiskService),
       Some(gcpDependencies.googleProjectDAO),
@@ -284,7 +283,6 @@ class GcpDependencyBuilder extends CloudDependenciesBuilder {
     val runtimeService = RuntimeService(
       baselineDependencies.runtimeServicesConfig,
       ConfigReader.appConfig.persistentDisk,
-      baselineDependencies.authProvider,
       baselineDependencies.dockerDAO,
       Some(gcpDependencies.googleStorageService),
       Some(gcpDependencies.googleComputeService),